benchmarking: Refactor boomer based Locust benchmarking

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/

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..."

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)

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

benchmarking/locust/runner.py

@@ -18,7 +18,7 @@
 to JSONL, and uploads everything to either GCS or local disk under
 <dest>/runs/<tag>/<timestamp>/.
 
-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

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()

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()

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

cmd/benchmarking/boomer-glutton/main.go → 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.
@@ -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)

internal/benchmarking/boomer/glutton/grpcclient.go → 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"

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
+}

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
+}

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"
@@ -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()
@@ -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
-}