rw2sender: Convert encoding and sorting tests to new framework; bumped Prom version

.github/workflows/compliance-tests.yml

@@ -18,7 +18,4 @@ jobs:
       - uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8 # v6.0.1
         with:
           persist-credentials: false
-      - run: cd remotewrite && make sender PROMETHEUS_RW_COMPLIANCE_SKIP_TEST_RE=${SKIP_TESTS}
-        env:
-          # Skip ST tests on CI as Prometheus does not implement those yet (see PROM-60).
-          SKIP_TESTS: "TestCompliance/prometheus/samples/rw2/start_timestamp*"
+      - run: cd remotewrite && make sender

remotewrite/sender/compliance_test.go

@@ -0,0 +1,25 @@
+// Copyright The Prometheus Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package sender_test
+
+import (
+	"testing"
+
+	"github.com/prometheus/compliance/remotewrite/sender"
+)
+
+// TestCompliance runs all tests in this compliance suite against prometheus{}
+func TestCompliance(t *testing.T) {
+	sender.RunTests(t, prometheus{}, sender.ComplianceTests())
+}

remotewrite/sender/helpers.go

@@ -393,24 +393,51 @@ func runTestCases(t *testing.T, tests []TestCase) {
 	}
 }
 
+type timeseriesResult struct {
+	TimeSeries *writev2.TimeSeries
+	Labels     map[string]string
+}
+
+func flattenTimeseriesResult(t testing.TB, results []timeseriesResult) (ret timeseriesResult) {
+	require.NotEmpty(t, results)
+
+	var lbls map[string]string
+	for _, r := range results {
+		if lbls == nil {
+			lbls = r.Labels
+		} else {
+			require.Equal(t, lbls, r.Labels, "found two different series with the same name; expected same series")
+		}
+		if ret.TimeSeries == nil {
+			ret.TimeSeries = r.TimeSeries
+		} else {
+			ret.TimeSeries.Samples = append(ret.TimeSeries.Samples, r.TimeSeries.Samples...)
+			ret.TimeSeries.Histograms = append(ret.TimeSeries.Histograms, r.TimeSeries.Histograms...)
+			ret.TimeSeries.Exemplars = append(ret.TimeSeries.Exemplars, r.TimeSeries.Exemplars...)
+		}
+	}
+	return ret
+}
+
 // findTimeseriesByMetricName finds a timeseries by metric name from a captured request.
