From 6ba44863a01e75a96ddea8b6398b341a831060b0 Mon Sep 17 00:00:00 2001 From: om7057 Date: Tue, 11 Aug 2026 22:39:47 +0530 Subject: [PATCH] remotewrite/receiver: propose programmatic Receiver harness (RFC #208) Ports the sender suite's programmatic pattern (Sender/RunTests/ComplianceTests, see remotewrite/sender) to the receiver side, so receiver compliance tests can be imported and run against a target Receiver implementation instead of being driven by an external config file + env vars. This converts metric_test.go's cases as a proof of concept (84 generated sub-cases) and adds a self-test that runs them against a downloaded Prometheus release binary. It intentionally lives at remotewrite/receiver/next rather than replacing remotewrite/receiver outright, to avoid breaking the existing suite while the design gets reviewed; if accepted, the remaining 9 files (combined, exemplar, histograms, metadata, request, rw1_compat) would be converted the same way and this would move up to replace remotewrite/receiver. One notable design deviation from sender: RunTests starts the receiver once per whole run rather than once per test case, since real receivers are too costly to restart per case and these tests only assert on the synchronous HTTP response to each write. Also note: remotewrite/receiver currently fails `go vet` on this branch (ts.CreatedTimestamp undefined) because the pinned prometheus/prometheus version predates writev2 start-timestamp support; the pending dependabot bump to v0.312.0 is a prerequisite for CreatedTimestamp-based cases. Signed-off-by: om7057 --- remotewrite/receiver/next/.gitignore | 4 + remotewrite/receiver/next/compliance.go | 229 +++++++++++ remotewrite/receiver/next/metric.go | 159 ++++++++ remotewrite/receiver/next/prometheus_test.go | 317 +++++++++++++++ remotewrite/receiver/next/receiver.go | 91 +++++ remotewrite/receiver/next/request.go | 406 +++++++++++++++++++ 6 files changed, 1206 insertions(+) create mode 100644 remotewrite/receiver/next/.gitignore create mode 100644 remotewrite/receiver/next/compliance.go create mode 100644 remotewrite/receiver/next/metric.go create mode 100644 remotewrite/receiver/next/prometheus_test.go create mode 100644 remotewrite/receiver/next/receiver.go create mode 100644 remotewrite/receiver/next/request.go diff --git a/remotewrite/receiver/next/.gitignore b/remotewrite/receiver/next/.gitignore new file mode 100644 index 0000000..97bf923 --- /dev/null +++ b/remotewrite/receiver/next/.gitignore @@ -0,0 +1,4 @@ +bin/ + +*.test +*.out diff --git a/remotewrite/receiver/next/compliance.go b/remotewrite/receiver/next/compliance.go new file mode 100644 index 0000000..30e4fb8 --- /dev/null +++ b/remotewrite/receiver/next/compliance.go @@ -0,0 +1,229 @@ +// 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 receiver + +import ( + "context" + "fmt" + "net/http" + "net/url" + "os" + "strconv" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// defaultReadyTimeout is generous by default because a Receiver implementation +// may need to download or build a binary before it can start (see e.g. the +// release-binary-based prometheus target). Override with +// PROMETHEUS_RW2_COMPLIANCE_READY_TIMEOUT (a value parseable by time.ParseDuration), +// mirroring the sender suite's PROMETHEUS_RW2_COMPLIANCE_TEST_TIMEOUT. +const defaultReadyTimeout = 3 * time.Minute + +func readyTimeout() time.Duration { + if v := os.Getenv("PROMETHEUS_RW2_COMPLIANCE_READY_TIMEOUT"); v != "" { + if d, err := time.ParseDuration(v); err == nil { + return d + } + } + return defaultReadyTimeout +} + +// ComplianceTests returns the official Remote Write receiver compliance tests. +func ComplianceTests() (ret []Test) { + ret = append(ret, metricTests()...) + return ret +} + +type RFCLevel string + +const ( + MustLevel RFCLevel = "MUST" + ShouldLevel RFCLevel = "SHOULD" + MayLevel RFCLevel = "MAY" +) + +func (r RFCLevel) annotate(t *testing.T) { + t.Attr("rfcLevel", string(r)) +} + +// ExpectedResponse describes the expected outcome of sending a request to the receiver. +type ExpectedResponse struct { + // Samples, Exemplars, Histograms are the expected counts reported via the + // X-Prometheus-Remote-Write-*-Written response headers. + Samples int + Exemplars int + Histograms int + // ExactStatusCode, if non-zero, is asserted exactly. Otherwise, any 2xx implies + // success and any 4xx defaults to expecting http.StatusBadRequest. + ExactStatusCode int +} + +// Test defines a single Remote Write receiver compliance test case. +// +// Each Test produces a MUST sub-test that asserts basic (non-strict) compliance, +// plus, when ExpectSuccess is true, an additional SHOULD sub-test that asserts the +// receiver responds with exactly http.StatusNoContent. +type Test struct { + // Name is a unique name for the test case; adds a "//" sub-test. + Name string + // Description describes what the test case is verifying. + Description string + // Opts is the request to send to the receiver under test. + Opts RequestOpts + // Expect describes the expected outcome. + Expect ExpectedResponse + // ExpectSuccess indicates whether the request as a whole is expected to succeed (2xx). + ExpectSuccess bool +} + +// RunTests starts target and runs each compliance test case against it. +// +// Unlike the sender suite (which restarts a fresh sender per test case), RunTests +// starts the receiver under test once for the whole run: real receivers (e.g. a full +// Prometheus binary) are too costly to restart per case, and these tests only assert +// on the synchronous HTTP response to each write, so a shared, long-lived receiver +// instance is sufficient and does not leak state between assertions. +func RunTests(t *testing.T, target Receiver, tcs []Test) { + t.Helper() + + require.NotNil(t, target) + require.NotEmpty(t, tcs) + + ctx, cancel := context.WithCancel(t.Context()) + t.Cleanup(cancel) + + readyCh := make(chan string, 1) + runErrCh := make(chan error, 1) + go func() { + runErrCh <- target.Run(ctx, func(url string) { + select { + case readyCh <- url: + default: + } + }) + }() + + var baseURL string + select { + case baseURL = <-readyCh: + case err := <-runErrCh: + t.Fatalf("receiver %q stopped before becoming ready: %v", target.Name(), err) + case <-time.After(readyTimeout()): + cancel() + t.Fatalf("receiver %q did not become ready in time", target.Name()) + } + + client := &http.Client{Timeout: 10 * time.Second} + + for _, tc := range tcs { + runComplianceTest(t, client, baseURL, target.Name(), tc) + } +} + +// runComplianceTest runs tc against the already-running receiver at baseURL, producing +// both a SHOULD (strict, 204-only) sub-test for expected-success cases and a MUST +// (basic compliance) sub-test for all cases, mirroring the pre-conversion behaviour. +func runComplianceTest(t *testing.T, client *http.Client, baseURL, targetName string, tc Test) { + t.Helper() + + if tc.ExpectSuccess { + t.Run(fmt.Sprintf("%s/%s returns 204", targetName, tc.Name), func(t *testing.T) { + ShouldLevel.annotate(t) + t.Attr("description", tc.Description) + + expect := tc.Expect + expect.ExactStatusCode = http.StatusNoContent + doAndValidate(t, client, baseURL, tc.Opts, expect, true) + }) + } + + t.Run(fmt.Sprintf("%s/%s", targetName, tc.Name), func(t *testing.T) { + MustLevel.annotate(t) + t.Attr("description", tc.Description) + + doAndValidate(t, client, baseURL, tc.Opts, tc.Expect, tc.ExpectSuccess) + }) +} + +func doAndValidate(t *testing.T, client *http.Client, baseURL string, opts RequestOpts, expect ExpectedResponse, expectSuccess bool) { + t.Helper() + + req := generateRequest(opts) + req.URL = mustParseURL(t, baseURL) + + resp, err := client.Do(req) + require.NoError(t, err, "request to receiver failed") + defer resp.Body.Close() + + validateResponse(t, expect, expectSuccess, resp) +} + +func getHeaderValue(t *testing.T, header http.Header, key string) int { + t.Helper() + v := header.Get("X-Prometheus-Remote-Write-" + key + "-Written") + if v == "" { + // Receivers CAN assume that any missing X-Prometheus-Remote-Write-*-Written + // response header means no element from this category was written (count of 0). + return 0 + } + i, err := strconv.Atoi(v) + require.NoError(t, err) + return i +} + +// validateResponse asserts resp matches expect/expectSuccess, mirroring the RFC-defined +// status code and X-Prometheus-Remote-Write-*-Written header semantics. +func validateResponse(t *testing.T, expect ExpectedResponse, expectSuccess bool, resp *http.Response) { + t.Helper() + + samplesWritten := getHeaderValue(t, resp.Header, "Samples") + exemplarsWritten := getHeaderValue(t, resp.Header, "Exemplars") + histogramsWritten := getHeaderValue(t, resp.Header, "Histograms") + + if expect.ExactStatusCode != 0 { + require.Equal(t, expect.ExactStatusCode, resp.StatusCode, "response code should be exactly %d", expect.ExactStatusCode) + } + + switch resp.StatusCode / 100 { + case 2: + require.True(t, expectSuccess, "response code is %d but success is false", resp.StatusCode) + require.Equal(t, expect.Samples, samplesWritten, "%d samples written", samplesWritten) + require.Equal(t, expect.Exemplars, exemplarsWritten, "%d exemplars written", exemplarsWritten) + require.Equal(t, expect.Histograms, histogramsWritten, "%d histograms written", histogramsWritten) + case 4: + if expect.ExactStatusCode == 0 { + require.Equal(t, http.StatusBadRequest, resp.StatusCode, "response code should be exactly %d", http.StatusBadRequest) + } + require.GreaterOrEqual(t, expect.Samples, samplesWritten, "%d samples written", samplesWritten) + require.GreaterOrEqual(t, expect.Exemplars, exemplarsWritten, "%d exemplars written", exemplarsWritten) + require.GreaterOrEqual(t, expect.Histograms, histogramsWritten, "%d histograms written", histogramsWritten) + case 5: + require.False(t, expectSuccess, "response code is %d but success is true", resp.StatusCode) + require.GreaterOrEqual(t, expect.Samples, samplesWritten, "%d samples written", samplesWritten) + require.GreaterOrEqual(t, expect.Exemplars, exemplarsWritten, "%d exemplars written", exemplarsWritten) + require.GreaterOrEqual(t, expect.Histograms, histogramsWritten, "%d histograms written", histogramsWritten) + default: + require.Fail(t, fmt.Sprintf("response code is %d but should be 2xx, 4xx, or 5xx", resp.StatusCode)) + } +} + +func mustParseURL(t *testing.T, raw string) *url.URL { + t.Helper() + u, err := url.Parse(raw) + require.NoError(t, err) + return u +} diff --git a/remotewrite/receiver/next/metric.go b/remotewrite/receiver/next/metric.go new file mode 100644 index 0000000..6cb81fe --- /dev/null +++ b/remotewrite/receiver/next/metric.go @@ -0,0 +1,159 @@ +// 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 receiver + +import ( + "fmt" + "math" + "time" + + "github.com/prometheus/prometheus/model/value" +) + +// specialFloatValues contains test float values including special values +// (NaN, Inf) shared across metric test cases. +var specialFloatValues = map[string]float64{ + "1.0": 1.0, + "StaleNaN": float64(value.StaleNaN), + "NaN": float64(value.NormalNaN), + "Inf": math.Inf(1), + "-Inf": math.Inf(-1), +} + +func basicMetric(name string) map[string]string { + return map[string]string{"__name__": name} +} + +func testJobInstanceLabels() map[string]string { + return map[string]string{"__name__": "up", "job": "testjob", "instance": "localhost:9090"} +} + +// metricTests returns compliance tests covering single and multiple metric +// samples with a variety of float values (including NaN/Inf) and label shapes. +func metricTests() (ret []Test) { + for name, v := range specialFloatValues { + ret = append(ret, singleMetricTests(name, v)...) + ret = append(ret, multipleMetricsTests(name, v)...) + } + ret = append(ret, counterWithCreatedTimestampTest()) + return ret +} + +func singleMetricTests(valueName string, v float64) []Test { + cases := []struct { + name string + labels map[string]string + success bool + }{ + {"simple metric", basicMetric("up"), true}, + {"simple metric with multiple labels", testJobInstanceLabels(), true}, + {"simple metric with newlines", map[string]string{"__name__": "up", "job": "test\njob\n", "instance": "localhost:9090"}, true}, + {"simple metric with dots", map[string]string{"__name__": "resource.cpu.usage", "job.name": "testjob", "instance.name": "localhost:9090"}, true}, + {"simple metric with spaces", map[string]string{"__name__": "resource cpu usage", "job name": "testjob", "instance name": "localhost:9090"}, true}, + {"simple metric without name label", map[string]string{"job": "testjob", "instance": "localhost:9090"}, false}, + {"empty metric", map[string]string{}, false}, + } + + var ret []Test + for _, tc := range cases { + expectedSamples := 0 + if tc.success { + expectedSamples = 1 + } + ret = append(ret, Test{ + Name: fmt.Sprintf("SingleMetric/%s/%s", valueName, tc.name), + Description: "Test single metric samples with different float values including special values (NaN, Inf)", + Opts: RequestOpts{ + Samples: []SampleWithLabels{{Labels: tc.labels, Value: v}}, + }, + Expect: ExpectedResponse{Samples: expectedSamples}, + ExpectSuccess: tc.success, + }) + } + return ret +} + +func multipleMetricsTests(valueName string, v float64) []Test { + cases := []struct { + name string + metrics []SampleWithLabels + success bool + validSamples int + }{ + { + name: "multiple metrics", + metrics: []SampleWithLabels{ + {Labels: testJobInstanceLabels(), Value: v}, + {Labels: map[string]string{"__name__": "up", "job": "testjob", "instance": "localhost:9091"}, Value: v}, + {Labels: map[string]string{"__name__": "up", "job": "testjob", "instance": "localhost:9092"}, Value: v}, + {Labels: map[string]string{"__name__": "up", "job": "testjob", "instance": "localhost:9093"}, Value: v}, + }, + success: true, + validSamples: 4, + }, + { + name: "multiple metrics, without name label", + metrics: []SampleWithLabels{ + {Labels: map[string]string{"job": "testjob", "instance": "localhost:9090"}, Value: v}, + {Labels: map[string]string{"job": "testjob", "instance": "localhost:9091"}, Value: v}, + {Labels: map[string]string{"job": "testjob", "instance": "localhost:9092"}, Value: v}, + {Labels: map[string]string{"job": "testjob", "instance": "localhost:9093"}, Value: v}, + }, + success: false, + validSamples: 0, + }, + { + name: "multiple metrics, 1 without name label", + metrics: []SampleWithLabels{ + {Labels: map[string]string{"job": "testjob", "instance": "localhost:9090"}, Value: v}, + {Labels: map[string]string{"__name__": "up", "job": "testjob", "instance": "localhost:9091"}, Value: v}, + {Labels: map[string]string{"__name__": "up", "job": "testjob", "instance": "localhost:9092"}, Value: v}, + {Labels: map[string]string{"__name__": "up", "job": "testjob", "instance": "localhost:9093"}, Value: v}, + }, + success: false, + validSamples: 3, + }, + } + + var ret []Test + for _, tc := range cases { + ret = append(ret, Test{ + Name: fmt.Sprintf("MultipleMetrics/%s/%s", valueName, tc.name), + Description: "Test multiple metric samples in single request with validation of partial success scenarios", + Opts: RequestOpts{Samples: tc.metrics}, + Expect: ExpectedResponse{Samples: tc.validSamples}, + ExpectSuccess: tc.success, + }) + } + return ret +} + +func counterWithCreatedTimestampTest() Test { + now := time.Now() + createdTime := now.Add(-1 * time.Hour) + + return Test{ + Name: "CounterWithCreatedTimestamp", + Description: "Test counter with created timestamp set in the past", + Opts: RequestOpts{ + Samples: []SampleWithLabels{{ + Labels: map[string]string{"__name__": "http_requests_total", "job": "api"}, + Value: 100.0, + CreatedTimestamp: &createdTime, + }}, + }, + Expect: ExpectedResponse{Samples: 1}, + ExpectSuccess: true, + } +} diff --git a/remotewrite/receiver/next/prometheus_test.go b/remotewrite/receiver/next/prometheus_test.go new file mode 100644 index 0000000..1a1c5bc --- /dev/null +++ b/remotewrite/receiver/next/prometheus_test.go @@ -0,0 +1,317 @@ +// 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 receiver_test + +import ( + "archive/tar" + "archive/zip" + "compress/gzip" + "context" + "fmt" + "io" + "net" + "net/http" + "net/url" + "os" + "path" + "path/filepath" + "runtime" + "strings" + "sync" + "testing" + "text/template" + "time" + + "github.com/prometheus/compliance/remotewrite/receiver/next" +) + +const prometheusDownloadURL = "https://github.com/prometheus/prometheus/releases/download/v3.11.0-rc.0/prometheus-3.11.0-rc.0.{{.OS}}-{{.Arch}}.tar.gz" + +type prometheus struct{} + +func (prometheus) Name() string { return "prometheus" } + +// Run downloads (and caches) a Prometheus release binary and runs it as the +// receiver under test, with remote write receiving enabled, until ctx is done. +func (prometheus) Run(ctx context.Context, ready func(remoteWriteURL string)) error { + binary, err := downloadBinary(prometheusDownloadURL, "prometheus") + if err != nil { + return err + } + + dir, err := os.MkdirTemp("", "receiver-test-*") + if err != nil { + return err + } + defer os.RemoveAll(dir) + + configFile := filepath.Join(dir, "prometheus.yml") + if err := os.WriteFile(configFile, []byte("global:\n scrape_interval: 15s\n"), 0o600); err != nil { + return err + } + + port, err := freePort() + if err != nil { + return err + } + addr := fmt.Sprintf("127.0.0.1:%d", port) + + done := make(chan error, 1) + go func() { + done <- receiver.RunCommand(ctx, dir, nil, binary, + fmt.Sprintf("--web.listen-address=%s", addr), + fmt.Sprintf("--storage.tsdb.path=%s", dir), + fmt.Sprintf("--config.file=%s", configFile), + "--web.enable-remote-write-receiver", + ) + }() + + readyURL := fmt.Sprintf("http://%s/-/ready", addr) + deadline := time.Now().Add(60 * time.Second) + for time.Now().Before(deadline) { + select { + case err := <-done: + if err != nil { + return fmt.Errorf("prometheus exited before becoming ready: %w", err) + } + return nil + default: + } + + resp, err := http.Get(readyURL) + if err == nil { + resp.Body.Close() + if resp.StatusCode == http.StatusOK { + ready(fmt.Sprintf("http://%s/api/v1/write", addr)) + return <-done + } + } + time.Sleep(300 * time.Millisecond) + } + return fmt.Errorf("prometheus did not become ready in time") +} + +func freePort() (int, error) { + l, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + return 0, err + } + defer l.Close() + return l.Addr().(*net.TCPAddr).Port, nil +} + +var _ receiver.Receiver = prometheus{} + +// TestReceiverCompliance runs the receiver compliance test suite against a +// downloaded Prometheus release binary. +func TestReceiverCompliance(t *testing.T) { + receiver.RunTests(t, prometheus{}, receiver.ComplianceTests()) +} + +// The functions below are copied as-is from ../../sender/prometheus_test.go to +// avoid introducing a shared internal package for this draft; if the design is +// accepted, this download helper should be deduplicated between sender and +// receiver (e.g. into an internal/targets helper package). + +var downloadMtx sync.Mutex + +func downloadBinary(urlPattern string, filenameInArchivePattern string) (string, error) { + downloadMtx.Lock() + defer downloadMtx.Unlock() + return downloadBinaryUnlocked(urlPattern, filenameInArchivePattern) +} + +func downloadBinaryUnlocked(urlPattern string, filenameInArchivePattern string) (string, error) { + urlToDownload, err := instantiateTemplate(urlPattern) + if err != nil { + return "", nil + } + + filenameInArchive, err := instantiateTemplate(filenameInArchivePattern) + if err != nil { + return "", nil + } + + parsedURL, err := url.Parse(urlToDownload) + if err != nil { + return "", nil + } + + cwd, err := os.Getwd() + if err != nil { + return "", err + } + + filename := path.Join(cwd, "bin", path.Base(parsedURL.Path)) + decompressTgz := strings.HasSuffix(filename, ".tar.gz") + if decompressTgz { + filename = strings.TrimSuffix(filename, ".tar.gz") + } + + decompressZip := strings.HasSuffix(filename, ".zip") + if decompressZip { + filename = strings.TrimSuffix(filename, ".zip") + } + + if _, err := os.Stat(filename); !os.IsNotExist(err) { + return filename, nil + } + + tempfile, err := downloadURL(urlToDownload) + if err != nil { + return "", nil + } + + if err := os.Mkdir(path.Dir(filename), 0o755); err != nil && !os.IsExist(err) { + return "", nil + } + + if decompressTgz { + if err := extractTarGz(tempfile, filenameInArchive, filename); err != nil { + return "", err + } + } else if decompressZip { + if err := extractZip(tempfile, filenameInArchive, filename); err != nil { + return "", err + } + } else { + if err := os.Rename(tempfile, filename); err != nil { + return "", err + } + } + + if err := os.Chmod(filename, 0o744); err != nil { + return "", err + } + + return filename, nil +} + +func instantiateTemplate(pattern string) (string, error) { + t := template.Must(template.New("url").Parse(pattern)) + var buf strings.Builder + err := t.Execute(&buf, map[string]interface{}{ + "OS": runtime.GOOS, + "Arch": runtime.GOARCH, + }) + return buf.String(), err +} + +func downloadURL(rawURL string) (filename string, err error) { + fmt.Println("Downloading", rawURL) + + tempfile, err := os.CreateTemp("", "") + if err != nil { + return "", err + } + defer tempfile.Close() + + resp, err := http.Get(rawURL) + if err != nil { + return "", err + } + defer resp.Body.Close() + + if resp.StatusCode != 200 { + return "", fmt.Errorf("error downloading: %d", resp.StatusCode) + } + + if _, err := io.Copy(tempfile, resp.Body); err != nil { + return "", err + } + + return tempfile.Name(), nil +} + +func extractZip(srcFile, filename, destFile string) error { + fmt.Println("Decompressing", srcFile) + + r, err := zip.OpenReader(srcFile) + if err != nil { + return err + } + defer r.Close() + + for _, f := range r.File { + if path.Base(f.Name) != filename { + continue + } + + src, err := f.Open() + if err != nil { + return err + } + defer src.Close() + + dest, err := os.Create(destFile) + if err != nil { + return err + } + defer dest.Close() + + _, err = io.Copy(dest, src) + return err + } + + return fmt.Errorf("did not find binary in .zip: %s", filename) +} + +func extractTarGz(srcFile, filename, destFile string) error { + fmt.Println("Decompressing", srcFile) + + f, err := os.Open(srcFile) + if err != nil { + return err + } + defer f.Close() + + gzf, err := gzip.NewReader(f) + if err != nil { + return err + } + defer gzf.Close() + + tarReader := tar.NewReader(gzf) + for { + header, err := tarReader.Next() + if err == io.EOF { + break + } + if err != nil { + return err + } + + if header.Typeflag != tar.TypeReg { + continue + } + + if path.Base(header.Name) != filename { + continue + } + + dest, err := os.Create(destFile) + if err != nil { + return err + } + defer dest.Close() + + if _, err := io.Copy(dest, tarReader); err != nil { + return err + } + + return nil + } + + return fmt.Errorf("did not find binary in .tar.gz: %s", filename) +} diff --git a/remotewrite/receiver/next/receiver.go b/remotewrite/receiver/next/receiver.go new file mode 100644 index 0000000..b9c36b4 --- /dev/null +++ b/remotewrite/receiver/next/receiver.go @@ -0,0 +1,91 @@ +// 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 receiver provides a programmatic Remote Write receiver compliance +// test suite, so that it can be imported and run against a target Receiver +// implementation (e.g. an in-process or subprocess Prometheus build), mirroring +// the pattern used by the sibling package "sender". +// +// NOTE: this package currently lives at remotewrite/receiver/next as a staging +// location so it doesn't collide with the existing remotewrite/receiver +// (package main) suite while under review (see #). Once the conversion of +// the remaining test files is complete and reviewed, this should move up to +// remotewrite/receiver, replacing the old package main suite. +package receiver + +import ( + "context" + "fmt" + "io" + "os" + "os/exec" + "syscall" +) + +// Receiver represents a Remote Write receiver under test. +// It generally needs to accept Prometheus Remote Write (RW1 and/or RW2) +// requests on an HTTP endpoint. +type Receiver interface { + // Name returns a unique receiver name. + Name() string + // Run starts the receiver under test until ctx is done. Once the receiver + // is ready to accept remote write requests, Run must invoke ready exactly + // once with the base URL of its remote-write endpoint. + // Premature stops (before ctx is done) are assumed to be failures. + Run(ctx context.Context, ready func(remoteWriteURL string)) error +} + +// RunCommand runs the given command with the given args until context is done. +// +// This is useful when starting process-based receiver targets. +func RunCommand(ctx context.Context, dir string, extraEnvVars []string, prog string, args ...string) error { + output := io.Discard + // Suppress output to avoid cluttering test results. + if os.Getenv("DEBUG") != "" { + output = os.Stdout + } + + cmd := exec.Command(prog, args...) + // Required for group process signalling on close. + cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} + cmd.Dir = dir + cmd.Env = os.Environ() + cmd.Env = append(cmd.Env, extraEnvVars...) + cmd.Stdout = output + cmd.Stderr = output + if err := cmd.Start(); err != nil { + return err + } + + cmdStopped := make(chan error) + defer close(cmdStopped) + go func() { + cmdStopped <- cmd.Wait() + }() + + select { + case <-ctx.Done(): + // Use group process ID. This allows using actually passing signal correctly + // to all processes when the parent does not support it (e.g. go run). + pgid, err := syscall.Getpgid(cmd.Process.Pid) + if err != nil { + return fmt.Errorf("failed to get pgid: %w", err) + } + if err := syscall.Kill(-pgid, syscall.SIGTERM); err != nil { + return fmt.Errorf("failed to send signal: %w", err) + } + return <-cmdStopped + case err := <-cmdStopped: + return err + } +} diff --git a/remotewrite/receiver/next/request.go b/remotewrite/receiver/next/request.go new file mode 100644 index 0000000..d5b73d7 --- /dev/null +++ b/remotewrite/receiver/next/request.go @@ -0,0 +1,406 @@ +// 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 receiver + +import ( + "bytes" + "io" + "net/http" + "time" + + "github.com/golang/snappy" + "github.com/prometheus/common/model" + writev2 "github.com/prometheus/prometheus/prompb/io/prometheus/write/v2" +) + +// SampleWithLabels describes a single float sample to send, along with its labels. +type SampleWithLabels struct { + Labels map[string]string + Value float64 + Offset time.Duration + CreatedTimestamp *time.Time +} + +// HistogramWithLabels describes a single native histogram to send, along with its labels. +type HistogramWithLabels struct { + Labels map[string]string + Histogram writev2.Histogram + Offset time.Duration + CreatedTimestamp *time.Time +} + +// ExemplarWithLabels describes a single exemplar to send, along with its metric and exemplar labels. +type ExemplarWithLabels struct { + Labels map[string]string + ExemplarLabels map[string]string + Value float64 + Offset time.Duration +} + +// MetadataWithLabels describes metric metadata to send, along with the labels of the metric it applies to. +type MetadataWithLabels struct { + Labels map[string]string + Type writev2.Metadata_MetricType + Help string + Unit string +} + +// RequestOpts contains all data required for generating a Remote Write v2 request. +type RequestOpts struct { + Samples []SampleWithLabels + Exemplars []ExemplarWithLabels + Metadata []MetadataWithLabels + Histograms []HistogramWithLabels + UnsafeRequest bool +} + +func labelsMatch(metric model.Metric, labels map[string]string) bool { + if len(metric) != len(labels) { + return false + } + for k, v := range labels { + if metricVal, exists := metric[model.LabelName(k)]; !exists || string(metricVal) != v { + return false + } + } + return true +} + +func mapToMetric(labels map[string]string) model.Metric { + metric := make(model.Metric) + for k, v := range labels { + metric[model.LabelName(k)] = model.LabelValue(v) + } + return metric +} + +// generateRequest generates a snappy-compressed Remote Write v2 HTTP request from opts. +func generateRequest(opts RequestOpts) *http.Request { + now := time.Now() + + if !opts.UnsafeRequest { + if len(opts.Samples) > 0 && len(opts.Histograms) > 0 { + panic("cannot have both Samples and Histograms in the same request") + } + for _, exemplar := range opts.Exemplars { + found := false + for _, sample := range opts.Samples { + if labelsMatch(mapToMetric(sample.Labels), exemplar.Labels) { + found = true + break + } + } + if !found { + for _, histogram := range opts.Histograms { + if labelsMatch(mapToMetric(histogram.Labels), exemplar.Labels) { + found = true + break + } + } + } + if !found { + panic("exemplar has no matching sample or histogram with same label set") + } + } + for _, metadata := range opts.Metadata { + metadataName := metadata.Labels["__name__"] + found := false + for _, sample := range opts.Samples { + if sample.Labels["__name__"] == metadataName { + found = true + break + } + } + if !found { + for _, histogram := range opts.Histograms { + if histogram.Labels["__name__"] == metadataName { + found = true + break + } + } + } + if !found { + panic("metadata has no matching sample or histogram with same metric name") + } + } + } + + symbols := writev2.NewSymbolTable() + var timeseries []writev2.TimeSeries + + metadataByName := make(map[string]writev2.Metadata) + for _, mw := range opts.Metadata { + metricName := mw.Labels["__name__"] + if metricName != "" || opts.UnsafeRequest { + metadataByName[metricName] = writev2.Metadata{ + Type: mw.Type, + HelpRef: symbols.Symbolize(mw.Help), + UnitRef: symbols.Symbolize(mw.Unit), + } + } + } + + seriesLastSample := make(map[string]int) + for i, s := range opts.Samples { + key := mapToMetric(s.Labels).String() + seriesLastSample[key] = i + } + + usedExemplars := make(map[int]bool) + + for i, s := range opts.Samples { + var labelRefs []uint32 + for k, v := range s.Labels { + labelRefs = append(labelRefs, symbols.Symbolize(k), symbols.Symbolize(v)) + } + + var sampleExemplars []writev2.Exemplar + seriesKey := mapToMetric(s.Labels).String() + isLastSampleForSeries := seriesLastSample[seriesKey] == i + + for ei, ew := range opts.Exemplars { + if usedExemplars[ei] { + continue + } + if labelsMatch(mapToMetric(s.Labels), ew.Labels) { + var exemplarLabelRefs []uint32 + for k, v := range ew.ExemplarLabels { + exemplarLabelRefs = append(exemplarLabelRefs, symbols.Symbolize(k), symbols.Symbolize(v)) + } + sampleExemplars = append(sampleExemplars, writev2.Exemplar{ + LabelsRefs: exemplarLabelRefs, + Value: ew.Value, + Timestamp: now.Add(ew.Offset).UnixMilli(), + }) + usedExemplars[ei] = true + if !isLastSampleForSeries { + break + } + } + } + + ts := writev2.TimeSeries{ + LabelsRefs: labelRefs, + Samples: []writev2.Sample{ + {Timestamp: now.Add(s.Offset).UnixMilli(), Value: s.Value}, + }, + Exemplars: sampleExemplars, + } + // NOTE: CreatedTimestamp (start-timestamp) is not wired into the request yet: + // the pinned github.com/prometheus/prometheus version predates writev2.TimeSeries + // start-timestamp support. A dependency bump (already tracked via dependabot) is + // a prerequisite for CounterWithCreatedTimestamp-style tests to be meaningful. + if metricName := s.Labels["__name__"]; metricName != "" || opts.UnsafeRequest { + if metadata, found := metadataByName[metricName]; found { + ts.Metadata = metadata + } + } + timeseries = append(timeseries, ts) + } + + histogramSeriesLastSample := make(map[string]int) + for i, hw := range opts.Histograms { + key := mapToMetric(hw.Labels).String() + histogramSeriesLastSample[key] = i + } + + for i, hw := range opts.Histograms { + var labelRefs []uint32 + for k, v := range hw.Labels { + labelRefs = append(labelRefs, symbols.Symbolize(k), symbols.Symbolize(v)) + } + + hist := hw.Histogram + hist.Timestamp = now.Add(hw.Offset).UnixMilli() + + var histogramExemplars []writev2.Exemplar + seriesKey := mapToMetric(hw.Labels).String() + isLastHistogramForSeries := histogramSeriesLastSample[seriesKey] == i + + for ei, ew := range opts.Exemplars { + if usedExemplars[ei] { + continue + } + if labelsMatch(mapToMetric(hw.Labels), ew.Labels) { + var exemplarLabelRefs []uint32 + for k, v := range ew.ExemplarLabels { + exemplarLabelRefs = append(exemplarLabelRefs, symbols.Symbolize(k), symbols.Symbolize(v)) + } + histogramExemplars = append(histogramExemplars, writev2.Exemplar{ + LabelsRefs: exemplarLabelRefs, + Value: ew.Value, + Timestamp: now.Add(ew.Offset).UnixMilli(), + }) + usedExemplars[ei] = true + if !isLastHistogramForSeries { + break + } + } + } + + ts := writev2.TimeSeries{ + LabelsRefs: labelRefs, + Histograms: []writev2.Histogram{hist}, + Exemplars: histogramExemplars, + } + if metricName := hw.Labels["__name__"]; metricName != "" || opts.UnsafeRequest { + if metadata, found := metadataByName[metricName]; found { + ts.Metadata = metadata + } + } + timeseries = append(timeseries, ts) + } + + for i, ew := range opts.Exemplars { + if usedExemplars[i] { + continue + } + matched := false + for _, s := range opts.Samples { + if labelsMatch(mapToMetric(s.Labels), ew.Labels) { + matched = true + break + } + } + if !matched { + for _, h := range opts.Histograms { + if labelsMatch(mapToMetric(h.Labels), ew.Labels) { + matched = true + break + } + } + } + if !matched { + var labelRefs []uint32 + for k, v := range ew.Labels { + labelRefs = append(labelRefs, symbols.Symbolize(k), symbols.Symbolize(v)) + } + var exemplarLabelRefs []uint32 + for k, v := range ew.ExemplarLabels { + exemplarLabelRefs = append(exemplarLabelRefs, symbols.Symbolize(k), symbols.Symbolize(v)) + } + timeseries = append(timeseries, writev2.TimeSeries{ + LabelsRefs: labelRefs, + Exemplars: []writev2.Exemplar{{ + LabelsRefs: exemplarLabelRefs, + Value: ew.Value, + Timestamp: now.Add(ew.Offset).UnixMilli(), + }}, + }) + } + } + + for _, mw := range opts.Metadata { + metricName := mw.Labels["__name__"] + if metricName == "" { + continue + } + matched := false + for _, s := range opts.Samples { + if s.Labels["__name__"] == metricName { + matched = true + break + } + } + if !matched { + for _, h := range opts.Histograms { + if h.Labels["__name__"] == metricName { + matched = true + break + } + } + } + if !matched { + var labelRefs []uint32 + for k, v := range mw.Labels { + labelRefs = append(labelRefs, symbols.Symbolize(k), symbols.Symbolize(v)) + } + timeseries = append(timeseries, writev2.TimeSeries{ + LabelsRefs: labelRefs, + Metadata: writev2.Metadata{ + Type: mw.Type, + HelpRef: symbols.Symbolize(mw.Help), + UnitRef: symbols.Symbolize(mw.Unit), + }, + }) + } + } + + req := &writev2.Request{ + Symbols: symbols.Symbols(), + Timeseries: timeseries, + } + data, _ := req.Marshal() + compressed := snappy.Encode(nil, data) + + return &http.Request{ + Method: http.MethodPost, + Header: http.Header{ + "Content-Encoding": []string{"snappy"}, + "Content-Type": []string{"application/x-protobuf;proto=io.prometheus.write.v2.Request"}, + "X-Prometheus-Remote-Write-Version": []string{"2.0.0"}, + }, + Body: io.NopCloser(bytes.NewReader(compressed)), + } +} + +// Histogram creates a writev2.Histogram with the specified parameters. Exported so +// ComplianceTests implementations across files can build histogram test fixtures. +func Histogram(sum float64, includePositive, includeNegative, includeZero, useCustomBuckets, badCount bool) writev2.Histogram { + var positiveSpans, negativeSpans []writev2.BucketSpan + var positiveDeltas, negativeDeltas []int64 + var customValues []float64 + count := uint64(0) + + if includePositive { + positiveSpans = []writev2.BucketSpan{{Offset: 0, Length: 3}, {Offset: 2, Length: 2}} + positiveDeltas = []int64{1, 2, 1, 3, 1} + count += 23 + } + if includeNegative { + negativeSpans = []writev2.BucketSpan{{Offset: 0, Length: 2}} + negativeDeltas = []int64{1, 1} + count += 3 + } + if includeZero { + count++ + } + if useCustomBuckets { + customValues = []float64{0.1, 0.5, 1.0, 2.5, 5.0, 10.0} + positiveSpans = []writev2.BucketSpan{{Offset: 0, Length: 6}} + positiveDeltas = []int64{2, 1, 2, 3, -1, -2} + count = 30 + } + if badCount { + count = 99 + } + + hist := writev2.Histogram{Count: &writev2.Histogram_CountInt{CountInt: count}, Sum: sum} + if includePositive || useCustomBuckets { + hist.PositiveSpans = positiveSpans + hist.PositiveDeltas = positiveDeltas + } + if includeNegative { + hist.NegativeSpans = negativeSpans + hist.NegativeDeltas = negativeDeltas + } + if includeZero { + hist.ZeroCount = &writev2.Histogram_ZeroCountInt{ZeroCountInt: 1} + } + if useCustomBuckets { + hist.Schema = -53 + hist.CustomValues = customValues + } + return hist +}