From b38b556fbd3633fb34489beb8f9408328dcb8a09 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Senart?= Date: Tue, 29 Sep 2026 23:23:17 -0700 Subject: [PATCH] relay: add the pprof upstream kind A customer names Go processes as targets of one pprof upstream; Perfloop names one target per read as the first path segment. The relay forwards only GET to the six documented /debug/pprof routes, admits `seconds` as the only query key within the configured cap (profile must carry it), and refuses a further read at once when the per-process in-flight cap is reached. Every refusal happens before any upstream request and carries the relay's own marker on Perfloop-Relay-Error, so a reader tells it from the process's answer. -print-routes lists each target and the caps, per process. link carries the marker values and the seconds bound both ends share. Ported from perfloop/perfloop#4331 (0bf7906fc8, 2a5fec2b88, 53d50ae9de, 1ecd357bf0, aebcedfdd9, a1a05b8c96) over the hardening in ce5260e8. Co-Authored-By: Claude Fable 5.1 --- cmd/perfloop-relay/main_test.go | 2 +- config.go | 74 +++++++++++++++++++-- config_test.go | 55 ++++++++++++++++ link/link.go | 19 ++++++ relay.go | 63 +++++++++++++++--- relay_test.go | 113 ++++++++++++++++++++++++++++++++ routes.go | 71 +++++++++++++++++++- routes_test.go | 70 ++++++++++++++++++++ 8 files changed, 448 insertions(+), 19 deletions(-) diff --git a/cmd/perfloop-relay/main_test.go b/cmd/perfloop-relay/main_test.go index 11fa27d..35b83ad 100644 --- a/cmd/perfloop-relay/main_test.go +++ b/cmd/perfloop-relay/main_test.go @@ -25,7 +25,7 @@ func TestPrintRoutesPrintsTheTableOrValidatesError(t *testing.T) { } bad := filepath.Join(t.TempDir(), "relay.yaml") for body, want := range map[string]string{ - "api: https://app.perfloop.ai\ntoken: x\nupstreams:\n - name: vm\n kind: pprof\n url: http://vm.internal\n": `kind "pprof" is not supported`, + "api: https://app.perfloop.ai\ntoken: x\nupstreams:\n - name: vm\n kind: pyroscope\n url: http://vm.internal\n": `kind "pyroscope" is not supported`, "api: https://app.perfloop.ai\ntoken: x\nlog_level: nonsense\nupstreams:\n - name: vm\n kind: prometheus\n url: http://vm.internal\n": `log_level:`, } { if err := os.WriteFile(bad, []byte(body), 0o600); err != nil { diff --git a/config.go b/config.go index 552310c..415214b 100644 --- a/config.go +++ b/config.go @@ -41,12 +41,28 @@ type Upstream struct { // lowercase DNS label. Name string `yaml:"name"` // Kind selects the read routes the relay permits; routes.go lists the - // kinds (today prometheus, victoriametrics, and loki). It selects the read - // routes the relay permits. + // kinds (today prometheus, victoriametrics, loki, and pprof). Kind string `yaml:"kind"` // URL is the provider base: scheme, host, port, and an optional path - // prefix. The relay appends the proxy's request path to it. + // prefix. The relay appends the proxy's request path to it. A pprof + // upstream has Targets instead. URL string `yaml:"url"` + // Targets are the processes of a pprof upstream: target name to the + // http(s) base of that process's net/http/pprof handler. Perfloop names + // one target per read as the first path segment; a name not listed here + // is refused. Each name is one lowercase DNS label. + Targets map[string]string `yaml:"targets"` + // MaxSeconds caps the `seconds` a pprof read may ask for, so a CPU + // profile cannot run longer than the customer allows. Zero means + // DefaultMaxSeconds. The bound is link.MaxProfileSeconds. + MaxSeconds int `yaml:"max_seconds"` + // MaxConcurrent caps the pprof reads in flight on this upstream across + // all of its targets, per relay process; a further read is refused at + // once with status 429. Perfloop spreads reads over every relay a tenant + // runs with the same token, so N relay processes admit N times this cap. + // The customer controls that count. Zero means DefaultMaxConcurrent. The + // bound is MaxConcurrentBound. + MaxConcurrent int `yaml:"max_concurrent"` // Headers are added to every upstream request, for example an // Authorization header holding a read token. They replace any header of // the same name from the proxy. @@ -56,6 +72,20 @@ type Upstream struct { CA string `yaml:"ca"` } +const ( + // DefaultMaxSeconds is the `seconds` cap of a pprof upstream that sets + // none: net/http/pprof's own default CPU profile length. The largest cap + // a config may set is link.MaxProfileSeconds. + DefaultMaxSeconds = 30 + // DefaultMaxConcurrent is the in-flight cap of a pprof upstream that + // sets none: one profile at a time per upstream. + DefaultMaxConcurrent = 1 + // MaxConcurrentBound is the largest max_concurrent a config may set. It + // is a config bound on how many processes profile at once, not a tunnel + // property: the relay keeps two tunnels of maxStreams each. + MaxConcurrentBound = 8 +) + // Load reads, expands, and strictly parses a relay configuration. New // validates it. func Load(path string) (Config, error) { @@ -102,9 +132,34 @@ func (c Config) Validate() error { if _, ok := base(up.Kind); !ok { return fmt.Errorf("upstream %s: kind %q is not supported", up.Name, up.Kind) } - u, err := url.Parse(up.URL) - if err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" || u.User != nil || u.RawQuery != "" || u.Fragment != "" || strings.HasSuffix(u.Path, "/") { - return fmt.Errorf("upstream %s: url must be an http(s) base without query, fragment, user info, or trailing slash", up.Name) + if up.Kind == "pprof" { + if up.URL != "" { + return fmt.Errorf("upstream %s: a pprof upstream names targets, not a url", up.Name) + } + if len(up.Targets) == 0 { + return fmt.Errorf("upstream %s: at least one target is required", up.Name) + } + for target, raw := range up.Targets { + if !link.ValidUpstreamName(target) { + return fmt.Errorf("upstream %s: target name %q must be one lowercase DNS label", up.Name, target) + } + if !validBase(raw) { + return fmt.Errorf("upstream %s: target %s: url must be an http(s) base without query, fragment, user info, or trailing slash", up.Name, target) + } + } + if up.MaxSeconds < 0 || up.MaxSeconds > link.MaxProfileSeconds { + return fmt.Errorf("upstream %s: max_seconds must be between 1 and %d", up.Name, link.MaxProfileSeconds) + } + if up.MaxConcurrent < 0 || up.MaxConcurrent > MaxConcurrentBound { + return fmt.Errorf("upstream %s: max_concurrent must be between 1 and %d", up.Name, MaxConcurrentBound) + } + } else { + if up.Targets != nil || up.MaxSeconds != 0 || up.MaxConcurrent != 0 { + return fmt.Errorf("upstream %s: targets, max_seconds, and max_concurrent apply only to kind pprof", up.Name) + } + if !validBase(up.URL) { + return fmt.Errorf("upstream %s: url must be an http(s) base without query, fragment, user info, or trailing slash", up.Name) + } } // Header names and values must be ones the HTTP client will send, and // one wire header must have one value: `Authorization` and @@ -142,6 +197,13 @@ func (c Config) Level() slog.Level { return level } +// validBase reports whether raw is an http(s) base the relay may append a +// route to: scheme, host, port, and an optional path prefix, nothing else. +func validBase(raw string) bool { + u, err := url.Parse(raw) + return err == nil && (u.Scheme == "http" || u.Scheme == "https") && u.Host != "" && u.User == nil && u.RawQuery == "" && u.Fragment == "" && !strings.HasSuffix(u.Path, "/") +} + // apiURL is the API origin: an https URL with a host and nothing else. The // token travels only inside TLS. func (c Config) apiURL() (*url.URL, error) { diff --git a/config_test.go b/config_test.go index 93a763b..84ef9d8 100644 --- a/config_test.go +++ b/config_test.go @@ -5,6 +5,8 @@ import ( "path/filepath" "strings" "testing" + + "github.com/perfloop/relay/link" ) func TestLoadExpandsSecretsAndValidates(t *testing.T) { @@ -29,6 +31,13 @@ upstreams: kind: loki url: https://loki-gateway.monitoring.svc ca: /etc/perfloop-relay/loki-ca.pem + - name: go + kind: pprof + targets: + api-1: http://10.0.0.1:6060 + api-2: http://10.0.0.2:6060/internal + max_seconds: 10 + max_concurrent: 2 `) cfg, err := Load(path) if err != nil { @@ -37,6 +46,12 @@ upstreams: if cfg.Token != "tenant-seven" || cfg.Upstreams[0].Headers["Authorization"] != "Bearer vm-read" || cfg.Upstreams[1].CA == "" { t.Fatalf("loaded %+v", cfg) } + if go1 := cfg.Upstreams[2]; len(go1.Targets) != 2 || go1.Targets["api-2"] != "http://10.0.0.2:6060/internal" || go1.MaxSeconds != 10 || go1.MaxConcurrent != 2 { + t.Fatalf("loaded pprof upstream %+v", go1) + } + if err := cfg.Validate(); err != nil { + t.Fatal(err) + } write("api: https://app.perfloop.ai\ntoken: ${RELAY_TEST_MISSING}\nupstreams: [{name: vm, kind: prometheus, url: http://vm:8428}]\n") if _, err := Load(path); err == nil || !strings.Contains(err.Error(), "RELAY_TEST_MISSING") { t.Fatalf("unset variable: %v", err) @@ -77,6 +92,32 @@ upstreams: "one header spelled twice": func(c *Config) { c.Upstreams[0].Headers = map[string]string{"Authorization": "Bearer x", "authorization": "Bearer y"} }, + "targets on a metrics upstream": func(c *Config) { c.Upstreams[0].Targets = map[string]string{"a": "http://a:6060"} }, + "max_seconds on a metrics upstream": func(c *Config) { c.Upstreams[0].MaxSeconds = 10 }, + "pprof with url": func(c *Config) { c.Upstreams[0] = pprof(func(u *Upstream) { u.URL = "http://a:6060" }) }, + "pprof without targets": func(c *Config) { c.Upstreams[0] = pprof(func(u *Upstream) { u.Targets = nil }) }, + "pprof target name with a dot": func(c *Config) { + c.Upstreams[0] = pprof(func(u *Upstream) { u.Targets = map[string]string{"api.1": "http://a:6060"} }) + }, + "pprof target name with a slash": func(c *Config) { + c.Upstreams[0] = pprof(func(u *Upstream) { u.Targets = map[string]string{"api/1": "http://a:6060"} }) + }, + "pprof target url with query": func(c *Config) { + c.Upstreams[0] = pprof(func(u *Upstream) { u.Targets = map[string]string{"a": "http://a:6060/?x=1"} }) + }, + "pprof target url with user": func(c *Config) { + c.Upstreams[0] = pprof(func(u *Upstream) { u.Targets = map[string]string{"a": "http://u:p@a:6060"} }) + }, + "pprof target url with trailing slash": func(c *Config) { + c.Upstreams[0] = pprof(func(u *Upstream) { u.Targets = map[string]string{"a": "http://a:6060/"} }) + }, + "pprof max_seconds over the bound": func(c *Config) { + c.Upstreams[0] = pprof(func(u *Upstream) { u.MaxSeconds = link.MaxProfileSeconds + 1 }) + }, + "pprof max_seconds negative": func(c *Config) { c.Upstreams[0] = pprof(func(u *Upstream) { u.MaxSeconds = -1 }) }, + "pprof max_concurrent over the bound": func(c *Config) { + c.Upstreams[0] = pprof(func(u *Upstream) { u.MaxConcurrent = MaxConcurrentBound + 1 }) + }, } { cfg := base cfg.Upstreams = append([]Upstream(nil), base.Upstreams...) @@ -86,3 +127,17 @@ upstreams: } } } + +// pprof is a valid pprof upstream with one change applied. +func pprof(change func(*Upstream)) Upstream { + up := Upstream{Name: "go", Kind: "pprof", Targets: map[string]string{"api-1": "http://10.0.0.1:6060"}} + change(&up) + return up +} + +func TestValidPprofUpstream(t *testing.T) { + cfg := Config{API: "https://app.perfloop.ai", Token: "t", Upstreams: []Upstream{pprof(func(*Upstream) {})}} + if err := cfg.Validate(); err != nil { + t.Fatal(err) + } +} diff --git a/link/link.go b/link/link.go index 1cc7b60..e47bf38 100644 --- a/link/link.go +++ b/link/link.go @@ -37,6 +37,25 @@ const ( // it, its tunnel token covers it, and the relay logs it and drops it // before the upstream request. It is absent on a read no tool call made. RequestIDHeader = "Perfloop-Request-Id" + // ErrorHeader marks a response that is not the provider's: Perfloop made + // it (Unavailable) or the relay refused the read (Refused). A reader can + // tell either from provider bytes; the API and the proxy act only on + // Unavailable, and a relay refusal reaches the adapter with its status. + ErrorHeader = "Perfloop-Relay-Error" + // Unavailable is the value Perfloop sets when no relay could serve the + // read: no tunnel, or a token the replica does not recognize. + Unavailable = "1" + // Refused is the value the relay sets on its own refusals: an unknown + // upstream or target, a route or query outside the allowlist, or the + // in-flight cap. The status says which. + Refused = "relay" + // MaxProfileSeconds is the longest `seconds` a pprof read may ask for on + // either end: the relay refuses a larger `max_seconds` in its config, and + // Perfloop's pprof adapter refuses a longer capture before the tunnel. A + // profile read crosses two one-minute deadlines, the controller's remote + // client and the proxy's read; 45 leaves them 15 seconds for the tunnel + // and the bytes. + MaxProfileSeconds = 45 ) // hopHeaders never cross a tunnel in either direction (RFC 9110 ยง 7.6.1), diff --git a/relay.go b/relay.go index be67879..294df44 100644 --- a/relay.go +++ b/relay.go @@ -7,6 +7,7 @@ package relay import ( + "cmp" "context" "crypto/tls" "crypto/x509" @@ -73,8 +74,14 @@ type Relay struct { type upstream struct { Upstream - base *url.URL - client *http.Client + // base is the provider URL of a metrics or logs upstream. + base *url.URL + // targets are the process bases of a pprof upstream, by target name. + targets map[string]*url.URL + // slots is the pprof in-flight cap: one token per permitted read. + slots chan struct{} + maxSeconds int + client *http.Client } // New validates the configuration and builds a relay. @@ -126,9 +133,16 @@ func New(cfg Config, logger *slog.Logger) (*Relay, error) { IdleConnTimeout: 90 * time.Second, ResponseHeaderTimeout: readTimeout, } + targets := make(map[string]*url.URL, len(up.Targets)) + for name, raw := range up.Targets { + targets[name], _ = url.Parse(raw) + } r.upstreams[up.Name] = &upstream{ - Upstream: up, - base: base, + Upstream: up, + base: base, + targets: targets, + slots: make(chan struct{}, cmp.Or(up.MaxConcurrent, DefaultMaxConcurrent)), + maxSeconds: cmp.Or(up.MaxSeconds, DefaultMaxSeconds), client: &http.Client{Transport: upstreamTransport, Timeout: readTimeout, CheckRedirect: func(*http.Request, []*http.Request) error { return errors.New("upstream redirects are not followed") }}, } @@ -224,7 +238,9 @@ func (r *Relay) serve(ctx context.Context) error { // ServeHTTP answers one proxy read from the tunnel. The Host names the // upstream; the path is forwarded under the upstream URL when its route is a -// documented read. One audit line records every decision with the read id +// documented read. For a pprof upstream the first path segment names the +// target process and the query is checked against the configured caps. One +// audit line records every decision with the read id // Perfloop gave it, `/`, which names the // transcript turn that asked; a read no tool call made has none. func (r *Relay) ServeHTTP(w http.ResponseWriter, req *http.Request) { @@ -250,10 +266,35 @@ func (r *Relay) forward(w http.ResponseWriter, req *http.Request) (int, string, if p := req.URL.Path; p == "" || p[0] != '/' || p != path.Clean(p) { return refuse(w, http.StatusBadRequest, "path is not clean"), "path", nil } - name, ok := route(up.Kind, req.URL.Path) + base, rest, query := up.base, req.URL.Path, req.URL.RawQuery + if up.Kind == "pprof" { + // The target and the query are checked here, before any request, in + // the same place as the route: the target map and the caps are the + // customer's control over which process runs a profile and for how + // long. The query forwarded is the one checked, not the raw bytes. + var name string + var ok bool + if name, rest, ok = target(req.URL.Path); !ok || up.targets[name] == nil { + return refuse(w, http.StatusNotFound, "unknown target"), "unknown-target", nil + } + base = up.targets[name] + } + name, ok := route(up.Kind, rest) if !ok { return refuse(w, http.StatusForbidden, "route is not a documented read"), "route", nil } + if up.Kind == "pprof" { + var err error + if query, err = pprofQuery(name, req.URL.RawQuery, up.maxSeconds); err != nil { + return refuse(w, http.StatusBadRequest, err.Error()), "query", nil + } + select { + case up.slots <- struct{}{}: + defer func() { <-up.slots }() + default: + return refuse(w, http.StatusTooManyRequests, "profile reads in flight are at the configured cap"), "busy", nil + } + } // One read is bounded end to end on both sides: the upstream client's // timeout, and the same deadline on the response write, so a tunnel peer // that withholds flow-control credit cannot hold the stream forever. @@ -262,9 +303,9 @@ func (r *Relay) forward(w http.ResponseWriter, req *http.Request) (int, string, if err := http.NewResponseController(w).SetWriteDeadline(time.Now().Add(readTimeout)); err != nil { return refuse(w, http.StatusInternalServerError, "write deadline is not supported"), "path", err } - target := *up.base - target.Path = path.Join(up.base.Path, req.URL.Path) - target.RawQuery = req.URL.RawQuery + target := *base + target.Path = path.Join(base.Path, rest) + target.RawQuery = query out, err := http.NewRequestWithContext(ctx, http.MethodGet, target.String(), http.NoBody) if err != nil { return refuse(w, http.StatusBadRequest, "request could not be built"), "path", err @@ -310,7 +351,11 @@ func (r *Relay) forward(w http.ResponseWriter, req *http.Request) (int, string, return resp.StatusCode, name, nil } +// refuse answers a read the relay will not forward. The error header names +// the relay as the author, so a reader can tell this 404 or 400 from one the +// upstream itself returned. func refuse(w http.ResponseWriter, status int, reason string) int { + w.Header().Set(link.ErrorHeader, link.Refused) http.Error(w, reason, status) return status } diff --git a/relay_test.go b/relay_test.go index c3226a6..b9430b7 100644 --- a/relay_test.go +++ b/relay_test.go @@ -479,3 +479,116 @@ func (l *lockedWriter) Write(p []byte) (int, error) { defer l.mu.Unlock() return l.w.Write(p) } + +// A pprof upstream forwards one documented route to one named process, with +// the checked query only. Every other target, route, method, and query key is +// refused before the process sees a request, and each refusal is marked as +// the relay's own. +func TestPprofReadsReachOnlyNamedTargetsAndDocumentedRoutes(t *testing.T) { + var mu sync.Mutex + var calls []string + provider := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + mu.Lock() + calls = append(calls, req.URL.RequestURI()+" auth="+req.Header.Get("Authorization")) + mu.Unlock() + w.Header().Set("Content-Type", "application/octet-stream") + io.WriteString(w, "pprof-bytes") + })) + defer provider.Close() + term := newTerminator(t, Upstream{Name: "go", Kind: "pprof", MaxSeconds: 5, + Targets: map[string]string{"api-1": provider.URL, "api-2": provider.URL + "/internal"}, + Headers: map[string]string{"Authorization": "Bearer local-token"}}) + term.run(t) + term.waitTunnels(t, tunnels) + + if resp := term.read(t, "go", "/api-1/debug/pprof/profile?seconds=2", "Authorization", "Bearer from-perfloop"); resp.status != http.StatusOK || resp.body != "pprof-bytes" || resp.header.Get(link.ErrorHeader) != "" { + t.Fatalf("profile: %+v", resp) + } + if resp := term.read(t, "go", "/api-2/debug/pprof/heap"); resp.status != http.StatusOK { + t.Fatalf("heap: %+v", resp) + } + if resp := term.read(t, "go", "/api-1/debug/pprof/goroutine?seconds=1"); resp.status != http.StatusOK { + t.Fatalf("goroutine delta: %+v", resp) + } + want := []string{ + "/debug/pprof/profile?seconds=2 auth=Bearer local-token", + "/internal/debug/pprof/heap auth=Bearer local-token", + "/debug/pprof/goroutine?seconds=1 auth=Bearer local-token", + } + mu.Lock() + got := append([]string(nil), calls...) + mu.Unlock() + if strings.Join(got, "\n") != strings.Join(want, "\n") { + t.Fatalf("process saw %q, want %q", got, want) + } + for _, refused := range []struct { + path string + status int + }{ + {"/api-3/debug/pprof/heap", http.StatusNotFound}, + {"/debug/pprof/heap", http.StatusNotFound}, + {"/api-1", http.StatusNotFound}, + {"/api-1/debug/pprof/cmdline", http.StatusForbidden}, + {"/api-1/debug/pprof/symbol", http.StatusForbidden}, + {"/api-1/debug/pprof/trace?seconds=1", http.StatusForbidden}, + {"/api-1/debug/pprof/", http.StatusBadRequest}, + {"/api-1/debug/pprof", http.StatusForbidden}, + {"/api-1/metrics", http.StatusForbidden}, + {"/api-1/debug/pprof/profile", http.StatusBadRequest}, + {"/api-1/debug/pprof/profile?seconds=6", http.StatusBadRequest}, + {"/api-1/debug/pprof/profile?seconds=2&debug=1", http.StatusBadRequest}, + {"/api-1/debug/pprof/goroutine?debug=2", http.StatusBadRequest}, + {"/api-1/debug/pprof/heap?gc=1", http.StatusBadRequest}, + } { + if resp := term.read(t, "go", refused.path); resp.status != refused.status || resp.header.Get(link.ErrorHeader) != link.Refused { + t.Fatalf("%s: %d %q, want %d marked as the relay's own refusal", refused.path, resp.status, resp.body, refused.status) + } + } + mu.Lock() + n := len(calls) + mu.Unlock() + if n != len(want) { + t.Fatalf("process saw %d requests after refusals, want %d", n, len(want)) + } +} + +// The in-flight cap is per upstream and refuses at once: a second profile +// while one runs gets a marked 429 and the process never sees it; the first +// completes. +func TestPprofConcurrencyCapRefusesAtOnce(t *testing.T) { + started := make(chan struct{}, 2) + release := make(chan struct{}) + // Free the blocked provider on every exit, so a failed assertion cannot + // leave the first read hanging and the test binary with it. + releaseAll := sync.OnceFunc(func() { close(release) }) + defer releaseAll() + var calls atomic.Int64 + provider := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + calls.Add(1) + started <- struct{}{} + <-release + io.WriteString(w, "pprof-bytes") + })) + defer provider.Close() + term := newTerminator(t, Upstream{Name: "go", Kind: "pprof", Targets: map[string]string{"api-1": provider.URL, "api-2": provider.URL}}) + term.run(t) + term.waitTunnels(t, tunnels) + + first := make(chan reply, 1) + go func() { first <- term.read(t, "go", "/api-1/debug/pprof/profile?seconds=1") }() + select { + case <-started: + case <-time.After(5 * time.Second): + t.Fatal("first profile never reached the process") + } + if resp := term.read(t, "go", "/api-2/debug/pprof/heap"); resp.status != http.StatusTooManyRequests || resp.header.Get(link.ErrorHeader) != link.Refused || calls.Load() != 1 { + t.Fatalf("second read while one is in flight: %+v, process calls %d", resp, calls.Load()) + } + releaseAll() + if resp := <-first; resp.status != http.StatusOK || resp.body != "pprof-bytes" { + t.Fatalf("first profile: %+v", resp) + } + if resp := term.read(t, "go", "/api-2/debug/pprof/heap"); resp.status != http.StatusOK || calls.Load() != 2 { + t.Fatalf("read after the slot is free: %+v, process calls %d", resp, calls.Load()) + } +} diff --git a/routes.go b/routes.go index 0d2a561..1b73174 100644 --- a/routes.go +++ b/routes.go @@ -1,10 +1,14 @@ package relay import ( + "cmp" + "errors" "fmt" "io" "maps" + "net/url" "slices" + "strconv" "strings" "github.com/perfloop/relay/link" @@ -18,6 +22,8 @@ func base(kind string) (string, bool) { return "/api/v1/", true case "loki": return "/loki/api/v1/", true + case "pprof": + return "/debug/pprof/", true } return "", false } @@ -35,16 +41,61 @@ func reads(kind string) []string { return []string{"query", "query_range", "labels", labelValues, "series", "metadata"} case "loki": return []string{"query", "query_range", "labels", labelValues} + case "pprof": + return []string{"profile", "heap", "allocs", "goroutine", "mutex", "block"} } return nil } +// target splits the first segment off a clean absolute path: the pprof +// target name Perfloop put in front of the route. `/api-1/debug/pprof/heap` +// gives `api-1` and `/debug/pprof/heap`. A path with no second segment has +// no route and is refused. +func target(path string) (name, rest string, ok bool) { + name, rest, ok = strings.Cut(path[1:], "/") + return name, "/" + rest, ok +} + +// pprofQuery checks the query of one admitted pprof route and returns the +// exact query to forward. `seconds` is the only key: it is the profile +// length for `profile` and a delta window for the other routes, one +// canonical decimal from 1 to maxSeconds. `profile` must carry it, because +// net/http/pprof runs thirty seconds without it and that may exceed the cap. +// Every other key, including `debug` (text output) and `gc`, is refused so +// nothing but the documented pprof bytes leaves the process. +func pprofQuery(route, rawQuery string, maxSeconds int) (string, error) { + values, err := url.ParseQuery(rawQuery) + if err != nil { + return "", errors.New("query is not well formed") + } + for key := range values { + if key != "seconds" { + return "", fmt.Errorf("query key %q is not permitted", key) + } + } + seconds := values["seconds"] + switch { + case len(seconds) == 0 && route == "profile": + return "", errors.New("profile requires seconds") + case len(seconds) == 0: + return "", nil + case len(seconds) > 1: + return "", errors.New("seconds is repeated") + } + n, err := strconv.Atoi(seconds[0]) + if err != nil || strconv.Itoa(n) != seconds[0] || n < 1 || n > maxSeconds { + return "", fmt.Errorf("seconds must be an integer from 1 to %d", maxSeconds) + } + return "seconds=" + seconds[0], nil +} + // route returns the read route that path names for kind. The path is the // provider API base and then exactly one documented read route, nothing // before and nothing else after: the customer's own prefix lives in the // upstream URL, so a Loki path never reads as a Prometheus one on a shared // gateway, and admin and write routes are refused before any upstream -// request. Callers pass a clean absolute path. +// request. Callers pass a clean absolute path; for a pprof upstream they pass +// the path after the target segment. func route(kind, path string) (string, bool) { base, ok := base(kind) if !ok { @@ -85,14 +136,28 @@ func Routes(w io.Writer, cfg Config) error { } fmt.Fprintf(&out, "upstream %s\n", up.Name) fmt.Fprintf(&out, " kind %s\n", up.Kind) - fmt.Fprintf(&out, " url %s\n", up.URL) + if up.Kind == "pprof" { + for _, name := range slices.Sorted(maps.Keys(up.Targets)) { + fmt.Fprintf(&out, " target %s %s\n", name, up.Targets[name]) + } + fmt.Fprintf(&out, " caps seconds at most %d; %d profile read(s) in flight per relay process\n", cmp.Or(up.MaxSeconds, DefaultMaxSeconds), cmp.Or(up.MaxConcurrent, DefaultMaxConcurrent)) + } else { + fmt.Fprintf(&out, " url %s\n", up.URL) + } fmt.Fprintf(&out, " passes %s from Perfloop; every other request header is dropped\n", strings.Join(link.ForwardedHeaders, ", ")) headers := slices.Sorted(maps.Keys(up.Headers)) if len(headers) > 0 { fmt.Fprintf(&out, " headers %s (values not shown)\n", strings.Join(headers, ", ")) } - out.WriteString(" forwards GET, with the query string unchanged, to:\n") base, _ := base(up.Kind) + if up.Kind == "pprof" { + out.WriteString(" forwards GET, with seconds as the only query key, to each target's:\n") + for _, read := range reads(up.Kind) { + fmt.Fprintf(&out, " %s%s\n", base, read) + } + continue + } + out.WriteString(" forwards GET, with the query string unchanged, to:\n") for _, read := range reads(up.Kind) { fmt.Fprintf(&out, " %s%s%s\n", up.URL, base, read) } diff --git a/routes_test.go b/routes_test.go index 8d13caf..bafbcb2 100644 --- a/routes_test.go +++ b/routes_test.go @@ -3,6 +3,7 @@ package relay import ( "errors" "os" + "strconv" "strings" "testing" ) @@ -16,11 +17,15 @@ func FuzzRouteNeverAdmitsWritesOrAdmin(f *testing.F) { "/api/v1/query", "/select/0/prometheus/api/v1/query_range", "/api/v1/label/x/values", "/admin-cluster/api/v1/labels", "/api/v1/admin/tsdb/delete_series", "/api/v1/admin/tsdb/snapshot/api/v1/query", "/api/v1/write/api/v1/labels", "/loki/api/v1/push", "/loki/api/v1/push/loki/api/v1/query", "/api/v1/query/../write", "/loki/api/v1/query", + "/debug/pprof/profile", "/debug/pprof/heap", "/debug/pprof/cmdline", "/debug/pprof/symbol", "/debug/pprof/trace", + "/debug/pprof/threadcreate", "/debug/pprof/", "/api-1/debug/pprof/heap", "/debug/pprof/heap/debug/pprof/profile", } { f.Add("prometheus", seed) f.Add("loki", seed) + f.Add("pprof", seed) } reads := map[string]bool{"query": true, "query_range": true, "labels": true, "series": true, "metadata": true} + pprofReads := map[string]bool{"profile": true, "heap": true, "allocs": true, "goroutine": true, "mutex": true, "block": true} f.Fuzz(func(t *testing.T, kind, path string) { name, ok := route(kind, path) if !ok { @@ -30,10 +35,19 @@ func FuzzRouteNeverAdmitsWritesOrAdmin(f *testing.F) { if kind == "loki" { base = "/loki/api/v1/" } + if kind == "pprof" { + base = "/debug/pprof/" + } rest, atStart := strings.CutPrefix(path, base) if !atStart { t.Fatalf("admitted %s %s as %s", kind, path, name) } + if kind == "pprof" { + if !pprofReads[rest] || rest != name { + t.Fatalf("admitted %s %s as %s", kind, path, name) + } + return + } if label, ok := strings.CutPrefix(rest, "label/"); ok { labelName, isLabel := strings.CutSuffix(label, "/values") if !isLabel { @@ -62,6 +76,14 @@ func TestRouteAdmitsExactlyBaseAndOneReadRoute(t *testing.T) { {"loki", "/api/v1/query", ""}, {"loki", "/loki/api/v1/series", ""}, {"prometheus", "/api/v1/admin/tsdb/snapshot", ""}, + {"pprof", "/debug/pprof/profile", "profile"}, + {"pprof", "/debug/pprof/block", "block"}, + {"pprof", "/debug/pprof/cmdline", ""}, + {"pprof", "/debug/pprof/symbol", ""}, + {"pprof", "/debug/pprof/trace", ""}, + {"pprof", "/debug/pprof/", ""}, + {"pprof", "/api/v1/query", ""}, + {"prometheus", "/debug/pprof/heap", ""}, } { if got, ok := route(tc.kind, tc.path); got != tc.want || ok != (tc.want != "") { t.Errorf("route(%s, %s) = %q, %v; want %q", tc.kind, tc.path, got, ok, tc.want) @@ -117,3 +139,51 @@ func TestRoutesPrintsTheTableAndNoSecret(t *testing.T) { type failingWriter struct{} func (failingWriter) Write([]byte) (int, error) { return 0, errors.New("disk full") } + +// A pprof query carries `seconds` and nothing else. `profile` must name its +// length so the upstream default cannot exceed the cap; the other routes may +// take a snapshot without it. The forwarded query is the checked one. +func TestPprofQueryAdmitsOnlyCappedSeconds(t *testing.T) { + for _, tc := range []struct { + route, query, want string + ok bool + }{ + {"profile", "seconds=5", "seconds=5", true}, + {"profile", "seconds=30", "seconds=30", true}, + {"heap", "", "", true}, + {"heap", "seconds=3", "seconds=3", true}, + {"profile", "", "", false}, + {"profile", "seconds=31", "", false}, + {"profile", "seconds=0", "", false}, + {"profile", "seconds=-1", "", false}, + {"profile", "seconds=05", "", false}, + {"profile", "seconds=+5", "", false}, + {"profile", "seconds=5&seconds=5", "", false}, + {"profile", "seconds=5&debug=1", "", false}, + {"goroutine", "debug=2", "", false}, + {"heap", "gc=1", "", false}, + {"heap", "seconds=%zz", "", false}, + } { + got, err := pprofQuery(tc.route, tc.query, 30) + if (err == nil) != tc.ok || got != tc.want { + t.Errorf("pprofQuery(%s, %q) = %q, %v; want %q ok=%v", tc.route, tc.query, got, err, tc.want, tc.ok) + } + } +} + +// The seconds check never admits a value outside 1..max, whatever the bytes. +func FuzzPprofQuerySecondsStayWithinCap(f *testing.F) { + for _, seed := range []string{"seconds=5", "seconds=99999999999999999999", "seconds=1e3", "seconds=5&x=1", "seconds=5;seconds=6", "seconds=%35"} { + f.Add(seed) + } + f.Fuzz(func(t *testing.T, raw string) { + query, err := pprofQuery("profile", raw, 30) + if err != nil { + return + } + n, convErr := strconv.Atoi(strings.TrimPrefix(query, "seconds=")) + if !strings.HasPrefix(query, "seconds=") || convErr != nil || n < 1 || n > 30 { + t.Fatalf("admitted %q as %q", raw, query) + } + }) +}