tests(scrape): add TestScrapeLoopAppend_WithStorage (#17937)

Signed-off-by: bwplotka <bwplotka@gmail.com>

scrape/scrape_test.go

@@ -1436,7 +1436,9 @@ func readTextParseTestMetrics(t testing.TB) []byte {
 	if err != nil {
 		t.Fatal(err)
 	}
-	return b
+
+	// Replace all Carriage Return chars that appear when testing on windows.
+	return bytes.ReplaceAll(b, []byte{'\r'}, nil)
 }
 
 func makeTestGauges(n int) []byte {
@@ -1543,6 +1545,184 @@ func TestPromTextToProto(t *testing.T) {
 	require.Equal(t, "promhttp_metric_handler_requests_total", got[236])
 }
 
+// TestScrapeLoopAppend_WithStorage tests appends and storage integration for the
+// large input files that are also used in benchmarks.
+func TestScrapeLoopAppend_WithStorage(t *testing.T) {
+	ts := time.Now()
+
+	for _, appV2 := range []bool{false, true} {
+		for _, tc := range []struct {
+			name         string
+			parsableText []byte
+
+			expectedSamplesLen  int
+			testAppendedSamples func(t *testing.T, committed []sample)
+			testExemplars       func(t *testing.T, er []exemplar.QueryResult)
+		}{
+			{
+				name:         "1Fam2000Gauges",
+				parsableText: makeTestGauges(2000),
+
+				expectedSamplesLen: 2000,
+				testAppendedSamples: func(t *testing.T, committed []sample) {
+					var expectedMF string
+					if appV2 {
+						expectedMF = "metric_a" // Only AppenderV2 supports metric family passing.
+					}
+					// Verify a few samples.
+					testutil.RequireEqual(t, sample{
+						MF: expectedMF,
+						M:  metadata.Metadata{Type: model.MetricTypeGauge, Help: "help text"},
+						L:  labels.FromStrings(model.MetricNameLabel, "metric_a", "foo", "0", "bar", "0"), V: 1, T: timestamp.FromTime(ts),
+					}, committed[0])
+					testutil.RequireEqual(t, sample{
+						MF: expectedMF,
+						M:  metadata.Metadata{Type: model.MetricTypeGauge, Help: "help text"},
+						L:  labels.FromStrings(model.MetricNameLabel, "metric_a", "foo", "1245", "bar", "124500"), V: 1, T: timestamp.FromTime(ts),
+					}, committed[1245])
+					testutil.RequireEqual(t, sample{
+						MF: expectedMF,
+						M:  metadata.Metadata{Type: model.MetricTypeGauge, Help: "help text"},
+						L:  labels.FromStrings(model.MetricNameLabel, "metric_a", "foo", "1999", "bar", "199900"), V: 1, T: timestamp.FromTime(ts),
+					}, committed[len(committed)-1])
+				},
+			},
+			{
+				name:         "237FamsAllTypes",
+				parsableText: readTextParseTestMetrics(t),
+
+				expectedSamplesLen: 1857,
+				testAppendedSamples: func(t *testing.T, committed []sample) {
+					// Verify a few samples.
+					testutil.RequireEqual(t, sample{
+						MF: func() string {
+							if !appV2 {
+								return ""
+							}
+							return "go_gc_gomemlimit_bytes"
+						}(),
+						M: metadata.Metadata{Type: model.MetricTypeGauge, Help: "Go runtime memory limit configured by the user, otherwise math.MaxInt64. This value is set by the GOMEMLIMIT environment variable, and the runtime/debug.SetMemoryLimit function. Sourced from /gc/gomemlimit:bytes"},
+						L: labels.FromStrings(model.MetricNameLabel, "go_gc_gomemlimit_bytes"), V: 9.03676723e+08, T: timestamp.FromTime(ts),
+					}, committed[11])
+					testutil.RequireEqual(t, sample{
+						MF: func() string {
+							if !appV2 {
+								return "" // Only AppenderV2 supports metric family passing.
+							}
+							return "prometheus_http_request_duration_seconds"
+						}(),
+						M: metadata.Metadata{Type: model.MetricTypeHistogram, Help: "Histogram of latencies for HTTP requests."},
+						L: labels.FromStrings(model.MetricNameLabel, "prometheus_http_request_duration_seconds_bucket", "handler", "/api/v1/query_range", "le", "120.0"), V: 118157, T: timestamp.FromTime(ts),
+					}, committed[448])
+					testutil.RequireEqual(t, sample{
+						MF: func() string {
+							if !appV2 {
+								return "" // Only AppenderV2 supports metric family passing.
+							}
+							return "promhttp_metric_handler_requests_total"
+						}(),
+						M: metadata.Metadata{Type: model.MetricTypeCounter, Help: "Total number of scrapes by HTTP status code."},
+						L: labels.FromStrings(model.MetricNameLabel, "promhttp_metric_handler_requests_total", "code", "503"), V: 0, T: timestamp.FromTime(ts),
+					}, committed[len(committed)-1])
+				},
+			},
+			{
+				name:         "100HistsWithExemplars",
+				parsableText: makeTestHistogramsWithExemplars(100),
+
+				expectedSamplesLen: 24 * 100,
+				testAppendedSamples: func(t *testing.T, committed []sample) {
+					// Verify a few samples.
+					m := metadata.Metadata{Type: model.MetricTypeHistogram, Help: "RPC latency distributions."}
+					testutil.RequireEqual(t, sample{
+						MF: func() string {
+							if !appV2 {
+								return "" // Only AppenderV2 supports metric family passing.
+							}
+							return "rpc_durations_histogram0_seconds"
+						}(),
+						M: m, L: labels.FromStrings(model.MetricNameLabel, "rpc_durations_histogram0_seconds_bucket", "le", "0.0003100000000000002"), V: 15, T: timestamp.FromTime(ts),
+						ES: []exemplar.Exemplar{
+							{Labels: labels.FromStrings("dummyID", "9818"), Value: 0.0002791130914009552, Ts: 1726839814982, HasTs: true},
+						},
+					}, committed[13])
+					testutil.RequireEqual(t, sample{
+						MF: func() string {
+							if !appV2 {
+								return "" // Only AppenderV2 supports metric family passing.
+							}
+							return "rpc_durations_histogram49_seconds"
+						}(),
+						M: m, L: labels.FromStrings(model.MetricNameLabel, "rpc_durations_histogram49_seconds_sum"), V: -8.452185437166741e-05, T: timestamp.FromTime(ts),
+					}, committed[24*50-3])
+
+					// This series does not have metadata, nor metric family, because of isSeriesPartOfFamily bug and OpenMetric 1.0 limitations around _created series.
+					// TODO(bwplotka): Fix with https://github.com/prometheus/prometheus/issues/17900
+					testutil.RequireEqual(t, sample{
+						L: labels.FromStrings(model.MetricNameLabel, "rpc_durations_histogram99_seconds_created"), V: 1.726839813016302e+09, T: timestamp.FromTime(ts),
+					}, committed[len(committed)-1])
+				},
+				testExemplars: func(t *testing.T, er []exemplar.QueryResult) {
+					// 12 out of 24 histogram series have exemplars.
+					require.Len(t, er, 12*100)
+					testutil.RequireEqual(t, exemplar.QueryResult{
+						SeriesLabels: labels.FromStrings(model.MetricNameLabel, "rpc_durations_histogram0_seconds_bucket", "le", "0.0003100000000000002"),
+						Exemplars: []exemplar.Exemplar{
+							{Labels: labels.FromStrings("dummyID", "9818"), Value: 0.0002791130914009552, Ts: 1726839814982, HasTs: true},
+						},
+					}, er[10])
+					testutil.RequireEqual(t, exemplar.QueryResult{
+						SeriesLabels: labels.FromStrings(model.MetricNameLabel, "rpc_durations_histogram9_seconds_bucket", "le", "1.0000000000000216e-05"),
+						Exemplars: []exemplar.Exemplar{
+							{Labels: labels.FromStrings("dummyID", "19206"), Value: -4.6156147425468016e-05, Ts: 1726839815133, HasTs: true},
+						},
+					}, er[len(er)-1])
+				},
+			},
+		} {
+			t.Run(fmt.Sprintf("appV2=%v/data=%v", appV2, tc.name), func(t *testing.T) {
+				s := teststorage.New(t, func(opt *tsdb.Options) {
+					opt.EnableMetadataWALRecords = true
+				})
+
+				appTest := teststorage.NewAppendable().Then(s)
+				sl, _ := newTestScrapeLoop(t, withAppendable(appTest, appV2))
+				app := sl.appender()
+
+				_, _, _, err := app.append(tc.parsableText, "application/openmetrics-text", ts)
+				require.NoError(t, err)
+				require.NoError(t, app.Commit())
+
+				// Check the recorded samples on the Appender layer.
+				require.Nil(t, appTest.PendingSamples())
+				require.Nil(t, appTest.RolledbackSamples())
+
+				got := appTest.ResultSamples()
+				require.Len(t, got, tc.expectedSamplesLen)
+				tc.testAppendedSamples(t, got)
+
+				// Check basic storage stats.
+				stats := s.Head().Stats(model.MetricNameLabel, 2000)
+				require.Equal(t, tc.expectedSamplesLen, int(stats.NumSeries))
+
+				// Check exemplars.
+				eq, err := s.ExemplarQuerier(t.Context())
+				require.NoError(t, err)
+
+				er, err := eq.Select(math.MinInt64, math.MaxInt64, nil)
+				require.NoError(t, err)
+
+				if tc.testExemplars != nil {
+					tc.testExemplars(t, er)
+				} else {
+					// Expect no exemplars.
+					require.Empty(t, er, "%v is not empty", er)
+				}
+			})
+		}
+	}
+}
+
 // BenchmarkScrapeLoopAppend benchmarks scrape appends for typical cases.
 //
 // Benchmark compares append function run across 4 dimensions:
@@ -1567,7 +1747,7 @@ func BenchmarkScrapeLoopAppend(b *testing.B) {
 				name         string
 				parsableText []byte
 			}{
-				{name: "1Fam1000Gauges", parsableText: makeTestGauges(2000)},         // ~68.1 KB, ~77.9 KB in proto.
+				{name: "1Fam2000Gauges", parsableText: makeTestGauges(2000)},         // ~68.1 KB, ~77.9 KB in proto.
 				{name: "237FamsAllTypes", parsableText: readTextParseTestMetrics(b)}, // ~185.7 KB, ~70.6 KB in proto.
 			} {
 				b.Run(fmt.Sprintf("appV2=%v/appendMetadataToWAL=%v/data=%v", appV2, appendMetadataToWAL, data.name), func(b *testing.B) {
@@ -3218,9 +3398,7 @@ metric: <
 				}
 				sl.alwaysScrapeClassicHist = test.alwaysScrapeClassicHist
 				// This test does not care about metadata.
-				// Having this true would mean we need to add metadata to sample
-				// expectations.
-				// TODO(bwplotka): Add cases for append metadata to WAL and pass metadata
+				// TODO(bwplotka): Add metadata expectations and turn it on.
 				sl.appendMetadataToWAL = false
 			})
 			app := sl.appender()

tsdb/head_append_v2.go

@@ -200,9 +200,6 @@ func (a *headAppenderV2) Append(ref storage.SeriesRef, ls labels.Labels, st, t i
 		// Currently only exemplars can return partial errors.
 		partialErr = a.appendExemplars(s, opts.Exemplars)
 	}
-
-	// TODO(bwplotka): Move/reuse metadata tests from scrape, once scrape adopts AppenderV2.
-	// Currently tsdb package does not test metadata.
 	if a.head.opts.EnableMetadataWALRecords && !opts.Metadata.IsEmpty() {
 		s.Lock()
 		metaChanged := s.meta == nil || !s.meta.Equals(opts.Metadata)