diff --git a/CHANGELOG.md b/CHANGELOG.md index d2bbb79..a702421 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,7 @@ ## unreleased +* [BUGFIX] Remote write: the path of `--remote-url` is now respected instead of always appending `api/v1/write`. URLs that relied on the previous prefix behavior (own path with `api/v1/write` appended) must now include the full write path. #196 + ## 0.7.0 / 2025-01-14 * [CHANGE] (breaking) Removed the deprecated `--metric-count` flag (use `--gauge-metric-count` instead). #119 diff --git a/README.md b/README.md index f8a9817..3d44ba4 100644 --- a/README.md +++ b/README.md @@ -3,7 +3,7 @@ Avalanche is a load-testing binary capable of generating metrics that can be either: * scraped via [Prometheus scrape formats](https://prometheus.io/docs/instrumenting/exposition_formats/) (including [OpenMetrics](https://github.com/OpenObservability/OpenMetrics)) endpoint. -* written via Prometheus Remote Write (v1 only for now) to a target endpoint. +* written via Prometheus Remote Write (v1, or experimental v2 with `--remote-write-v2`) to a target endpoint. This allows load testing services that can scrape (e.g. Prometheus, OpenTelemetry Collector and so), as well as, services accepting data via Prometheus remote_write API such as [Thanos](https://github.com/thanos-io/thanos), [Cortex](https://github.com/cortexproject/cortex), [M3DB](https://m3db.github.io/m3/integrations/prometheus/), [VictoriaMetrics](https://github.com/VictoriaMetrics/VictoriaMetrics/) and other services [listed here](https://prometheus.io/docs/operating/integrations/#remote-endpoints-and-storage). diff --git a/metricsgen/write.go b/metricsgen/write.go index cd8d43e..065eaa4 100644 --- a/metricsgen/write.go +++ b/metricsgen/write.go @@ -25,6 +25,7 @@ import ( "net/url" "os" "sort" + "strings" "sync" "time" @@ -57,7 +58,7 @@ type ConfigWrite struct { func NewWriteConfigFromFlags(flagReg func(name, help string) *kingpin.FlagClause) *ConfigWrite { cfg := &ConfigWrite{} - flagReg("remote-url", "URL to send samples via remote_write API. By default, path is set to api/v1/write"). + flagReg("remote-url", "URL to send samples via remote_write API. A URL path (other than a bare '/') is used after the client's path cleaning (e.g. trailing slashes are dropped); host-only URLs get the default path api/v1/write appended."). URLVar(&cfg.URL) flagReg("remote-concurrency-limit", "how many concurrent writes can happen at any given time").Default("20"). IntVar(&cfg.Concurrency) @@ -165,11 +166,7 @@ func RunRemoteWriting(ctx context.Context, logger *slog.Logger, cfg *ConfigWrite rt = &userAgentRoundTripper{userAgent: "avalanche", rt: rt} httpClient := &http.Client{Transport: rt} - remoteAPI, err := remote.NewAPI( - cfg.URL.String(), - remote.WithAPIHTTPClient(httpClient), - remote.WithAPILogger(logger.With("component", "remote_write_api")), - ) + remoteAPI, err := newRemoteAPI(cfg, logger, httpClient) if err != nil { return err } @@ -189,6 +186,44 @@ func RunRemoteWriting(ctx context.Context, logger *slog.Logger, cfg *ConfigWrite return writer.write(ctx) } +// newRemoteAPI builds the remote write client. remote.NewAPI takes the base +// URL and the API path separately and path.Join-s the latter onto the former, +// so a --remote-url that already carries a path (Thanos Receive's +// /api/v1/receive, say) grew the default api/v1/write on top of it. Split the +// flag's URL along the same seam: everything but the path becomes the base, +// the path becomes the API path, subject to the client's path cleaning, so +// trailing and duplicate slashes are dropped. Host-only URLs and a bare "/" +// keep the client's default path, api/v1/write; posting samples to "/" is +// almost never intended. +// See https://github.com/prometheus-community/avalanche/issues/173. +func newRemoteAPI(cfg *ConfigWrite, logger *slog.Logger, httpClient *http.Client) (*remote.API, error) { + // url.URL carries the path twice: Path decoded, and RawPath holding the + // original spelling whenever it is not the canonical encoding of Path. + // remote.NewAPI assigns its joined path to Path alone, so passing the path + // separately from the base URL leaves RawPath behind and the endpoint sees + // the decoded form. That is correct for unreserved characters (%61 is just + // "a"), but an encoded separator is not a separator: %2F would silently + // turn one segment into two and post the samples somewhere else. Slashes + // are the only delimiter at risk, since url.URL escapes the others when it + // re-encodes a path, so comparing slash counts detects exactly that case. + if cfg.URL.RawPath != "" && + strings.Count(cfg.URL.Path, "/") != strings.Count(cfg.URL.RawPath, "/") { + return nil, fmt.Errorf("--remote-url path %q contains an escaped separator that the remote write client cannot preserve; it would post to %q instead", cfg.URL.EscapedPath(), cfg.URL.Path) + } + + opts := []remote.APIOption{ + remote.WithAPIHTTPClient(httpClient), + remote.WithAPILogger(logger.With("component", "remote_write_api")), + } + + baseURL := *cfg.URL + baseURL.Path, baseURL.RawPath = "", "" + if apiPath := cfg.URL.Path; apiPath != "" && apiPath != "/" { + opts = append(opts, remote.WithAPIPath(apiPath)) + } + return remote.NewAPI(baseURL.String(), opts...) +} + // Add the tenant ID header func (rt *tenantRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) { req = cloneRequest(req) diff --git a/metricsgen/write_test.go b/metricsgen/write_test.go index 9fc380a..fc3fdf1 100644 --- a/metricsgen/write_test.go +++ b/metricsgen/write_test.go @@ -14,10 +14,20 @@ package metricsgen import ( + "context" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "net/url" + "sync" "testing" "time" + "github.com/prometheus/client_golang/exp/api/remote" + writev2 "github.com/prometheus/client_golang/exp/api/remote/genproto/v2" "github.com/prometheus/prometheus/prompb" + "github.com/stretchr/testify/require" ) func TestShuffleTimestamps(t *testing.T) { @@ -58,3 +68,81 @@ func TestShuffleTimestamps(t *testing.T) { t.Error("Timestamps are not out of order") } } + +func TestNewRemoteAPIPath(t *testing.T) { + for _, tc := range []struct { + name string + urlPath string + wantPath string + wantQuery string + wantErr string + }{ + {name: "no path appends default", urlPath: "", wantPath: "/api/v1/write"}, + {name: "root path appends default", urlPath: "/", wantPath: "/api/v1/write"}, + {name: "custom path is respected", urlPath: "/api/v1/receive", wantPath: "/api/v1/receive"}, + {name: "trailing slash is cleaned", urlPath: "/api/v1/receive/", wantPath: "/api/v1/receive"}, + {name: "prefix path is respected", urlPath: "/prometheus", wantPath: "/prometheus"}, + {name: "query string preserved", urlPath: "/api/v1/receive?tenant=a", wantPath: "/api/v1/receive", wantQuery: "tenant=a"}, + // Escaping an unreserved character says nothing about the endpoint, + // so decoding it is plain normalization. Escaping a separator does + // say something, and the client cannot carry it, so it is rejected + // rather than silently posted to a different path. + {name: "escaped unreserved character is normalized", urlPath: "/ten%61nt", wantPath: "/tenant"}, + {name: "escaped separator is rejected", urlPath: "/tenant%2Freceive", wantErr: "contains an escaped separator"}, + {name: "double slash is cleaned", urlPath: "//foo", wantPath: "/foo"}, + } { + t.Run(tc.name, func(t *testing.T) { + var ( + mu sync.Mutex + gotPath string + gotQuery string + ) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + gotPath = r.URL.EscapedPath() + gotQuery = r.URL.RawQuery + mu.Unlock() + w.Header().Set("X-Prometheus-Remote-Write-Samples-Written", "0") + w.Header().Set("X-Prometheus-Remote-Write-Histograms-Written", "0") + w.Header().Set("X-Prometheus-Remote-Write-Exemplars-Written", "0") + w.WriteHeader(http.StatusNoContent) + })) + t.Cleanup(srv.Close) + + u, err := url.Parse(srv.URL + tc.urlPath) + require.NoError(t, err) + + api, err := newRemoteAPI( + &ConfigWrite{URL: u}, + slog.New(slog.NewTextHandler(io.Discard, nil)), + srv.Client(), + ) + if tc.wantErr != "" { + require.ErrorContains(t, err, tc.wantErr) + return + } + require.NoError(t, err) + + for _, msg := range []struct { + typ remote.WriteMessageType + req any + }{ + {typ: remote.WriteV1MessageType, req: &prompb.WriteRequest{}}, + {typ: remote.WriteV2MessageType, req: &writev2.Request{Symbols: []string{""}}}, + } { + mu.Lock() + gotPath = "" + gotQuery = "" + mu.Unlock() + + _, err = api.Write(context.Background(), msg.typ, msg.req) + require.NoError(t, err) + + mu.Lock() + require.Equal(t, tc.wantPath, gotPath) + require.Equal(t, tc.wantQuery, gotQuery) + mu.Unlock() + } + }) + } +}