test(promqle2etest): add vanilla Prometheus remote write 2.0 to GCM test case

google/internal/promqle2etest/backend_prw_gcm.go

@@ -0,0 +1,169 @@
+// Copyright 2026 Google LLC
+//
+// 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 promqle2etest
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"strconv"
+	"testing"
+
+	gcm "cloud.google.com/go/monitoring/apiv3/v2"
+	"github.com/efficientgo/e2e"
+	e2emon "github.com/efficientgo/e2e/monitoring"
+	"github.com/prometheus/client_golang/api"
+	v1 "github.com/prometheus/client_golang/api/prometheus/v1"
+	"github.com/prometheus/compliance/promqle2e"
+	"golang.org/x/oauth2"
+	"golang.org/x/oauth2/google"
+)
+
+var _ promqle2e.Backend = PrometheusRemoteWriteGCMBackend{}
+
+// PrometheusRemoteWriteGCMBackend represents a vanilla Prometheus scraping
+// metrics and pushing to GCM API via Prometheus Remote Write 2.0 protocol.
+// This generally follows https://cloud.google.com/stackdriver/docs/managed-prometheus/setup-unmanaged.
+type PrometheusRemoteWriteGCMBackend struct {
+	Image string
+	Name  string
+	GCMSA []byte
+}
+
+func (p PrometheusRemoteWriteGCMBackend) Ref() string {
+	return p.Name
+}
+
+// newPrometheusRemoteWrite creates a new Prometheus runnable configured with remote write 2.0 against GCM.
+func newPrometheusRemoteWrite(env e2e.Environment, name string, image string, scrapeTargetAddress string, projectID string, location string, cluster string, gcmSA []byte) *e2emon.Prometheus {
+	ports := map[string]int{"http": 9090}
+
+	f := env.Runnable(name).WithPorts(ports).Future()
+	credsFile := filepath.Join(f.Dir(), "gcm-sa.json")
+	if err := os.WriteFile(credsFile, gcmSA, 0600); err != nil {
+		return &e2emon.Prometheus{Runnable: e2e.NewFailedRunnable(name, fmt.Errorf("write JSON creds failed: %w", err))}
+	}
+
+	config := fmt.Sprintf(`
+global:
+  scrape_interval: 5s
+  scrape_timeout: 5s
+  convert_classic_histograms_to_nhcb: true
+  external_labels:
+    collector: %v
+    project_id: %v
+    location: %v
+    cluster: %v
+scrape_configs:
+- job_name: 'test'
+  scrape_interval: 5s
+  scrape_timeout: 5s
+  static_configs:
+  - targets: [%s]
+  metric_relabel_configs:
+  - regex: instance
+    action: labeldrop
+remote_write:
+- name: "google_cloud"
+  url: "https://staging-monitoring.sandbox.googleapis.com/v1/prometheus/api/v1/write"
+  protobuf_message: "io.prometheus.write.v2.Request"
+  send_exemplars: true
+  queue_config:
+    retry_on_http_429: true
+  google_iam:
+    credentials_file: "%s"
+  write_relabel_configs:
+  - source_labels: ['__type__']
+    action: labeldrop
+  - source_labels: ['__unit__']
+    action: labeldrop
+`, name, projectID, location, cluster, scrapeTargetAddress, credsFile)
+	if err := os.WriteFile(filepath.Join(f.Dir(), "prometheus.yml"), []byte(config), 0600); err != nil {
+		return &e2emon.Prometheus{Runnable: e2e.NewFailedRunnable(name, fmt.Errorf("create prometheus config failed: %w", err))}
+	}
+
+	args := map[string]string{
+		"--web.listen-address":                    fmt.Sprintf(":%d", ports["http"]),
+		"--config.file":                           filepath.Join(f.Dir(), "prometheus.yml"),
+		"--storage.tsdb.path":                     f.Dir(),
+		"--enable-feature=exemplar-storage":       "",
+		"--enable-feature=native-histograms":      "",
+		"--enable-feature=promql-nhcb-as-classic": "",
+		"--enable-feature=st-storage":             "",
+		"--enable-feature=st-synthesis":           "",
+		"--enable-feature=type-and-unit-labels":   "",
+		"--storage.tsdb.no-lockfile":              "",
+		"--storage.tsdb.retention.time":           "1d",
+		"--storage.tsdb.wal-compression":          "",
+		"--storage.tsdb.min-block-duration":       "2h",
+		"--storage.tsdb.max-block-duration":       "2h",
+		"--web.enable-lifecycle":                  "",
+		"--log.format":                            "json",
+		"--log.level":                             "info",
+	}
+
+	p := e2emon.AsInstrumented(f.Init(e2e.StartOptions{
+		Image:     image,
+		Command:   e2e.NewCommandWithoutEntrypoint("prometheus", e2e.BuildArgs(args)...),
+		Readiness: e2e.NewHTTPReadinessProbe("http", "/-/ready", 200, 200),
+		User:      strconv.Itoa(os.Getuid()),
+	}), "http")
+
+	return &e2emon.Prometheus{
+		Runnable:     p,
+		Instrumented: p,
+	}
+}
+
+func (p PrometheusRemoteWriteGCMBackend) StartAndWaitReady(t testing.TB, env e2e.Environment) promqle2e.RunningBackend {
+	t.Helper()
+
+	ctx := t.Context()
+
+	creds, err := google.CredentialsFromJSON(ctx, p.GCMSA, gcm.DefaultAuthScopes()...)
+	if err != nil {
+		t.Fatalf("create credentials from JSON: %s", err)
+	}
+
+	// Fake, does not matter.
+	cluster := "pe-github-action"
+	location := "europe-west3-a"
+
+	cl, err := api.NewClient(api.Config{
+		Address: fmt.Sprintf("https://staging-monitoring.sandbox.googleapis.com/v1/projects/%s/location/global/prometheus", creds.ProjectID),
+		Client:  oauth2.NewClient(ctx, creds.TokenSource),
+	})
+	if err != nil {
+		t.Fatalf("create Prometheus client: %s", err)
+	}
+
+	replayer := promqle2e.StartIngestByScrapeReplayer(t, env)
+	prom := newPrometheusRemoteWrite(env, p.Name, p.Image, replayer.Endpoint(env), creds.ProjectID, location, cluster, p.GCMSA)
+	if err := e2e.StartAndWaitReady(prom); err != nil {
+		t.Fatal(err)
+	}
+
+	return promqle2e.NewRunningScrapeReplayBasedBackend(
+		replayer,
+		map[string]string{
+			"cluster":    cluster,
+			"location":   location,
+			"project_id": creds.ProjectID,
+			"collector":  p.Name,
+			"job":        "test",
+		},
+		v1.NewAPI(cl),
+	)
+}

