From 69920f248c92f3f1f356f9cfd8be672897ae8ee4 Mon Sep 17 00:00:00 2001 From: Thomas Henry Thirlwall Date: Tue, 23 Jun 2026 22:00:43 -0500 Subject: [PATCH 1/2] feat(profiling): add WithProfiling continuous CPU/heap profiling Add an opt-in background profiler that, on an interval (default 60s, CPU window ~30s capped to interval/2), captures CPU and heap pprof profiles and POSTs each as a separate request to /api/profiles/ingest, reusing the connection-string token and host. The ingest URL is derived from the existing report URL (scheme+host + /api/profiles/ingest); Bearer auth, serverName and appVersion reuse the existing report plumbing. Profiling never affects the host app: capture and upload failures are isolated and silent unless WithDebug is set, and a bad URL disables profiling without failing Init. Verified against the backend contract (tracewayapp/traceway /api/profiles/ingest): path, Bearer auth, service/serverName/appVersion query params, and raw already-gzipped pprof body with no extra Content-Encoding. Core package only; HTTP middleware pass-through options are a follow-up. Co-Authored-By: Claude Opus 4.8 --- docs/profiling.md | 78 +++++++++++++ profiling.go | 191 ++++++++++++++++++++++++++++++++ profiling_test.go | 272 ++++++++++++++++++++++++++++++++++++++++++++++ traceway.go | 11 ++ 4 files changed, 552 insertions(+) create mode 100644 docs/profiling.md create mode 100644 profiling.go create mode 100644 profiling_test.go diff --git a/docs/profiling.md b/docs/profiling.md new file mode 100644 index 0000000..a8ebe54 --- /dev/null +++ b/docs/profiling.md @@ -0,0 +1,78 @@ +# Continuous profiling + +The SDK can periodically capture Go runtime profiles and ship them to Traceway, +where they power flame graphs and allocation views alongside your traces and +metrics. + +Profiling is **opt-in** and runs in a single background goroutine. When enabled, +each cycle captures: + +- a **CPU profile** (`runtime/pprof` `StartCPUProfile` / `StopCPUProfile`) over a + ~30s window, and +- a **heap profile** (`pprof.Lookup("heap")`), + +then POSTs each one as a separate request to `/api/profiles/ingest`. + +## Enabling + +```go +err := traceway.Init( + connectionString, + traceway.WithProfiling("checkout-api"), +) +``` + +`WithProfiling(serviceName)` enables profiling. `serviceName` is the application +name the profiles are grouped under in the dashboard. If you pass an empty +string, it defaults to the running binary's name (`filepath.Base(os.Args[0])`). + +### Interval + +```go +traceway.WithProfiling("checkout-api"), +traceway.WithProfilingInterval(2 * time.Minute), +``` + +`WithProfilingInterval` controls how often a cycle runs. The default is **60s**. +The CPU window is **30s**, automatically capped to `interval / 2` so a short +interval can never be overrun by an in-flight CPU capture. The first profile is +captured one interval after `Init` (not at startup). + +## What gets sent + +Each profile is uploaded as its own request, reusing the token and server from +the connection string: + +| Part | Value | +| --- | --- | +| Method / path | `POST :///api/profiles/ingest` | +| `Authorization` | `Bearer ` | +| Query `service` | the app name from `WithProfiling` | +| Query `serverName` | the host (same value as `WithServerName`, defaults to the hostname) | +| Query `appVersion` | the value from `WithVersion` | +| `Content-Type` | `application/octet-stream` | +| Body | the raw pprof bytes | + +The ingest URL is derived from the connection string's report URL by keeping its +scheme and host and replacing the path with `/api/profiles/ingest`. + +pprof output is already gzip-compressed, so the body is sent **as-is** — no extra +`Content-Encoding: gzip`. + +## Notes + +- A heap snapshot is taken **without** forcing a `runtime.GC()` first, to avoid + injecting GC latency spikes into the host application. +- Profiling never affects your application's behavior on failure. A capture or + upload error skips that one profile for the cycle; the other profile is still + attempted. Errors are silent unless `WithDebug(true)` is set. +- If another part of the process already holds the CPU profiler, + `StartCPUProfile` fails for that cycle; the heap profile is unaffected. + +## Middleware + +`WithProfiling` lives on the core `traceway` package. Apps that initialize via +`traceway.Init` directly (for example, worker/task processes using `MeasureTask`) +can enable it today. A pass-through option for the HTTP middleware wrappers +(`tracewayhttp`, `tracewaygin`, `tracewayfiber`, `tracewaychi`, +`tracewayfasthttp`) is a follow-up. diff --git a/profiling.go b/profiling.go new file mode 100644 index 0000000..c5e5960 --- /dev/null +++ b/profiling.go @@ -0,0 +1,191 @@ +package traceway + +import ( + "bytes" + "fmt" + "io" + "log" + "net/http" + "net/url" + "os" + "path/filepath" + "runtime/pprof" + "time" +) + +const ( + profileIngestPath = "/api/profiles/ingest" + defaultProfilingInterval = 60 * time.Second + defaultCPUProfileWindow = 30 * time.Second + profileUploadTimeout = 30 * time.Second +) + +func WithProfiling(serviceName string) func(*TracewayOptions) { + return func(s *TracewayOptions) { + s.profilingEnabled = true + if serviceName == "" { + serviceName = filepath.Base(os.Args[0]) + } + s.profilingService = serviceName + } +} + +func WithProfilingInterval(d time.Duration) func(*TracewayOptions) { + return func(s *TracewayOptions) { + s.profilingInterval = d + } +} + +func profileIngestURL(reportURL string) (string, error) { + u, err := url.Parse(reportURL) + if err != nil { + return "", err + } + if u.Scheme == "" || u.Host == "" { + return "", fmt.Errorf("traceway: cannot derive profile ingest URL from %q", reportURL) + } + return u.Scheme + "://" + u.Host + profileIngestPath, nil +} + +func captureCPUProfile(d time.Duration) ([]byte, error) { + var buf bytes.Buffer + if err := pprof.StartCPUProfile(&buf); err != nil { + return nil, err + } + time.Sleep(d) + pprof.StopCPUProfile() + return buf.Bytes(), nil +} + +func captureHeapProfile() ([]byte, error) { + prof := pprof.Lookup("heap") + if prof == nil { + return nil, fmt.Errorf("traceway: heap profile unavailable") + } + var buf bytes.Buffer + if err := prof.WriteTo(&buf, 0); err != nil { + return nil, err + } + return buf.Bytes(), nil +} + +type profiler struct { + url string + token string + service string + serverName string + appVersion string + interval time.Duration + cpuWindow time.Duration + client *http.Client + debug bool +} + +func (p *profiler) uploadProfile(body []byte) error { + req, err := http.NewRequest(http.MethodPost, p.url, bytes.NewReader(body)) + if err != nil { + return err + } + + q := req.URL.Query() + q.Set("service", p.service) + q.Set("serverName", p.serverName) + q.Set("appVersion", p.appVersion) + req.URL.RawQuery = q.Encode() + + req.Header.Set("Authorization", "Bearer "+p.token) + req.Header.Set("Content-Type", "application/octet-stream") + + client := p.client + if client == nil { + client = http.DefaultClient + } + + resp, err := client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + _, _ = io.Copy(io.Discard, resp.Body) + + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("traceway: profile ingest returned status %d", resp.StatusCode) + } + return nil +} + +func (p *profiler) collectOnce() { + window := p.cpuWindow + if window <= 0 { + window = defaultCPUProfileWindow + } + + if cpu, err := captureCPUProfile(window); err != nil { + p.logError("cpu profile capture failed", err) + } else if err := p.uploadProfile(cpu); err != nil { + p.logError("cpu profile upload failed", err) + } + + if heap, err := captureHeapProfile(); err != nil { + p.logError("heap profile capture failed", err) + } else if err := p.uploadProfile(heap); err != nil { + p.logError("heap profile upload failed", err) + } +} + +func (p *profiler) safeCollectOnce() { + defer func() { + if r := recover(); r != nil { + p.logError("profiling cycle panicked", fmt.Errorf("%v", r)) + } + }() + p.collectOnce() +} + +func (p *profiler) run() { + ticker := time.NewTicker(p.interval) + defer ticker.Stop() + + for range ticker.C { + p.safeCollectOnce() + } +} + +func (p *profiler) logError(msg string, err error) { + if p.debug { + log.Printf("Traceway: %s: %v", msg, err) + } +} + +func startProfiler(reportURL, token string, opts *TracewayOptions) error { + ingestURL, err := profileIngestURL(reportURL) + if err != nil { + return err + } + + interval := opts.profilingInterval + if interval <= 0 { + interval = defaultProfilingInterval + } + + cpuWindow := defaultCPUProfileWindow + if half := interval / 2; half < cpuWindow { + cpuWindow = half + } + + p := &profiler{ + url: ingestURL, + token: token, + service: opts.profilingService, + serverName: opts.serverName, + appVersion: opts.version, + interval: interval, + cpuWindow: cpuWindow, + client: &http.Client{Timeout: profileUploadTimeout}, + debug: opts.debug, + } + + go p.run() + + return nil +} diff --git a/profiling_test.go b/profiling_test.go new file mode 100644 index 0000000..f95b217 --- /dev/null +++ b/profiling_test.go @@ -0,0 +1,272 @@ +package traceway + +import ( + "bytes" + "compress/gzip" + "io" + "net/http" + "net/http/httptest" + "net/url" + "os" + "path/filepath" + "sync" + "testing" + "time" +) + +func assertCompressedPprof(t *testing.T, b []byte) { + t.Helper() + if len(b) == 0 { + t.Fatal("profile is empty") + } + r, err := gzip.NewReader(bytes.NewReader(b)) + if err != nil { + t.Fatalf("profile is not a valid gzip stream (pprof must be gzip-compressed): %v", err) + } + defer r.Close() + decoded, err := io.ReadAll(r) + if err != nil { + t.Fatalf("profile gzip stream failed to fully inflate (truncated/corrupt?): %v", err) + } + if len(decoded) == 0 { + t.Fatal("profile decompressed to zero bytes") + } +} + +type recordedRequest struct { + method string + path string + authorization string + contentType string + contentEncoding string + query url.Values + body []byte +} + +func TestProfileIngestURL(t *testing.T) { + cases := []struct { + name string + reportURL string + want string + wantErr bool + }{ + {"report path", "http://localhost:19876/api/report", "http://localhost:19876/api/profiles/ingest", false}, + {"https host", "https://ingest.tracewayapp.com/api/report", "https://ingest.tracewayapp.com/api/profiles/ingest", false}, + {"unrelated path and port", "http://127.0.0.1:8080/noop", "http://127.0.0.1:8080/api/profiles/ingest", false}, + {"empty", "", "", true}, + {"no scheme or host", "/api/report", "", true}, + } + + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + got, err := profileIngestURL(c.reportURL) + if c.wantErr { + if err == nil { + t.Fatalf("expected error for %q, got %q", c.reportURL, got) + } + return + } + if err != nil { + t.Fatalf("unexpected error for %q: %v", c.reportURL, err) + } + if got != c.want { + t.Fatalf("profileIngestURL(%q) = %q, want %q", c.reportURL, got, c.want) + } + }) + } +} + +func TestCaptureHeapProfile(t *testing.T) { + b, err := captureHeapProfile() + if err != nil { + t.Fatalf("captureHeapProfile error: %v", err) + } + assertCompressedPprof(t, b) +} + +func TestCaptureCPUProfile(t *testing.T) { + b, err := captureCPUProfile(80 * time.Millisecond) + if err != nil { + t.Fatalf("captureCPUProfile error: %v", err) + } + assertCompressedPprof(t, b) + + if _, err := captureCPUProfile(80 * time.Millisecond); err != nil { + t.Fatalf("second captureCPUProfile error (global state not reset?): %v", err) + } +} + +func newRecordingServer(t *testing.T, status *int, sink func(recordedRequest)) *httptest.Server { + t.Helper() + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + sink(recordedRequest{ + method: r.Method, + path: r.URL.Path, + authorization: r.Header.Get("Authorization"), + contentType: r.Header.Get("Content-Type"), + contentEncoding: r.Header.Get("Content-Encoding"), + query: r.URL.Query(), + body: body, + }) + code := http.StatusOK + if status != nil { + code = *status + } + w.WriteHeader(code) + _, _ = w.Write([]byte("{}")) + })) +} + +func TestUploadProfile(t *testing.T) { + var ( + mu sync.Mutex + got recordedRequest + status = http.StatusOK + ) + srv := newRecordingServer(t, &status, func(r recordedRequest) { + mu.Lock() + got = r + mu.Unlock() + }) + defer srv.Close() + + p := &profiler{ + url: srv.URL + profileIngestPath, + token: "testtoken", + service: "checkout-api", + serverName: "host-1", + appVersion: "1.2.3", + client: srv.Client(), + } + + payload := []byte{0x1f, 0x8b, 0x08, 0x00, 0x01, 0x02, 0x03} + if err := p.uploadProfile(payload); err != nil { + t.Fatalf("uploadProfile returned error on 200: %v", err) + } + + mu.Lock() + defer mu.Unlock() + + if got.method != http.MethodPost { + t.Errorf("method = %q, want POST", got.method) + } + if got.path != profileIngestPath { + t.Errorf("path = %q, want %q", got.path, profileIngestPath) + } + if got.authorization != "Bearer testtoken" { + t.Errorf("Authorization = %q, want %q", got.authorization, "Bearer testtoken") + } + if v := got.query.Get("service"); v != "checkout-api" { + t.Errorf("query service = %q, want checkout-api", v) + } + if v := got.query.Get("serverName"); v != "host-1" { + t.Errorf("query serverName = %q, want host-1", v) + } + if v := got.query.Get("appVersion"); v != "1.2.3" { + t.Errorf("query appVersion = %q, want 1.2.3", v) + } + if !bytes.Equal(got.body, payload) { + t.Errorf("body = %x, want %x (must be raw pprof, no extra gzip)", got.body, payload) + } + if got.contentEncoding == "gzip" { + t.Error("Content-Encoding must not be gzip; pprof bytes are already compressed") + } + if got.contentType != "application/octet-stream" { + t.Errorf("Content-Type = %q, want application/octet-stream", got.contentType) + } +} + +func TestUploadProfileNon200ReturnsError(t *testing.T) { + status := http.StatusInternalServerError + srv := newRecordingServer(t, &status, func(recordedRequest) {}) + defer srv.Close() + + p := &profiler{ + url: srv.URL + profileIngestPath, + token: "testtoken", + service: "svc", + serverName: "host", + appVersion: "v", + client: srv.Client(), + } + if err := p.uploadProfile([]byte{0x1f, 0x8b, 0x00}); err == nil { + t.Fatal("expected error on non-200 response, got nil") + } +} + +func TestProfilingOptions(t *testing.T) { + def := NewTracewayOptions() + if def.profilingEnabled { + t.Error("profiling should be disabled by default") + } + if def.profilingInterval != 60*time.Second { + t.Errorf("default profilingInterval = %v, want 60s", def.profilingInterval) + } + + enabled := NewTracewayOptions(WithProfiling("checkout-api")) + if !enabled.profilingEnabled { + t.Error("WithProfiling should enable profiling") + } + if enabled.profilingService != "checkout-api" { + t.Errorf("profilingService = %q, want checkout-api", enabled.profilingService) + } + + defaulted := NewTracewayOptions(WithProfiling("")) + wantService := filepath.Base(os.Args[0]) + if defaulted.profilingService != wantService { + t.Errorf("empty service should default to %q, got %q", wantService, defaulted.profilingService) + } + + custom := NewTracewayOptions(WithProfiling("svc"), WithProfilingInterval(15*time.Second)) + if custom.profilingInterval != 15*time.Second { + t.Errorf("profilingInterval = %v, want 15s", custom.profilingInterval) + } +} + +func TestCollectOnceSendsCPUAndHeap(t *testing.T) { + var ( + mu sync.Mutex + reqs []recordedRequest + ) + srv := newRecordingServer(t, nil, func(r recordedRequest) { + mu.Lock() + reqs = append(reqs, r) + mu.Unlock() + }) + defer srv.Close() + + p := &profiler{ + url: srv.URL + profileIngestPath, + token: "testtoken", + service: "svc", + serverName: "host", + appVersion: "1.0.0", + cpuWindow: 80 * time.Millisecond, + client: srv.Client(), + } + + p.collectOnce() + + mu.Lock() + defer mu.Unlock() + + if len(reqs) != 2 { + t.Fatalf("collectOnce sent %d requests, want 2 (cpu + heap)", len(reqs)) + } + for i, r := range reqs { + if r.method != http.MethodPost { + t.Errorf("req %d method = %q, want POST", i, r.method) + } + if r.path != profileIngestPath { + t.Errorf("req %d path = %q, want %q", i, r.path, profileIngestPath) + } + if r.authorization != "Bearer testtoken" { + t.Errorf("req %d Authorization = %q, want Bearer testtoken", i, r.authorization) + } + if r.query.Get("service") != "svc" || r.query.Get("serverName") != "host" || r.query.Get("appVersion") != "1.0.0" { + t.Errorf("req %d query = %v, want service=svc serverName=host appVersion=1.0.0", i, r.query) + } + assertCompressedPprof(t, r.body) + } +} diff --git a/traceway.go b/traceway.go index 39e2be8..e9b8bdc 100644 --- a/traceway.go +++ b/traceway.go @@ -627,6 +627,9 @@ type TracewayOptions struct { serverName string sampleRate float64 errorSampleRate float64 + profilingEnabled bool + profilingService string + profilingInterval time.Duration } func NewTracewayOptions(options ...func(*TracewayOptions)) *TracewayOptions { @@ -639,6 +642,7 @@ func NewTracewayOptions(options ...func(*TracewayOptions)) *TracewayOptions { serverName: getHostname(), sampleRate: 1, errorSampleRate: 1, + profilingInterval: defaultProfilingInterval, } for _, o := range options { o(svr) @@ -715,6 +719,13 @@ func Init(connectionString string, options ...func(*TracewayOptions)) error { tracewayOptions.sampleRate, tracewayOptions.errorSampleRate, ) + + if tracewayOptions.profilingEnabled { + if err := startProfiler(apiUrl, token, tracewayOptions); err != nil && tracewayOptions.debug { + log.Printf("Traceway: profiling disabled: %v", err) + } + } + return nil } From a8749852f70047ca7eddd1cc8a39375651f00346 Mon Sep 17 00:00:00 2001 From: Thomas Henry Thirlwall Date: Tue, 23 Jun 2026 22:15:12 -0500 Subject: [PATCH 2/2] fix(init): reject malformed connection strings instead of panicking Init split the connection string on "@" and unconditionally indexed connParts[1], panicking with "index out of range" when the string had no "@". Use SplitN with a length guard so a malformed connection string returns an error, and so a "@" inside the URL (e.g. userinfo) no longer silently truncates the endpoint. Co-Authored-By: Claude Opus 4.8 --- traceway.go | 5 ++++- traceway_test.go | 18 ++++++++++++++++++ 2 files changed, 22 insertions(+), 1 deletion(-) create mode 100644 traceway_test.go diff --git a/traceway.go b/traceway.go index e9b8bdc..8015be7 100644 --- a/traceway.go +++ b/traceway.go @@ -699,7 +699,10 @@ func Init(connectionString string, options ...func(*TracewayOptions)) error { if collectionFrameStore != nil { return fmt.Errorf("Second Traceway initialization detected") } - connParts := strings.Split(connectionString, "@") + connParts := strings.SplitN(connectionString, "@", 2) + if len(connParts) != 2 { + return fmt.Errorf("traceway: invalid connection string, expected \"@\"") + } token := connParts[0] apiUrl := connParts[1] diff --git a/traceway_test.go b/traceway_test.go new file mode 100644 index 0000000..cbce2c3 --- /dev/null +++ b/traceway_test.go @@ -0,0 +1,18 @@ +package traceway + +import "testing" + +func TestInitInvalidConnectionString(t *testing.T) { + prev := collectionFrameStore + defer func() { collectionFrameStore = prev }() + + for _, conn := range []string{"missing-at-separator", ""} { + collectionFrameStore = nil + if err := Init(conn); err == nil { + t.Errorf("Init(%q) = nil error, want error for malformed connection string", conn) + } + if collectionFrameStore != nil { + t.Errorf("Init(%q) initialized the store on invalid input", conn) + } + } +}