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 +}