google/internal/promqle2etest/gcm_test.go

@@ -65,8 +65,12 @@ func gmpPrometheusImageOrFail(t testing.TB) string {
 	return image
 }
 
-// TODO(bwplotka): Add target --PromProto--> Prometheus vanilla --PRW 2.0--> GCM case once GCM exposes PRW 2.0.
-func setupBackends(t testing.TB) (promqle2e.PrometheusBackend, PrometheusForkGCMBackend, LocalExportGCMBackend) {
+// setupBackends sets up the backends used in acceptance tests:
+// - prom: target --PromProto--> Prometheus (vanilla OSS reference behaviour)
+// - gmpPromGCM: target --PromProto--> Prometheus GMP fork --GCM API--> GCM
+// - promPRWGCM: target --PromProto--> Prometheus vanilla --PRW 2.0--> GCM
+// - localExportGCM: local prometheus-engine/pkg/export code --GCM API--> GCM
+func setupBackends(t testing.TB) (promqle2e.PrometheusBackend, PrometheusForkGCMBackend, PrometheusRemoteWriteGCMBackend, LocalExportGCMBackend) {
 	// target --PromProto--> Prometheus (referencing OSS behaviour).
 	prom := promqle2e.PrometheusBackend{
 		Name:  "prom",
@@ -78,20 +82,26 @@ func setupBackends(t testing.TB) (promqle2e.PrometheusBackend, PrometheusForkGCM
 		Image: gmpPrometheusImageOrFail(t),
 		GCMSA: gcmServiceAccountOrFail(t),
 	}
+	// target --PromProto--> Prometheus vanilla --PRW 2.0--> GCM
+	promPRWGCM := PrometheusRemoteWriteGCMBackend{
+		Name:  "prom-prw-gcm",
+		Image: prom.Image,
+		GCMSA: gcmServiceAccountOrFail(t),
+	}
 	// local prometheus-engine/pkg/export code --GCM API--> GCM.
 	localExportGCM := LocalExportGCMBackend{
 		Name:  "local-export-gcm",
 		GCMSA: gcmServiceAccountOrFail(t),
 	}
-	return prom, gmpPromGCM, localExportGCM
+	return prom, gmpPromGCM, promPRWGCM, localExportGCM
 }
 
 // TestExportGCM_PrometheusCounter_NoCT tests a counter sample behaviour
 // when the CT is not provided in the scrape.
 func TestExportGCM_PrometheusCounter_NoCT(t *testing.T) {
 	const interval = 15 * time.Second
 
-	prom, gmpPromGCM, localExportGCM := setupBackends(t)
+	prom, gmpPromGCM, promPRWGCM, localExportGCM := setupBackends(t)
 
 	pt := promqle2e.NewScrapeStyleTest(t)
 	pt.SetCurrentTime(time.Now().Add(-10 * time.Minute)) // We only do a few scrapes, so -10m buffer is enough.
@@ -119,12 +129,14 @@ func TestExportGCM_PrometheusCounter_NoCT(t *testing.T) {
 	pt.RecordScrape(interval).
 		Expect(c, 10, localExportGCM).
 		Expect(c, 10, gmpPromGCM).
+		Expect(c, 10, promPRWGCM).
 		Expect(c, 210, prom)
 
 	c.Add(40)
 	pt.RecordScrape(interval).
 		Expect(c, 50, localExportGCM).
 		Expect(c, 50, gmpPromGCM).
+		Expect(c, 50, promPRWGCM).
 		Expect(c, 250, prom)
 
 	// Reset to 0 (simulating instrumentation resetting metric or restarting target).
@@ -133,12 +145,14 @@ func TestExportGCM_PrometheusCounter_NoCT(t *testing.T) {
 	pt.RecordScrape(interval).
 		Expect(c, 0, localExportGCM).
 		Expect(c, 0, gmpPromGCM).
+		Expect(c, 0, promPRWGCM).
 		Expect(c, 0, prom)
 
 	c.Add(150)
 	pt.RecordScrape(interval).
 		Expect(c, 150, localExportGCM).
 		Expect(c, 150, gmpPromGCM).
+		Expect(c, 150, promPRWGCM).
 		Expect(c, 150, prom)
 
 	// Reset to 0 with addition.
@@ -148,18 +162,21 @@ func TestExportGCM_PrometheusCounter_NoCT(t *testing.T) {
 	pt.RecordScrape(interval).
 		Expect(c, 20, localExportGCM).
 		Expect(c, 20, gmpPromGCM).
+		Expect(c, 20, promPRWGCM).
 		Expect(c, 20, prom)
 
 	c.Add(50)
 	pt.RecordScrape(interval).
 		Expect(c, 70, localExportGCM).
 		Expect(c, 70, gmpPromGCM).
+		Expect(c, 70, promPRWGCM).
 		Expect(c, 70, prom)
 
 	c.Add(10)
 	pt.RecordScrape(interval).
 		Expect(c, 80, localExportGCM).
 		Expect(c, 80, gmpPromGCM).
+		Expect(c, 80, promPRWGCM).
 		Expect(c, 80, prom)
 
 	// Tricky reset case, unnoticeable reset for Prometheus without created timestamp as well.
@@ -169,6 +186,7 @@ func TestExportGCM_PrometheusCounter_NoCT(t *testing.T) {
 	pt.RecordScrape(interval).
 		Expect(c, 600, localExportGCM).
 		Expect(c, 600, gmpPromGCM).
+		Expect(c, 600, promPRWGCM).
 		Expect(c, 600, prom)
 
 	// Prometheus SDK used for replies actually emit CTs.
@@ -197,7 +215,7 @@ func TestExportGCM_PrometheusCounter_NoCT(t *testing.T) {
 func TestExportGCM_PrometheusCounter_WithCT(t *testing.T) {
 	const interval = 15 * time.Second
 
-	prom, gmpPromGCM, localExportGCM := setupBackends(t)
+	prom, gmpPromGCM, promPRWGCM, localExportGCM := setupBackends(t)
 
 	pt := promqle2e.NewScrapeStyleTest(t)
 	pt.SetCurrentTime(time.Now().Add(-10 * time.Minute)) // We only do a few scrapes, so -10m buffer is enough.
@@ -216,6 +234,7 @@ func TestExportGCM_PrometheusCounter_WithCT(t *testing.T) {
 	c = counter.WithLabelValues("bar")
 	c.Add(200)
 	pt.RecordScrape(interval).
+		Expect(c, 200, promPRWGCM).
 		Expect(c, 200, prom)
 	// Nothing is expected for GCM due to cannibalization required if the target does not emit CT (which this metric does not).
 	// See https://cloud.google.com/stackdriver/docs/managed-prometheus/troubleshooting#counter-sums
@@ -225,12 +244,14 @@ func TestExportGCM_PrometheusCounter_WithCT(t *testing.T) {
 	pt.RecordScrape(interval).
 		Expect(c, 10, localExportGCM).
 		Expect(c, 10, gmpPromGCM).
+		Expect(c, 210, promPRWGCM).
 		Expect(c, 210, prom)
 
 	c.Add(40)
 	pt.RecordScrape(interval).
 		Expect(c, 50, localExportGCM).
 		Expect(c, 50, gmpPromGCM).
+		Expect(c, 250, promPRWGCM).
 		Expect(c, 250, prom)
 
 	// Reset to 0 (simulating instrumentation resetting metric or restarting target).
@@ -239,12 +260,14 @@ func TestExportGCM_PrometheusCounter_WithCT(t *testing.T) {
 	pt.RecordScrape(interval).
 		Expect(c, 0, localExportGCM).
 		Expect(c, 0, gmpPromGCM).
+		Expect(c, 0, promPRWGCM).
 		Expect(c, 0, prom)
 
 	c.Add(150)
 	pt.RecordScrape(interval).
 		Expect(c, 150, localExportGCM).
 		Expect(c, 150, gmpPromGCM).
+		Expect(c, 150, promPRWGCM).
 		Expect(c, 150, prom)
 
 	// Reset to 0 with addition.
@@ -254,18 +277,21 @@ func TestExportGCM_PrometheusCounter_WithCT(t *testing.T) {
 	pt.RecordScrape(interval).
 		Expect(c, 20, localExportGCM).
 		Expect(c, 20, gmpPromGCM).
+		Expect(c, 20, promPRWGCM).
 		Expect(c, 20, prom)
 
 	c.Add(50)
 	pt.RecordScrape(interval).
 		Expect(c, 70, localExportGCM).
 		Expect(c, 70, gmpPromGCM).
+		Expect(c, 70, promPRWGCM).
 		Expect(c, 70, prom)
 
 	c.Add(10)
 	pt.RecordScrape(interval).
 		Expect(c, 80, localExportGCM).
 		Expect(c, 80, gmpPromGCM).
+		Expect(c, 80, promPRWGCM).
 		Expect(c, 80, prom)
 
 	// Tricky reset case, unnoticeable reset for Prometheus without created timestamp as well.
@@ -275,6 +301,7 @@ func TestExportGCM_PrometheusCounter_WithCT(t *testing.T) {
 	pt.RecordScrape(interval).
 		Expect(c, 600, localExportGCM).
 		Expect(c, 600, gmpPromGCM).
+		Expect(c, 600, promPRWGCM).
 		Expect(c, 600, prom)
 
 	// Prometheus SDK supports CTs. This "transform" validates that invariance.
@@ -301,7 +328,7 @@ func TestExportGCM_PrometheusCounter_WithCT(t *testing.T) {
 func TestExportGCM_PrometheusGauge(t *testing.T) {
 	const interval = 15 * time.Second
 
-	prom, gmpPromGCM, localExportGCM := setupBackends(t)
+	prom, gmpPromGCM, promPRWGCM, localExportGCM := setupBackends(t)
 
 	pt := promqle2e.NewScrapeStyleTest(t)
 	pt.SetCurrentTime(time.Now().Add(-10 * time.Minute)) // We only do a few scrapes, so -10m buffer is enough.
@@ -322,18 +349,21 @@ func TestExportGCM_PrometheusGauge(t *testing.T) {
 	pt.RecordScrape(interval).
 		Expect(g, 200, localExportGCM).
 		Expect(g, 200, gmpPromGCM).
+		Expect(g, 200, promPRWGCM).
 		Expect(g, 200, prom)
 
 	g.Sub(10)
 	pt.RecordScrape(interval).
 		Expect(g, 190, localExportGCM).
 		Expect(g, 190, gmpPromGCM).
+		Expect(g, 190, promPRWGCM).
 		Expect(g, 190, prom)
 
 	g.Add(40)
 	pt.RecordScrape(interval).
 		Expect(g, 230, localExportGCM).
 		Expect(g, 230, gmpPromGCM).
+		Expect(g, 230, promPRWGCM).
 		Expect(g, 230, prom)
 
 	// Reset to 0 (simulating instrumentation resetting metric or restarting target).
@@ -342,6 +372,7 @@ func TestExportGCM_PrometheusGauge(t *testing.T) {
 	pt.RecordScrape(interval).
 		Expect(g, 0, localExportGCM).
 		Expect(g, 0, gmpPromGCM).
+		Expect(g, 0, promPRWGCM).
 		Expect(g, 0, prom)
 
 	ctx, cancel := context.WithTimeout(t.Context(), 5*time.Minute)
@@ -355,7 +386,7 @@ func TestExportGCM_MetricHelpIngestion(t *testing.T) {
 		mName    = "promqle2e_test_gauge_help"
 	)
 
-	_, _, localExportGCM := setupBackends(t)
+	_, _, _, localExportGCM := setupBackends(t)
 
 	pt := promqle2e.NewScrapeStyleTest(t)
 	pt.SetCurrentTime(time.Now().Add(-10 * time.Minute)) // We only do a few scrapes, so -10m buffer is enough.