Add benchmarking e2e test

Author: maxsmythe

internal/benchmarking/boomer/glutton/exercise.go

@@ -0,0 +1,60 @@
+// 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 glutton
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+
+	"github.com/agent-substrate/substrate/internal/benchmarking/boomer/dynconfig"
+	"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
+	"go.opentelemetry.io/otel"
+)
+
+// ExerciseBenchmark runs one full boomer VU lifecycle against a live
+// cluster. It returns an error if there are any problems running the benchmark.
+func ExerciseBenchmark(ctx context.Context, apiStub ateapipb.ControlClient, httpClient *http.Client, routerURL string) error {
+	cfg := &Config{
+		APIStub:    apiStub,
+		HTTPClient: httpClient,
+		RouterURL:  routerURL,
+		Atespace:   "benchmark",
+		Dyn:        dynconfig.NewHolder(dynconfig.Config{}),
+		Tracer:     otel.Tracer("substrate-boomer/glutton"),
+	}
+	u := &gluttonUser{
+		cfg:         cfg,
+		actorID:     "contract",
+		firstResume: true,
+	}
+	u.hostHeader = u.actorID + "." + cfg.Atespace + "." + actorDomain
+
+	if err := u.ensureAtespace(ctx); err != nil {
+		return fmt.Errorf("EnsureAtespace: %w", err)
+	}
+	if err := u.create(ctx); err != nil {
+		return fmt.Errorf("CreateActor: %w", err)
+	}
+	for _, phase := range []string{"cold", "warm"} {
+		if err := u.runIteration(ctx); err != nil {
+			return fmt.Errorf("%s iteration: %w", phase, err)
+		}
+	}
+	if err := u.delete(ctx); err != nil {
+		return fmt.Errorf("DeleteActor: %w", err)
+	}
+	return nil
+}

internal/benchmarking/boomer/glutton/lifecycle.go

@@ -20,6 +20,7 @@ package glutton
 import (
 	"bytes"
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"log/slog"
@@ -113,16 +114,27 @@ func (r *taskRuntime) iterate() {
 	}
 	user := val.(*gluttonUser)
 
-	ctx := context.Background()
-	if !user.resume(ctx) {
-		return
-	}
-	user.ping(ctx)
-	user.suspend(ctx)
+	_ = user.runIteration(context.Background())
 
 	time.Sleep(r.dynamicWait())
 }
 
