From fb42c716db0effacca3456e225c7be94a0828961 Mon Sep 17 00:00:00 2001 From: Parnika Mandewalkar Date: Fri, 28 Aug 2026 18:53:13 +0000 Subject: [PATCH 1/3] benchmarking: Rename boomer-glutton to boomer-worker --- benchmarking/locust/Dockerfile | 6 +++--- benchmarking/locust/build_and_push.sh | 2 +- benchmarking/locust/common/boomer_config.py | 4 ++-- benchmarking/locust/manifests/locust.yaml | 10 +++++----- benchmarking/locust/runner.py | 6 +++--- benchmarking/locust/tests/durdir.py | 4 ++-- benchmarking/locust/tests/glutton.py | 4 ++-- benchmarking/observability.md | 2 +- .../{boomer-glutton => boomer-worker}/main.go | 2 +- internal/benchmarking/boomer/trace/trace.go | 2 +- 10 files changed, 21 insertions(+), 21 deletions(-) rename cmd/benchmarking/{boomer-glutton => boomer-worker}/main.go (98%) diff --git a/benchmarking/locust/Dockerfile b/benchmarking/locust/Dockerfile index 5d1c4b4246..6de127ec1d 100644 --- a/benchmarking/locust/Dockerfile +++ b/benchmarking/locust/Dockerfile @@ -21,7 +21,7 @@ WORKDIR /app COPY benchmarking/locust/requirements.txt . RUN pip install --no-cache-dir --target=/app/deps -r requirements.txt -# Stage 2: Go build of the boomer-glutton worker. Lives in the same image so +# Stage 2: Go build of the boomer-worker worker. Lives in the same image so # runner.py can spawn it as a subprocess during headless runs of glutton. # CGO_ENABLED=0 keeps us on boomer's pure-Go ZMQ (gomq), no libzmq needed. FROM golang:1.26-bookworm AS goboomer @@ -32,14 +32,14 @@ COPY vendor ./vendor COPY cmd ./cmd COPY internal ./internal COPY pkg ./pkg -RUN CGO_ENABLED=0 GOFLAGS=-mod=vendor go build -o /out/boomer-glutton ./cmd/benchmarking/boomer-glutton +RUN CGO_ENABLED=0 GOFLAGS=-mod=vendor go build -o /out/boomer-worker ./cmd/benchmarking/boomer-worker # Stage 3: Distroless runner. Carries the locust install AND the boomer # binary; the manifest decides which to launch via the `args` field. FROM gcr.io/distroless/python3-debian12 WORKDIR /app COPY --from=pydeps /app/deps /app/deps -COPY --from=goboomer /out/boomer-glutton /app/boomer-glutton +COPY --from=goboomer /out/boomer-worker /app/boomer-worker COPY benchmarking/locust/common/ /app/common/ COPY benchmarking/locust/shapes/ /app/shapes/ COPY benchmarking/locust/tests/ /app/tests/ diff --git a/benchmarking/locust/build_and_push.sh b/benchmarking/locust/build_and_push.sh index 393254a775..b4b4c5b68f 100755 --- a/benchmarking/locust/build_and_push.sh +++ b/benchmarking/locust/build_and_push.sh @@ -36,7 +36,7 @@ PLATFORM="${LOCUST_IMAGE_PLATFORM:-linux/amd64}" echo "Building Docker image: $IMAGE (platform: $PLATFORM)" # Build context is the monorepo root because the Dockerfile compiles the -# boomer-glutton Go binary alongside the Python install (see Dockerfile). +# boomer-worker Go binary alongside the Python install (see Dockerfile). docker build --platform "$PLATFORM" -t "$IMAGE" -f benchmarking/locust/Dockerfile . echo "Pushing Docker image..." diff --git a/benchmarking/locust/common/boomer_config.py b/benchmarking/locust/common/boomer_config.py index 9494a085d5..d11fb3f978 100644 --- a/benchmarking/locust/common/boomer_config.py +++ b/benchmarking/locust/common/boomer_config.py @@ -26,7 +26,7 @@ * init_boomer_config(): ensures the owning init_*() hooks have run, then serves the current parsed values at /boomer-config on the master. * build_config_json(): parses an argv list and returns the JSON payload - that runner.py hands to boomer-glutton via --config-json in headless + that runner.py hands to boomer-worker via --config-json in headless mode (no web UI to fetch from). * serve_config_headless(): the same /boomer-config payload from a plain HTTP server, for a headless run whose values change while it runs. @@ -64,7 +64,7 @@ def _attr(flag: str) -> str: def build_config_json(argv: Iterable[str]) -> str: - """Parse `argv` and return the JSON config payload for boomer-glutton's + """Parse `argv` and return the JSON config payload for boomer-worker's --config-json flag. Unknown args are ignored; unset flags are omitted so boomer falls back to its own defaults.""" p = argparse.ArgumentParser(add_help=False) diff --git a/benchmarking/locust/manifests/locust.yaml b/benchmarking/locust/manifests/locust.yaml index 73bdb75ac9..36374e61b4 100644 --- a/benchmarking/locust/manifests/locust.yaml +++ b/benchmarking/locust/manifests/locust.yaml @@ -20,7 +20,7 @@ # LOCUST_NO_GLUTTON_USER=1 prevents tests/glutton.py from declaring the # stub User on this worker so GluttonUser spawns are owned exclusively # by the boomer container. -# * boomer-glutton: Go re-implementation of GluttonUser and DurdirUser. Connects to the +# * boomer-worker: Go re-implementation of GluttonUser and DurdirUser. Connects to the # master at localhost:5557 via the locust worker ZMQ protocol; exposes # its own per-worker diagnostics at /metrics on :8001 (aggregate stats # flow through the master via boomer.RecordSuccess). @@ -114,12 +114,12 @@ spec: # cpu: "500m" # memory: "512Mi" - - name: boomer-glutton - # Same image as the master/worker — the boomer-glutton Go binary is - # baked in at /app/boomer-glutton by benchmarking/locust/Dockerfile. + - name: boomer-worker + # Same image as the master/worker — the boomer-worker Go binary is + # baked in at /app/boomer-worker by benchmarking/locust/Dockerfile. image: us-docker.pkg.dev/${PROJECT_ID}/gcr.io/ate-images/locust-test:latest imagePullPolicy: Always - command: ["/app/boomer-glutton"] + command: ["/app/boomer-worker"] args: - "--prometheus-addr=:8001" # Fetch runtime config (trace probability, wait-time bounds) from diff --git a/benchmarking/locust/runner.py b/benchmarking/locust/runner.py index 1504a04fdc..14bd15629d 100644 --- a/benchmarking/locust/runner.py +++ b/benchmarking/locust/runner.py @@ -18,7 +18,7 @@ to JSONL, and uploads everything to either GCS or local disk under /runs///. -When the test target is glutton.py, also spawns the boomer-glutton Go +When the test target is glutton.py, also spawns the boomer-worker Go worker as a subprocess (locust runs in --master + --expect-workers=1 mode) so the GluttonUser load comes from boomer instead of Python+gevent. @@ -45,9 +45,9 @@ from common.boomer_config import build_config_json -# Path inside the locust image to the boomer-glutton binary baked in by +# Path inside the locust image to the boomer-worker binary baked in by # benchmarking/locust/Dockerfile. -BOOMER_BINARY = "/app/boomer-glutton" +BOOMER_BINARY = "/app/boomer-worker" # Port for the headless /boomer-config server (common/boomer_config.py), which # gives boomer the values that change while a run continues. Locust already diff --git a/benchmarking/locust/tests/durdir.py b/benchmarking/locust/tests/durdir.py index c97285d387..8790e81984 100644 --- a/benchmarking/locust/tests/durdir.py +++ b/benchmarking/locust/tests/durdir.py @@ -15,7 +15,7 @@ """Stub DurdirUser declaration. The real load implementation lives in the boomer-Go worker at -cmd/benchmarking/boomer-glutton/; this Python class is declared only so the +cmd/benchmarking/boomer-worker/; this Python class is declared only so the master recognizes the name and attributes boomer's stats rows to it. The master loads this stub file (or glutton.py), selected by ${BENCHMARK_USER_CLASS} in locust/manifests/locust.yaml. The Python worker @@ -29,7 +29,7 @@ from locust import User, task from common.boomer_config import init_boomer_config - # Master serves /boomer-config so the boomer-glutton workers can fetch + # Master serves /boomer-config so the boomer-worker workers can fetch # runtime flag values (trace probability, wait times, durdir config) the # operator set in the web UI form. No-op on workers without a web UI. init_boomer_config() diff --git a/benchmarking/locust/tests/glutton.py b/benchmarking/locust/tests/glutton.py index dba9ea71b7..0d19a36e69 100644 --- a/benchmarking/locust/tests/glutton.py +++ b/benchmarking/locust/tests/glutton.py @@ -15,7 +15,7 @@ """Stub GluttonUser declaration. The real load implementation lives in the boomer-Go worker at -cmd/boomer-glutton/; this Python class is declared only so the master's +cmd/benchmarking/boomer-worker/; this Python class is declared only so the master's --class-picker enumerates the name. The Python worker container sets LOCUST_NO_GLUTTON_USER=1 to skip loading this file, leaving boomer as the sole owner of GluttonUser load. @@ -27,7 +27,7 @@ from locust import User, task from common.boomer_config import init_boomer_config - # Master serves /boomer-config so the boomer-glutton workers can fetch + # Master serves /boomer-config so the boomer-worker workers can fetch # runtime flag values (trace probability, wait times) the operator set # in the web UI form. No-op on workers without a web UI. init_boomer_config() diff --git a/benchmarking/observability.md b/benchmarking/observability.md index 9f90dd09a6..c635b000e6 100644 --- a/benchmarking/observability.md +++ b/benchmarking/observability.md @@ -51,7 +51,7 @@ scheduler and not the telemetry of a working system. Raise `workerCount` with the steps to go higher. The locust web UI is for manual examination only, and it holds no ladder. Two -conditions apply there. The `boomer-glutton` sidecar makes its own load for +conditions apply there. The `boomer-worker` sidecar makes its own load for each user class that you select in the form. Also, the form changes the sample rate of the boomer worker but not of the Python workers: `locust.yaml` gives boomer `--master-web-port`, thus boomer reads `/boomer-config` from the master diff --git a/cmd/benchmarking/boomer-glutton/main.go b/cmd/benchmarking/boomer-worker/main.go similarity index 98% rename from cmd/benchmarking/boomer-glutton/main.go rename to cmd/benchmarking/boomer-worker/main.go index 86cb487286..0215dd16f5 100644 --- a/cmd/benchmarking/boomer-glutton/main.go +++ b/cmd/benchmarking/boomer-worker/main.go @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -// boomer-glutton is the Go re-implementation of the GluttonUser locust test. +// boomer-worker is the Go re-implementation of the GluttonUser locust test. // It speaks the locust worker protocol via myzhan/boomer, so it appears as a // regular worker to the Python locust master while sidestepping gevent's // scheduling tax. diff --git a/internal/benchmarking/boomer/trace/trace.go b/internal/benchmarking/boomer/trace/trace.go index e19e3de1b9..09961988bf 100644 --- a/internal/benchmarking/boomer/trace/trace.go +++ b/internal/benchmarking/boomer/trace/trace.go @@ -92,7 +92,7 @@ func Init(ctx context.Context, serviceName string, sampler sdktrace.Sampler) (*s } // Route OTel SDK errors (export failures, queue drops) into slog so they - // land in boomer-glutton's stdout — runner.py pumps that into logs.txt. + // land in boomer-worker's stdout — runner.py pumps that into logs.txt. // Without this they go to the SDK's default handler (stderr via log.Println) // and can be lost depending on stream wiring. otel.SetErrorHandler(otel.ErrorHandlerFunc(func(err error) { From b98cd97fafa01bebb6ce4496fccbb5d8035be9df Mon Sep 17 00:00:00 2001 From: Parnika Mandewalkar Date: Fri, 28 Aug 2026 19:21:17 +0000 Subject: [PATCH 2/3] benchmarking: Extract shared boomer benchmarking utilities --- cmd/benchmarking/boomer-worker/main.go | 7 +- .../grpcclient.go => boomerutil/client.go} | 2 +- .../boomer/boomerutil/telemetry.go | 82 +++++++++++++++++++ .../benchmarking/boomer/glutton/durdir.go | 34 +++----- .../benchmarking/boomer/glutton/lifecycle.go | 64 +++------------ 5 files changed, 111 insertions(+), 78 deletions(-) rename internal/benchmarking/boomer/{glutton/grpcclient.go => boomerutil/client.go} (99%) create mode 100644 internal/benchmarking/boomer/boomerutil/telemetry.go diff --git a/cmd/benchmarking/boomer-worker/main.go b/cmd/benchmarking/boomer-worker/main.go index 0215dd16f5..45b7aceb94 100644 --- a/cmd/benchmarking/boomer-worker/main.go +++ b/cmd/benchmarking/boomer-worker/main.go @@ -28,12 +28,15 @@ import ( "strings" "time" + "github.com/agent-substrate/substrate/internal/benchmarking/boomer/boomerutil" "github.com/agent-substrate/substrate/internal/benchmarking/boomer/dynconfig" - "github.com/agent-substrate/substrate/internal/benchmarking/boomer/glutton" bmetrics "github.com/agent-substrate/substrate/internal/benchmarking/boomer/metrics" btrace "github.com/agent-substrate/substrate/internal/benchmarking/boomer/trace" "github.com/agent-substrate/substrate/internal/benchmarking/boomer/userclass" "github.com/myzhan/boomer" + + // Register user classes via init(): + _ "github.com/agent-substrate/substrate/internal/benchmarking/boomer/glutton" ) func main() { @@ -74,7 +77,7 @@ func main() { _ = tp.Shutdown(shutdownCtx) }() - conn, apiStub, err := glutton.DialControl(*apiEndpoint) + conn, apiStub, err := boomerutil.DialControl(*apiEndpoint) if err != nil { slog.Error("failed to dial ateapi", slog.String("err", err.Error())) os.Exit(1) diff --git a/internal/benchmarking/boomer/glutton/grpcclient.go b/internal/benchmarking/boomer/boomerutil/client.go similarity index 99% rename from internal/benchmarking/boomer/glutton/grpcclient.go rename to internal/benchmarking/boomer/boomerutil/client.go index 696f0a3b26..01d6e04c33 100644 --- a/internal/benchmarking/boomer/glutton/grpcclient.go +++ b/internal/benchmarking/boomer/boomerutil/client.go @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -package glutton +package boomerutil import ( "fmt" diff --git a/internal/benchmarking/boomer/boomerutil/telemetry.go b/internal/benchmarking/boomer/boomerutil/telemetry.go new file mode 100644 index 0000000000..e989f9ce18 --- /dev/null +++ b/internal/benchmarking/boomer/boomerutil/telemetry.go @@ -0,0 +1,82 @@ +// 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 boomerutil + +import ( + "log/slog" + "net/http" + "strconv" + "time" + + "go.opentelemetry.io/otel/trace" + "google.golang.org/grpc/metadata" +) + +const ( + SourceClient = "client" + SourceServer = "server" +) + +// MsFloat converts a duration to a float64 representing milliseconds. +func MsFloat(d time.Duration) float64 { + return float64(d.Nanoseconds()) / 1e6 +} + +// LogSampledTrace emits a single structured line per sampled span. Operators +// parse these lines (e.g. via fluentbit) to rebuild the trace stream. +func LogSampledTrace(span trace.Span, name string, latency time.Duration, source string, err error) { + sc := span.SpanContext() + if !sc.IsSampled() { + return + } + attrs := []any{ + slog.String("name", name), + slog.String("trace_id", sc.TraceID().String()), + slog.Float64("duration_ms", MsFloat(latency)), + slog.String("source", source), + } + if err != nil { + attrs = append(attrs, slog.String("err", err.Error())) + slog.Info("traced span (failed)", attrs...) + return + } + slog.Info("traced span", attrs...) +} + +// ElapsedFromMD extracts the server elapsed time from gRPC metadata, falling back to client latency. +func ElapsedFromMD(tr metadata.MD, key string, fallback time.Duration) (time.Duration, string) { + vals := tr.Get(key) + if len(vals) == 0 { + return fallback, SourceClient + } + us, err := strconv.ParseInt(vals[0], 10, 64) + if err != nil { + return fallback, SourceClient + } + return time.Duration(us) * time.Microsecond, SourceServer +} + +// ElapsedFromHeader extracts the server elapsed time from HTTP headers, falling back to client latency. +func ElapsedFromHeader(h http.Header, key string, fallback time.Duration) (time.Duration, string) { + val := h.Get(key) + if val == "" { + return fallback, SourceClient + } + us, err := strconv.ParseInt(val, 10, 64) + if err != nil { + return fallback, SourceClient + } + return time.Duration(us) * time.Microsecond, SourceServer +} diff --git a/internal/benchmarking/boomer/glutton/durdir.go b/internal/benchmarking/boomer/glutton/durdir.go index 0d6427dadf..dcea76eb41 100644 --- a/internal/benchmarking/boomer/glutton/durdir.go +++ b/internal/benchmarking/boomer/glutton/durdir.go @@ -24,12 +24,12 @@ import ( "log/slog" "math/rand/v2" "net/http" - "strconv" "strings" "sync" "time" "github.com/agent-substrate/substrate/internal/ateinterceptors" + "github.com/agent-substrate/substrate/internal/benchmarking/boomer/boomerutil" "github.com/agent-substrate/substrate/internal/benchmarking/boomer/dynconfig" bmetrics "github.com/agent-substrate/substrate/internal/benchmarking/boomer/metrics" "github.com/agent-substrate/substrate/internal/benchmarking/boomer/userclass" @@ -251,11 +251,11 @@ func (u *durDirUser) tracedCall(ctx context.Context, name string, do func(contex err := do(ctx, &tr) clientLatency := time.Since(start) - latency, source := elapsedFromMD(tr, ateinterceptors.ServerElapsedTrailer, clientLatency) - if source == sourceServer { - span.SetAttributes(attribute.Float64("server.elapsed_ms", msFloat(latency))) + latency, source := boomerutil.ElapsedFromMD(tr, ateinterceptors.ServerElapsedTrailer, clientLatency) + if source == boomerutil.SourceServer { + span.SetAttributes(attribute.Float64("server.elapsed_ms", boomerutil.MsFloat(latency))) } - logSampledTrace(span, name, latency, source, err) + boomerutil.LogSampledTrace(span, name, latency, source, err) if err != nil { bmetrics.RecordFailure("grpc", name, u.userClass, latency, err.Error()) return err @@ -432,39 +432,27 @@ func (u *durDirUser) httpProtoCall(ctx context.Context, metricName, route string return nil, readErr } - serverLatency, source := elapsedFromHeader(resp.Header, ateinterceptors.ServerElapsedTrailer, clientLatency) - if source == sourceServer { - span.SetAttributes(attribute.Float64("server.elapsed_ms", msFloat(serverLatency))) + serverLatency, source := boomerutil.ElapsedFromHeader(resp.Header, ateinterceptors.ServerElapsedTrailer, clientLatency) + if source == boomerutil.SourceServer { + span.SetAttributes(attribute.Float64("server.elapsed_ms", boomerutil.MsFloat(serverLatency))) } if resp.StatusCode >= 400 { httpErr := fmt.Errorf("HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(respBody))) - logSampledTrace(span, metricName, clientLatency, sourceClient, httpErr) + boomerutil.LogSampledTrace(span, metricName, clientLatency, boomerutil.SourceClient, httpErr) bmetrics.RecordFailure("http", metricName, u.userClass, clientLatency, httpErr.Error()) return nil, httpErr } if validate != nil { if err := validate(respBody); err != nil { - logSampledTrace(span, metricName, clientLatency, sourceClient, err) + boomerutil.LogSampledTrace(span, metricName, clientLatency, boomerutil.SourceClient, err) bmetrics.RecordFailure("http", metricName, u.userClass, clientLatency, err.Error()) return nil, err } } - logSampledTrace(span, metricName, clientLatency, sourceClient, nil) + boomerutil.LogSampledTrace(span, metricName, clientLatency, boomerutil.SourceClient, nil) bmetrics.RecordSuccess("http", metricName, u.userClass, clientLatency, int64(len(respBody))) return respBody, nil } - -func elapsedFromHeader(h http.Header, key string, fallback time.Duration) (time.Duration, string) { - val := h.Get(key) - if val == "" { - return fallback, sourceClient - } - us, err := strconv.ParseInt(val, 10, 64) - if err != nil { - return fallback, sourceClient - } - return time.Duration(us) * time.Microsecond, sourceServer -} diff --git a/internal/benchmarking/boomer/glutton/lifecycle.go b/internal/benchmarking/boomer/glutton/lifecycle.go index 29438bdac3..8d474313bc 100644 --- a/internal/benchmarking/boomer/glutton/lifecycle.go +++ b/internal/benchmarking/boomer/glutton/lifecycle.go @@ -32,6 +32,7 @@ import ( "time" "github.com/agent-substrate/substrate/internal/ateinterceptors" + "github.com/agent-substrate/substrate/internal/benchmarking/boomer/boomerutil" bmetrics "github.com/agent-substrate/substrate/internal/benchmarking/boomer/metrics" "github.com/agent-substrate/substrate/internal/benchmarking/boomer/userclass" gluttonpb "github.com/agent-substrate/substrate/internal/proto/glutton" @@ -40,7 +41,6 @@ import ( "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/propagation" - "go.opentelemetry.io/otel/trace" "google.golang.org/grpc" "google.golang.org/grpc/codes" "google.golang.org/grpc/metadata" @@ -51,8 +51,6 @@ import ( const ( userClass = "GluttonUser" templateName = "glutton" - // templateAtespace is the atespace holding the benchmark ActorTemplates - // (see benchmarking/workloads/manifests). templateAtespace = "benchmark-workloads" actorDomain = "actors.resources.substrate.ate.dev" pingPath = "/ping" @@ -60,9 +58,6 @@ const ( readRAMPath = "/readram" memLoadKey = "memload" memReadAll = "all" - - sourceClient = "client" - sourceServer = "server" ) func init() { @@ -276,11 +271,11 @@ func (u *gluttonUser) tracedCall(ctx context.Context, name string, do func(conte err := do(ctx, &tr) clientLatency := time.Since(start) - latency, source := elapsedFromMD(tr, ateinterceptors.ServerElapsedTrailer, clientLatency) - if source == sourceServer { - span.SetAttributes(attribute.Float64("server.elapsed_ms", msFloat(latency))) + latency, source := boomerutil.ElapsedFromMD(tr, ateinterceptors.ServerElapsedTrailer, clientLatency) + if source == boomerutil.SourceServer { + span.SetAttributes(attribute.Float64("server.elapsed_ms", boomerutil.MsFloat(latency))) } - logSampledTrace(span, name, latency, source, err) + boomerutil.LogSampledTrace(span, name, latency, source, err) if err != nil { bmetrics.RecordFailure("grpc", name, userClass, latency, err.Error()) return err @@ -326,24 +321,24 @@ func (u *gluttonUser) ping(ctx context.Context) { if resp.StatusCode >= 400 { httpErr := fmt.Errorf("HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(respBody))) - logSampledTrace(span, "GluttonPing", clientLatency, sourceClient, httpErr) + boomerutil.LogSampledTrace(span, "GluttonPing", clientLatency, boomerutil.SourceClient, httpErr) bmetrics.RecordFailure("http", "GluttonPing", userClass, clientLatency, httpErr.Error()) return } pong := &gluttonpb.PingResponse{} if err := proto.Unmarshal(respBody, pong); err != nil { - logSampledTrace(span, "GluttonPing", clientLatency, sourceClient, err) + boomerutil.LogSampledTrace(span, "GluttonPing", clientLatency, boomerutil.SourceClient, err) bmetrics.RecordFailure("http", "GluttonPing", userClass, clientLatency, err.Error()) return } if pong.Message != message { mismatch := fmt.Errorf("ping echo mismatch: sent=%q recv=%q", message, pong.Message) - logSampledTrace(span, "GluttonPing", clientLatency, sourceClient, mismatch) + boomerutil.LogSampledTrace(span, "GluttonPing", clientLatency, boomerutil.SourceClient, mismatch) bmetrics.RecordFailure("http", "GluttonPing", userClass, clientLatency, mismatch.Error()) return } - logSampledTrace(span, "GluttonPing", clientLatency, sourceClient, nil) + boomerutil.LogSampledTrace(span, "GluttonPing", clientLatency, boomerutil.SourceClient, nil) bmetrics.RecordSuccess("http", "GluttonPing", userClass, clientLatency, int64(len(respBody))) } @@ -370,7 +365,7 @@ func (u *gluttonUser) ensureRAMFilled(ctx context.Context) { err := u.writeRAM(ctx, memLoadKey, target, gluttonpb.WriteMode_WRITE_MODE_TRUNCATE) clientLatency := time.Since(start) - logSampledTrace(span, "GluttonFillRAM", clientLatency, sourceClient, err) + boomerutil.LogSampledTrace(span, "GluttonFillRAM", clientLatency, boomerutil.SourceClient, err) if err != nil { bmetrics.RecordFailure("http", "GluttonFillRAM", userClass, clientLatency, err.Error()) return @@ -400,7 +395,7 @@ func (u *gluttonUser) churnRAM(ctx context.Context) { err := u.writeRAM(ctx, memLoadKey, churn, gluttonpb.WriteMode_WRITE_MODE_OVERWRITE_ROTATE) clientLatency := time.Since(start) - logSampledTrace(span, "GluttonChurnRAM", clientLatency, sourceClient, err) + boomerutil.LogSampledTrace(span, "GluttonChurnRAM", clientLatency, boomerutil.SourceClient, err) if err != nil { bmetrics.RecordFailure("http", "GluttonChurnRAM", userClass, clientLatency, err.Error()) return @@ -431,7 +426,7 @@ func (u *gluttonUser) readRAM(ctx context.Context) { resp := &gluttonpb.ReadRAMResponse{} err := u.postProto(ctx, readRAMPath, &gluttonpb.ReadRAMRequest{Key: memLoadKey, Size: size}, resp) clientLatency := time.Since(start) - logSampledTrace(span, "GluttonReadRAM", clientLatency, sourceClient, err) + boomerutil.LogSampledTrace(span, "GluttonReadRAM", clientLatency, boomerutil.SourceClient, err) if err != nil { bmetrics.RecordFailure("http", "GluttonReadRAM", userClass, clientLatency, err.Error()) return @@ -484,42 +479,7 @@ func (u *gluttonUser) postProto(ctx context.Context, path string, req, resp prot return proto.Unmarshal(respBody, resp) } -// logSampledTrace emits a single structured line per sampled span. Operators -// (and runner.py) grep stdout for `trace_id=` to find the trace IDs to look -// up in the OTLP backend. Matches the format of the Python locust workers' -// equivalent log line so a single regex covers both sources. -func logSampledTrace(span trace.Span, name string, latency time.Duration, source string, err error) { - sc := span.SpanContext() - if !sc.IsSampled() { - return - } - attrs := []any{ - slog.String("name", name), - slog.String("trace_id", sc.TraceID().String()), - slog.Float64("duration_ms", msFloat(latency)), - slog.String("source", source), - } - if err != nil { - attrs = append(attrs, slog.String("err", err.Error())) - slog.Info("traced span (failed)", attrs...) - return - } - slog.Info("traced span", attrs...) -} - -func elapsedFromMD(tr metadata.MD, key string, fallback time.Duration) (time.Duration, string) { - vals := tr.Get(key) - if len(vals) == 0 { - return fallback, sourceClient - } - us, err := strconv.ParseInt(vals[0], 10, 64) - if err != nil { - return fallback, sourceClient - } - return time.Duration(us) * time.Microsecond, sourceServer -} -func msFloat(d time.Duration) float64 { return float64(d.Nanoseconds()) / 1e6 } // goroutineID extracts the runtime's per-goroutine ID via the standard // runtime.Stack trick. Used to key per-VU state because boomer's Task model From 02cd850e072e01557459f3ddb55bf73b305161f5 Mon Sep 17 00:00:00 2001 From: Parnika Mandewalkar Date: Fri, 28 Aug 2026 22:15:35 +0000 Subject: [PATCH 3/3] Refactor goroutine ID extraction into boomerutil helper --- .../boomer/boomerutil/goroutine.go | 41 +++++++++++++++++++ .../benchmarking/boomer/glutton/durdir.go | 2 +- .../boomer/glutton/durdir_test.go | 3 +- .../benchmarking/boomer/glutton/lifecycle.go | 30 ++------------ 4 files changed, 47 insertions(+), 29 deletions(-) create mode 100644 internal/benchmarking/boomer/boomerutil/goroutine.go diff --git a/internal/benchmarking/boomer/boomerutil/goroutine.go b/internal/benchmarking/boomer/boomerutil/goroutine.go new file mode 100644 index 0000000000..08f4558763 --- /dev/null +++ b/internal/benchmarking/boomer/boomerutil/goroutine.go @@ -0,0 +1,41 @@ +// 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 boomerutil + +import ( + "runtime" + "strconv" + "strings" +) + +// goroutineID extracts the runtime's per-goroutine ID via the standard +// runtime.Stack trick. Used to key per-VU state because boomer's Task model +// has no built-in per-VU hook — see the runtime.shutdown comment for the +// limitation this implies on user-count rescale. +func GoroutineID() int64 { + var buf [64]byte + n := runtime.Stack(buf[:], false) + line := string(buf[:n]) + const prefix = "goroutine " + if !strings.HasPrefix(line, prefix) { + return 0 + } + end := strings.IndexByte(line[len(prefix):], ' ') + if end < 0 { + return 0 + } + id, _ := strconv.ParseInt(line[len(prefix):len(prefix)+end], 10, 64) + return id +} diff --git a/internal/benchmarking/boomer/glutton/durdir.go b/internal/benchmarking/boomer/glutton/durdir.go index dcea76eb41..f526be6a71 100644 --- a/internal/benchmarking/boomer/glutton/durdir.go +++ b/internal/benchmarking/boomer/glutton/durdir.go @@ -93,7 +93,7 @@ func (r *durDirRuntime) dynamicWait() time.Duration { } func (r *durDirRuntime) iterate() { - gid := goroutineID() + gid := boomerutil.GoroutineID() val, loaded := r.users.Load(gid) if !loaded { dynCfg := r.cfg.Dyn.Load() diff --git a/internal/benchmarking/boomer/glutton/durdir_test.go b/internal/benchmarking/boomer/glutton/durdir_test.go index b99bccbcd2..ebca00d78e 100644 --- a/internal/benchmarking/boomer/glutton/durdir_test.go +++ b/internal/benchmarking/boomer/glutton/durdir_test.go @@ -24,6 +24,7 @@ import ( "slices" "testing" + "github.com/agent-substrate/substrate/internal/benchmarking/boomer/boomerutil" "github.com/agent-substrate/substrate/internal/benchmarking/boomer/dynconfig" "github.com/agent-substrate/substrate/internal/benchmarking/boomer/userclass" "github.com/agent-substrate/substrate/internal/benchmarking/glutton/fake" @@ -262,7 +263,7 @@ func TestDurDirShutdownSuspendsBeforeDelete(t *testing.T) { du := newTestDurDirUser(t, &fake.Server{}, cfg) rt := &durDirRuntime{cfg: du.cfg} - rt.users.Store(goroutineID(), du) + rt.users.Store(boomerutil.GoroutineID(), du) rt.shutdown(context.Background()) calls := fakeCtrl.recordedCalls() diff --git a/internal/benchmarking/boomer/glutton/lifecycle.go b/internal/benchmarking/boomer/glutton/lifecycle.go index 8d474313bc..5ade96de2a 100644 --- a/internal/benchmarking/boomer/glutton/lifecycle.go +++ b/internal/benchmarking/boomer/glutton/lifecycle.go @@ -25,8 +25,6 @@ import ( "log/slog" "math/rand/v2" "net/http" - "runtime" - "strconv" "strings" "sync" "time" @@ -49,8 +47,8 @@ import ( ) const ( - userClass = "GluttonUser" - templateName = "glutton" + userClass = "GluttonUser" + templateName = "glutton" templateAtespace = "benchmark-workloads" actorDomain = "actors.resources.substrate.ate.dev" pingPath = "/ping" @@ -90,7 +88,7 @@ type taskRuntime struct { // (the analog of locust's per-user on_start); subsequent calls run a // resume/ping/suspend cycle. func (r *taskRuntime) iterate() { - gid := goroutineID() + gid := boomerutil.GoroutineID() val, loaded := r.users.Load(gid) if !loaded { u, err := r.startUser(context.Background()) @@ -478,25 +476,3 @@ func (u *gluttonUser) postProto(ctx context.Context, path string, req, resp prot } return proto.Unmarshal(respBody, resp) } - - - -// goroutineID extracts the runtime's per-goroutine ID via the standard -// runtime.Stack trick. Used to key per-VU state because boomer's Task model -// has no built-in per-VU hook — see the runtime.shutdown comment for the -// limitation this implies on user-count rescale. -func goroutineID() int64 { - var buf [64]byte - n := runtime.Stack(buf[:], false) - line := string(buf[:n]) - const prefix = "goroutine " - if !strings.HasPrefix(line, prefix) { - return 0 - } - end := strings.IndexByte(line[len(prefix):], ' ') - if end < 0 { - return 0 - } - id, _ := strconv.ParseInt(line[len(prefix):len(prefix)+end], 10, 64) - return id -}