-func findTimeseriesByMetricName(req *writev2.Request, metricName string) (*writev2.TimeSeries, map[string]string) {
+func findTimeseriesByMetricName(req *writev2.Request, metricName string) []timeseriesResult {
+	var results []timeseriesResult
 	for i := range req.Timeseries {
 		ts := &req.Timeseries[i]
 		labels := extractLabels(ts, req.Symbols)
 		if labels["__name__"] == metricName {
-			return ts, labels
+			results = append(results, timeseriesResult{TimeSeries: ts, Labels: labels})
 		}
 	}
-	return nil, nil
+	return results
 }
 
 // requireTimeseriesByMetricName finds a timeseries by metric name and fails the test if not found.
-func requireTimeseriesByMetricName(t *testing.T, req *writev2.Request, metricName string) (*writev2.TimeSeries, map[string]string) {
+func requireTimeseriesByMetricName(t *testing.T, req *writev2.Request, metricName string) []timeseriesResult {
 	t.Helper()
-	ts, labels := findTimeseriesByMetricName(req, metricName)
-	require.NotNil(t, ts, "Timeseries with metric name %q must be present", metricName)
-	return ts, labels
+	results := findTimeseriesByMetricName(req, metricName)
+	require.NotEmpty(t, results, "Timeseries with metric name %q must be present", metricName)
+	return results
 }
 
 // requireTimeseriesRW1ByMetricName finds a timeseries by metric name and fails the test if not found.
@@ -455,15 +482,15 @@ func findHistogramData(req *writev2.Request, baseName string) (classicFound bool
 // Returns (count, found) where found indicates if count was successfully extracted.
 func extractHistogramCount(req *writev2.Request, baseName string) (float64, bool) {
 	// Try classic format first.
-	ts, _ := findTimeseriesByMetricName(req, baseName+"_count")
-	if ts != nil && len(ts.Samples) > 0 {
-		return ts.Samples[0].Value, true
+	results := findTimeseriesByMetricName(req, baseName+"_count")
+	if len(results) > 0 && len(results[0].TimeSeries.Samples) > 0 {
+		return results[0].TimeSeries.Samples[0].Value, true
 	}
 
 	// Try native format.
-	ts, _ = findTimeseriesByMetricName(req, baseName)
-	if ts != nil && len(ts.Histograms) > 0 {
-		hist := ts.Histograms[0]
+	results = findTimeseriesByMetricName(req, baseName)
+	if len(results) > 0 && len(results[0].TimeSeries.Histograms) > 0 {
+		hist := results[0].TimeSeries.Histograms[0]
 		if hist.Count != nil {
 			if countInt, ok := hist.Count.(*writev2.Histogram_CountInt); ok {
 				return float64(countInt.CountInt), true
@@ -479,15 +506,15 @@ func extractHistogramCount(req *writev2.Request, baseName string) (float64, bool
 // extractHistogramSum extracts sum from either classic or native histogram format.
 func extractHistogramSum(req *writev2.Request, baseName string) (float64, bool) {
 	// Try classic format first.
-	ts, _ := findTimeseriesByMetricName(req, baseName+"_sum")
-	if ts != nil && len(ts.Samples) > 0 {
-		return ts.Samples[0].Value, true
+	results := findTimeseriesByMetricName(req, baseName+"_sum")
+	if len(results) > 0 && len(results[0].TimeSeries.Samples) > 0 {
+		return results[0].TimeSeries.Samples[0].Value, true
 	}
 
 	// Try native format.
-	ts, _ = findTimeseriesByMetricName(req, baseName)
-	if ts != nil && len(ts.Histograms) > 0 {
-		return ts.Histograms[0].Sum, true
+	results = findTimeseriesByMetricName(req, baseName)
+	if len(results) > 0 && len(results[0].TimeSeries.Histograms) > 0 {
+		return results[0].TimeSeries.Histograms[0].Sum, true
 	}
 
 	return 0, false

remotewrite/sender/labels_test.go

@@ -19,6 +19,22 @@ import (
 	"testing"
 )
 
+/*
+TODO later
+{
+					Name:        "job_instance_labels_present",
+					Description: "Sender SHOULD include job and instance labels in samples",
+					RFCLevel:    ShouldLevel,
+					Validate: func(t *testing.T, res ReceiverResult) {
+						results := requireTimeseriesByMetricName(t, res.Requests[0].RW2, "test_float")
+						require.Len(t, results, 1, "Should receive exactly one timeseries for test_float")
+						labels := results[0].Labels
+						require.NotEmpty(t, labels["job"], "Sample should include 'job' label")
+						require.NotEmpty(t, labels["instance"], "Sample should include 'instance' label")
+					},
+				},
+*/
+
 // TestLabelValidation validates label encoding and formatting.
 func TestLabelValidation_Old(t *testing.T) {
 	t.Skip("TODO: Revise and move to a new framework")

remotewrite/sender/test_test.go → remotewrite/sender/prometheus_test.go

@@ -29,14 +29,13 @@ import (
 	"runtime"
 	"strings"
 	"sync"
-	"testing"
 	"text/template"
 
 	"github.com/prometheus/compliance/remotewrite/sender"
 )
 
 const (
-	prometheusDownloadURL = "https://github.com/prometheus/prometheus/releases/download/v3.9.1/prometheus-3.9.1.{{.OS}}-{{.Arch}}.tar.gz"
+	prometheusDownloadURL = "https://github.com/prometheus/prometheus/releases/download/v3.11.0-rc.0/prometheus-3.11.0-rc.0.{{.OS}}-{{.Arch}}.tar.gz"
 	scrapeConfigTemplate  = `
 global:
   scrape_interval: 1s
@@ -71,7 +70,8 @@ func (p prometheus) Name() string { return "prometheus" }
 // Run runs a Prometheus process for a test target options, until ctx is done.
 //
 // It auto-downloads Prometheus binary from the official release URL (see prometheusDownloadURL).
-// TODO(bwplotka): Process based runners are prone to leaking processes; add docker runner and/or figure out cleanup. Manually this could be done with 'killall -m "prometheus-3." -kill'.
+// TODO(bwplotka): Process based runners are prone to leaking processes; add docker runner and/or figure out cleanup.
+// Manually this could be done with 'killall -m "prometheus-3." -kill'.
 func (p prometheus) Run(ctx context.Context, opts sender.Options) error {
 	binary, err := downloadBinary(prometheusDownloadURL, "prometheus")
 	if err != nil {
@@ -97,6 +97,7 @@ func (p prometheus) Run(ctx context.Context, opts sender.Options) error {
 		`--web.listen-address=0.0.0.0:0`,
 		fmt.Sprintf("--storage.tsdb.path=%v", dir),
 		fmt.Sprintf("--config.file=%s", configFile),
+		"--enable-feature=st-storage",
 	)
 }
 
@@ -298,7 +299,3 @@ func extractTarGz(srcFile, filename, destFile string) error {
 
 	return fmt.Errorf("did not find binary in .tar.gz: %s", filename)
 }
-
-func TestCompliance(t *testing.T) {
-	sender.RunTests(t, prometheus{}, sender.ComplianceTests())
-}