+// runIteration is one boomer iteration for a single user: Resume, then
+// (on successful resume) Ping and Suspend. Ping and Suspend both run
+// regardless of Ping's outcome so a failed ping still puts the actor
+// back to Suspended. Returned error aggregates whatever went wrong;
+// production callers ignore it (failures land in metrics inside
+// tracedCall), tests use it to fail loudly on contract breakage.
+func (u *gluttonUser) runIteration(ctx context.Context) error {
+	ok, err := u.resume(ctx)
+	if !ok {
+		return err
+	}
+	pingErr := u.ping(ctx)
+	suspendErr := u.suspend(ctx)
+	return errors.Join(pingErr, suspendErr)
+}
+
 func (r *taskRuntime) startUser(ctx context.Context) (*gluttonUser, error) {
 	u := &gluttonUser{
 		cfg:         r.cfg,
@@ -150,9 +162,9 @@ func (r *taskRuntime) shutdown(ctx context.Context) {
 	r.users.Range(func(_, val any) bool {
 		u := val.(*gluttonUser)
 		if u.actorRunning {
-			u.suspend(ctx)
+			_ = u.suspend(ctx)
 		}
-		u.delete(ctx)
+		_ = u.delete(ctx)
 		bmetrics.UpdateUsers(userClass, -1)
 		return true
 	})
@@ -209,7 +221,11 @@ func (u *gluttonUser) create(ctx context.Context) error {
 	})
 }
 
-func (u *gluttonUser) resume(ctx context.Context) bool {
+// resume returns (ok, err): ok mirrors iterate()'s "should I keep going"
+// gate, err carries the underlying gRPC error so contract tests can fail
+// loudly. Production callers ignore err — the failure is already reported
+// to metrics inside tracedCall.
+func (u *gluttonUser) resume(ctx context.Context) (bool, error) {
 	metricName := "ResumeActor"
 	if u.firstResume {
 		metricName = "ResumeActorColdStart"
@@ -222,25 +238,26 @@ func (u *gluttonUser) resume(ctx context.Context) bool {
 		return err
 	})
 	if err != nil {
-		return false
+		return false, err
 	}
 	u.firstResume = false
 	u.actorRunning = true
-	return true
+	return true, nil
 }
 
-func (u *gluttonUser) suspend(ctx context.Context) {
-	_ = u.tracedCall(ctx, "SuspendActor", func(callCtx context.Context, tr *metadata.MD) error {
+func (u *gluttonUser) suspend(ctx context.Context) error {
+	err := u.tracedCall(ctx, "SuspendActor", func(callCtx context.Context, tr *metadata.MD) error {
 		_, err := u.cfg.APIStub.SuspendActor(callCtx, &ateapipb.SuspendActorRequest{
 			ActorRef: u.ref(),
 		}, grpc.Trailer(tr))
 		return err
 	})
 	u.actorRunning = false
+	return err
 }
 
-func (u *gluttonUser) delete(ctx context.Context) {
-	_ = u.tracedCall(ctx, "DeleteActor", func(callCtx context.Context, tr *metadata.MD) error {
+func (u *gluttonUser) delete(ctx context.Context) error {
+	return u.tracedCall(ctx, "DeleteActor", func(callCtx context.Context, tr *metadata.MD) error {
 		_, err := u.cfg.APIStub.DeleteActor(callCtx, &ateapipb.DeleteActorRequest{
 			ActorRef: u.ref(),
 		}, grpc.Trailer(tr))
@@ -274,21 +291,24 @@ func (u *gluttonUser) tracedCall(ctx context.Context, name string, do func(conte
 	return nil
 }
 
-func (u *gluttonUser) ping(ctx context.Context) {
+// ping returns nil on a successful echo. Production callers ignore the
+// return value — failures are already reported to metrics — but contract
+// tests use it to fail loudly on router/actor breakage.
+func (u *gluttonUser) ping(ctx context.Context) error {
 	ctx, span := u.cfg.Tracer.Start(ctx, "GluttonPing")
 	defer span.End()
 
 	message := uuid.NewString()
 	body, err := proto.Marshal(&gluttonpb.PingRequest{Message: message})
 	if err != nil {
 		bmetrics.RecordFailure("http", "GluttonPing", userClass, 0, err.Error())
-		return
+		return err
 	}
 
 	httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, u.cfg.RouterURL+pingPath, bytes.NewReader(body))
 	if err != nil {
 		bmetrics.RecordFailure("http", "GluttonPing", userClass, 0, err.Error())
-		return
+		return err
 	}
 	httpReq.Host = u.hostHeader
 	httpReq.Header.Set("Content-Type", "application/x-protobuf")
@@ -299,37 +319,38 @@ func (u *gluttonUser) ping(ctx context.Context) {
 	clientLatency := time.Since(start)
 	if err != nil {
 		bmetrics.RecordFailure("http", "GluttonPing", userClass, clientLatency, err.Error())
-		return
+		return err
 	}
 	defer resp.Body.Close()
 
 	respBody, readErr := io.ReadAll(resp.Body)
 	if readErr != nil {
 		bmetrics.RecordFailure("http", "GluttonPing", userClass, clientLatency, readErr.Error())
-		return
+		return readErr
 	}
 
 	if resp.StatusCode >= 400 {
 		httpErr := fmt.Errorf("HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(respBody)))
 		logSampledTrace(span, "GluttonPing", clientLatency, sourceClient, httpErr)
 		bmetrics.RecordFailure("http", "GluttonPing", userClass, clientLatency, httpErr.Error())
-		return
+		return httpErr
 	}
 
 	pong := &gluttonpb.PingResponse{}
 	if err := proto.Unmarshal(respBody, pong); err != nil {
 		logSampledTrace(span, "GluttonPing", clientLatency, sourceClient, err)
 		bmetrics.RecordFailure("http", "GluttonPing", userClass, clientLatency, err.Error())
-		return
+		return err
 	}
 	if pong.Message != message {
 		mismatch := fmt.Errorf("ping echo mismatch: sent=%q recv=%q", message, pong.Message)
 		logSampledTrace(span, "GluttonPing", clientLatency, sourceClient, mismatch)
 		bmetrics.RecordFailure("http", "GluttonPing", userClass, clientLatency, mismatch.Error())
-		return
+		return mismatch
 	}
 	logSampledTrace(span, "GluttonPing", clientLatency, sourceClient, nil)
 	bmetrics.RecordSuccess("http", "GluttonPing", userClass, clientLatency, int64(len(respBody)))
+	return nil
 }
 
 // logSampledTrace emits a single structured line per sampled span. Operators

internal/e2e/router_client.go

@@ -193,6 +193,12 @@ func (c *RouterClient) Close() {
 	close(c.stopCh)
 }
 
+// BaseURL returns the local URL of the port-forwarded router (e.g.
+// http://127.0.0.1:34567). HTTP requests against it must set Host to the
+// actor's mesh DNS name (see resources.ActorDNSName) for the router to
+// route them.
+func (c *RouterClient) BaseURL() string { return c.baseURL }
+
 // Get issues GET path to (atespace, actorID) through the router, setting the
 // actor's mesh Host so the router routes (and resumes) it. The caller must close
 // the body.

internal/e2e/suites/benchmarking/benchmarking_test.go

@@ -0,0 +1,110 @@
+// 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 benchmarking is the contract gate for the load-test workflow.
+// It deploys the real benchmarking/workloads manifest and drives one
+// boomer VU's full lifecycle via glutton.ExerciseBenchmark, so a change
+// to the ATE API surface or the WorkerPool/ActorTemplate manifest schema
+// that would silently break benchmarking fails here.
+package benchmarking
+
+import (
+	"context"
+	"net/http"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/agent-substrate/substrate/internal/benchmarking/boomer/glutton"
+	"github.com/agent-substrate/substrate/internal/e2e"
+	"github.com/agent-substrate/substrate/pkg/api/v1alpha1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+const (
+	workloadsNamespace = "benchmark-workloads"
+	gluttonTemplate    = "glutton"
+)
+
+// TestGluttonBenchmarkContract helps ensure that changes will not disrupt benchmarking
+func TestGluttonBenchmarkContract(t *testing.T) {
+	// Fail fast if the env the workloads manifest depends on is missing —
+	// benchmarking/workloads/deploy.sh sources this from .ate-dev-env.sh
+	// and substitutes it into the ActorTemplate's snapshotsConfig.location.
+	if _, err := e2e.CheckEnv("BUCKET_NAME", "KO_DOCKER_REPO"); err != nil {
+		t.Fatalf("CheckEnv failed: %v", err)
+	}
+
+	ctx := context.Background()
+	clients := e2e.GetClients()
+
+	deployWorkloads(t)
+	waitForGluttonTemplateReady(ctx, t, clients)
+
+	rc, err := e2e.NewRouterClient(ctx)
+	if err != nil {
+		t.Fatalf("NewRouterClient: %v", err)
+	}
+	defer rc.Close()
+
+	if err := glutton.ExerciseBenchmark(ctx, clients.SubstrateAPI, &http.Client{Timeout: 30 * time.Second}, rc.BaseURL()); err != nil {
+		t.Fatalf("glutton.ExerciseBenchmark: %v", err)
+	}
+}
+
+// deployWorkloads applies benchmarking/workloads/manifests/workloads.yaml.tmpl
+// via its own deploy.sh so any drift in the template or the script (which
+// orchestrator.py also invokes) is caught here. Registers a Cleanup to
+// tear down the workloads.
+func deployWorkloads(t *testing.T) {
+	t.Helper()
+	root, err := e2e.FindRepoRoot()
+	if err != nil {
+		t.Fatalf("FindRepoRoot: %v", err)
+	}
+	deploy := filepath.Join(root, "benchmarking/workloads/deploy.sh")
+
+	t.Logf("Deploying workloads via %s --deploy", deploy)
+	e2e.RunCmd(t, deploy, "--deploy", "--worker-count", "1")
+
+	t.Cleanup(func() {
+		t.Logf("Tearing down workloads via %s --delete", deploy)
+		e2e.RunCmd(t, deploy, "--delete")
+	})
+}
+
+// waitForGluttonTemplateReady polls the ActorTemplate until it reaches
+// PhaseReady (golden snapshot has been created). Mirrors the wait in
+// orchestrator.py after deploy_workloads.
+func waitForGluttonTemplateReady(ctx context.Context, t *testing.T, clients *e2e.Clients) {
+	t.Helper()
+	timeout := 5 * time.Minute
+	deadline := time.Now().Add(timeout)
+	var lastPhase v1alpha1.PhaseType
+	for time.Now().Before(deadline) {
+		at, err := clients.SubstrateK8s.ApiV1alpha1().ActorTemplates(workloadsNamespace).Get(ctx, gluttonTemplate, metav1.GetOptions{})
+		if err == nil {
+			lastPhase = at.Status.Phase
+			if lastPhase == v1alpha1.PhaseReady {
+				t.Logf("ActorTemplate %s/%s is Ready (golden=%q)", workloadsNamespace, gluttonTemplate, at.Status.GoldenSnapshot)
+				return
+			}
+			if lastPhase == v1alpha1.PhaseFailed {
+				t.Fatalf("ActorTemplate %s/%s entered PhaseFailed", workloadsNamespace, gluttonTemplate)
+			}
+		}
+		time.Sleep(2 * time.Second)
+	}
+	t.Fatalf("timed out after %v waiting for ActorTemplate %s/%s (last phase: %s)", timeout, workloadsNamespace, gluttonTemplate, lastPhase)
+}

internal/e2e/suites/benchmarking/testmain_test.go

@@ -0,0 +1,28 @@
+// 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 benchmarking
+
+import (
+	"os"
+	"testing"
+
+	"github.com/agent-substrate/substrate/internal/e2e"
+)
+
+func run(m *testing.M) int {
+	return e2e.RunTestMain(m)
+}
+
+func TestMain(m *testing.M) { os.Exit(run(m)) }