Add benchmarking e2e test

internal/benchmarking/boomer/glutton/exercise.go

@@ -0,0 +1,68 @@
+// 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"
+	"time"
+
+	"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.
+//
+// interCallDelay is the pause inserted between successive RPC calls in
+// each iteration (Resume→Ping and Ping→Suspend). Pass zero to disable;
+// pass a small delay (e.g. 200ms) on slow CI runners where the atenet
+// router needs time to pick up a Resume before the Ping is routable.
+func ExerciseBenchmark(ctx context.Context, apiStub ateapipb.ControlClient, httpClient *http.Client, routerURL string, interCallDelay time.Duration) error {
+	cfg := &Config{
+		APIStub:        apiStub,
+		HTTPClient:     httpClient,
+		RouterURL:      routerURL,
+		Atespace:       "benchmark",
+		Dyn:            dynconfig.NewHolder(dynconfig.Config{}),
+		Tracer:         otel.Tracer("substrate-boomer/glutton"),
+		InterCallDelay: interCallDelay,
+	}
+	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("u.ensureAtespace: %w", err)
+	}
+	if err := u.createActor(ctx); err != nil {
+		return fmt.Errorf("u.createActor: %w", err)
+	}
+	for _, phase := range []string{"cold", "warm"} {
+		if err := u.runIteration(ctx); err != nil {
+			return fmt.Errorf("%s u.runIteration: %w", phase, err)
+		}
+	}
+	if err := u.delete(ctx); err != nil {
+		return fmt.Errorf("u.delete: %w", err)
+	}
+	return nil
+}

internal/benchmarking/boomer/glutton/lifecycle.go

@@ -20,6 +20,7 @@ package glutton
 import (
 	"bytes"
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"log/slog"
@@ -77,6 +78,11 @@ type Config struct {
 	Dyn *dynconfig.Holder
 	// Tracer anchors sampled spans; falls back to the otel global if nil.
 	Tracer trace.Tracer
+	// InterCallDelay is the pause between successive RPC calls inside a
+	// single boomer iteration (Resume→Ping and Ping→Suspend). Zero
+	// (default) disables it. Useful in slower environments where the atenet router needs a moment to
+	// pick up the actor's Resume before a Ping can be routed correctly.
+	InterCallDelay time.Duration
 }
 
 // Register creates a runtime tied to cfg and returns a boomer-compatible task
@@ -113,16 +119,35 @@ 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
+	}
+	u.interCallDelay()
+	pingErr := u.ping(ctx)
+	u.interCallDelay()
+	suspendErr := u.suspend(ctx)
+	return errors.Join(pingErr, suspendErr)
+}
+
+func (u *gluttonUser) interCallDelay() {
+	if d := u.cfg.InterCallDelay; d > 0 {
+		time.Sleep(d)
+	}
+}
+
 func (r *taskRuntime) startUser(ctx context.Context) (*gluttonUser, error) {
 	u := &gluttonUser{
 		cfg:         r.cfg,
@@ -135,7 +160,7 @@ func (r *taskRuntime) startUser(ctx context.Context) (*gluttonUser, error) {
 		bmetrics.UpdateUsers(userClass, -1)
 		return nil, err
 	}
-	if err := u.create(ctx); err != nil {
+	if err := u.createActor(ctx); err != nil {
 		bmetrics.UpdateUsers(userClass, -1)
 		return nil, err
 	}
@@ -150,9 +175,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
 	})
@@ -202,7 +227,7 @@ func (u *gluttonUser) ensureAtespace(ctx context.Context) error {
 	})
 }
 
-func (u *gluttonUser) create(ctx context.Context) error {
+func (u *gluttonUser) createActor(ctx context.Context) error {
 	return u.tracedCall(ctx, "CreateActor", func(callCtx context.Context, tr *metadata.MD) error {
 		_, err := u.cfg.APIStub.CreateActor(callCtx, &ateapipb.CreateActorRequest{
 			Actor: &ateapipb.Actor{
@@ -215,7 +240,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"
@@ -228,25 +257,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{
 			Actor: 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{
 			Actor: u.ref(),
 		}, grpc.Trailer(tr))
@@ -280,21 +310,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")
@@ -305,37 +338,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.