benchmarking: move glutton's internals into internal/benchmarking/glutton

internal/benchmarking/glutton/fake/server.go

@@ -25,17 +25,18 @@ import (
 	"testing"
 
 	"github.com/agent-substrate/substrate/internal/ateinterceptors"
+	"github.com/agent-substrate/substrate/internal/benchmarking/glutton"
 	gluttonpb "github.com/agent-substrate/substrate/internal/proto/glutton"
 	"google.golang.org/protobuf/proto"
 )
 
-// Routes the fake serves, mirroring glutton's real HTTP mux. Declared here so
-// the fake depends on nothing; collapses onto one source when glutton's core moves.
+// Routes the fake serves, re-exported from the real server so the stand-in
+// cannot answer a path the actor does not.
 const (
-	WriteDiskRoute = "/writedisk"
-	ReadDiskRoute  = "/readdisk"
-	WriteRAMRoute  = "/writeram"
-	ReadRAMRoute   = "/readram"
+	WriteDiskRoute = glutton.WriteDiskRoute
+	ReadDiskRoute  = glutton.ReadDiskRoute
+	WriteRAMRoute  = glutton.WriteRAMRoute
+	ReadRAMRoute   = glutton.ReadRAMRoute
 )
 
 // Server is an httptest-backed stand-in for a glutton actor holding one file.

internal/benchmarking/glutton/glutton.go

@@ -0,0 +1,44 @@
+// 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 implements the benchmarking workload served by
+// cmd/benchmarking/glutton: an API for consuming RAM, disk, and file
+// descriptors, and for gossiping with other glutton instances, over either
+// gRPC or protobuf-over-HTTP. See internal/proto/glutton/glutton.proto.
+package glutton
+
+// Name is the workload's identity to the telemetry SDK: the tracer and meter
+// scope, and the service name the binary reports. Instrument names are written
+// out in full at each call site so they match docs/metrics/registry/metrics.yaml
+// literally.
+const Name = "glutton"
+
+// Wire modes the main listener can serve, selected by the binary's --mode flag.
+const (
+	ModeGRPC = "grpc"
+	ModeHTTP = "http"
+)
+
+// Routes the HTTP-mode mux serves. ReadyzRoute is served in both modes: ateom
+// blocks RestoreWorkload until it answers 200, so ResumeActor cannot report
+// success before the listener is reachable. The fake in ./fake aliases these,
+// which is what keeps the stand-in and the real mux from drifting apart.
+const (
+	ReadyzRoute    = "/readyz"
+	PingRoute      = "/ping"
+	WriteDiskRoute = "/writedisk"
+	ReadDiskRoute  = "/readdisk"
+	WriteRAMRoute  = "/writeram"
+	ReadRAMRoute   = "/readram"
+)

internal/benchmarking/glutton/gossip.go

@@ -0,0 +1,151 @@
+// 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"
+	"errors"
+	"log/slog"
+	"time"
+
+	"github.com/google/uuid"
+	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/status"
+
+	gluttonpb "github.com/agent-substrate/substrate/internal/proto/glutton"
+)
+
+type peerGossip struct {
+	host    string
+	delayMs int32
+	cancel  context.CancelFunc
+	done    chan struct{}
+}
+
+// Sends network traffic to a peer glutton. Messages will be sent
+// on regular intervals separated by delay_ms.
+func (s *Service) Gossip(_ context.Context, req *gluttonpb.GossipRequest) (*gluttonpb.GossipResponse, error) {
+	want := make(map[string]*gluttonpb.Peer, len(req.GetPeers()))
+	for _, p := range req.GetPeers() {
+		if p.GetHost() == "" {
+			return nil, status.Error(codes.InvalidArgument, "peer host is required")
+		}
+		if p.GetDelayMs() <= 0 {
+			return nil, status.Errorf(codes.InvalidArgument, "peer %q delay_ms must be positive", p.GetHost())
+		}
+		want[p.GetHost()] = p
+	}
+
+	s.mu.Lock()
+	var toStop []*peerGossip
+	for host, existing := range s.peers {
+		w, ok := want[host]
+		if !ok || w.GetDelayMs() != existing.delayMs {
+			toStop = append(toStop, existing)
+			delete(s.peers, host)
+		}
+	}
+	var toStart []*gluttonpb.Peer
+	for host, w := range want {
+		if _, ok := s.peers[host]; !ok {
+			toStart = append(toStart, w)
+		}
+	}
+	s.mu.Unlock()
+
+	for _, p := range toStop {
+		p.cancel()
+		<-p.done
+	}
+
+	for _, w := range toStart {
+		gctx, cancel := context.WithCancel(context.Background())
+		pg := &peerGossip{
+			host:    w.GetHost(),
+			delayMs: w.GetDelayMs(),
+			cancel:  cancel,
+			done:    make(chan struct{}),
+		}
+		s.mu.Lock()
+		s.peers[w.GetHost()] = pg
+		s.mu.Unlock()
+		go s.runGossip(gctx, pg)
+	}
+
+	return &gluttonpb.GossipResponse{}, nil
+}
+
+func (s *Service) runGossip(ctx context.Context, pg *peerGossip) {
+	defer close(pg.done)
+
+	// grpc.NewClient resolves and connects lazily; the first RPC surfaces
+	// any failure, so the peer doesn't have to be reachable at start time.
+	conn, err := grpc.NewClient(pg.host,
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithStatsHandler(otelgrpc.NewClientHandler()),
+	)
+	if err != nil {
+		slog.ErrorContext(ctx, "Failed to dial gossip peer", slog.String("host", pg.host), slog.Any("err", err))
+		return
+	}
+	defer conn.Close()
+	client := gluttonpb.NewGluttonClient(conn)
+
+	hostAttr := attribute.String("host", pg.host)
+	ticker := time.NewTicker(time.Duration(pg.delayMs) * time.Millisecond)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+		}
+		msg := uuid.NewString()
+		start := time.Now()
+		resp, err := client.Ping(ctx, &gluttonpb.PingRequest{Message: msg})
+		latency := time.Since(start).Seconds()
+		outcome := "ok"
+		cancelled := err != nil && errors.Is(ctx.Err(), context.Canceled)
+		switch {
+		case cancelled:
+			outcome = "cancelled"
+		case err != nil:
+			outcome = "error"
+		}
+		attrs := metric.WithAttributes(hostAttr, attribute.String("outcome", outcome))
+		s.gossipSent.Add(ctx, 1, attrs)
+		s.gossipLatency.Record(ctx, latency, attrs)
+		if cancelled {
+			return
+		}
+		if err != nil {
+			slog.WarnContext(ctx, "Gossip ping failed", slog.String("host", pg.host), slog.Any("err", err))
+			continue
+		}
+		if resp.GetMessage() != msg {
+			slog.WarnContext(ctx, "Gossip ping returned unexpected message",
+				slog.String("host", pg.host),
+				slog.String("sent", msg),
+				slog.String("received", resp.GetMessage()),
+			)
+		}
+	}
+}

internal/benchmarking/glutton/metrics.go

@@ -0,0 +1,120 @@
+// 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"
+
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
+)
+
+// initMetrics builds every instrument the service reports and stores the
+// synchronous ones on s. The two observable gauges stay local: only the
+// callback registered here reads them.
+func (s *Service) initMetrics() error {
+	m := otel.Meter(Name)
+
+	var err error
+	s.ramWriteBytes, err = m.Int64Counter(
+		"glutton.ram.write.bytes",
+		metric.WithUnit("By"),
+		metric.WithDescription("Total bytes written to RAM via WriteRAM over the process lifetime."),
+	)
+	if err != nil {
+		return fmt.Errorf("create glutton.ram.write.bytes counter: %w", err)
+	}
+	s.ramReadBytes, err = m.Int64Counter(
+		"glutton.ram.read.bytes",
+		metric.WithUnit("By"),
+		metric.WithDescription("Total bytes walked by ReadRAM over the process lifetime."),
+	)
+	if err != nil {
+		return fmt.Errorf("create glutton.ram.read.bytes counter: %w", err)
+	}
+	s.diskWriteBytes, err = m.Int64Counter(
+		"glutton.disk.write.bytes",
+		metric.WithUnit("By"),
+		metric.WithDescription("Total bytes written to disk via WriteDisk over the process lifetime."),
+	)
+	if err != nil {
+		return fmt.Errorf("create glutton.disk.write.bytes counter: %w", err)
+	}
+	s.diskReadBytes, err = m.Int64Counter(
+		"glutton.disk.read.bytes",
+		metric.WithUnit("By"),
+		metric.WithDescription("Total bytes read from disk via ReadDisk over the process lifetime."),
+	)
+	if err != nil {
+		return fmt.Errorf("create glutton.disk.read.bytes counter: %w", err)
+	}
+	s.pingsReceived, err = m.Int64Counter(
+		"glutton.ping.requests",
+		metric.WithDescription("Number of Ping requests received."),
+	)
+	if err != nil {
+		return fmt.Errorf("create glutton.ping.requests counter: %w", err)
+	}
+	s.gossipSent, err = m.Int64Counter(
+		"glutton.gossip.requests.sent",
+		metric.WithDescription("Number of gossip Ping requests sent per peer."),
+	)
+	if err != nil {
+		return fmt.Errorf("create glutton.gossip.requests.sent counter: %w", err)
+	}
+	s.gossipLatency, err = m.Float64Histogram(
+		"glutton.gossip.latency",
+		metric.WithUnit("s"),
+		metric.WithDescription("Latency of gossip Ping requests per peer."),
+		metric.WithExplicitBucketBoundaries(
+			0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10,
+		),
+	)
+	if err != nil {
+		return fmt.Errorf("create glutton.gossip.latency histogram: %w", err)
+	}
+
+	fdsOpen, err := m.Int64ObservableGauge(
+		"glutton.fds.open",
+		metric.WithDescription("File descriptors currently held open by OpenFD."),
+	)
+	if err != nil {
+		return fmt.Errorf("create glutton.fds.open gauge: %w", err)
+	}
+	peerDelay, err := m.Int64ObservableGauge(
+		"glutton.gossip.delay",
+		metric.WithUnit("ms"),
+		metric.WithDescription("Configured gossip delay per peer."),
+	)
+	if err != nil {
+		return fmt.Errorf("create glutton.gossip.delay gauge: %w", err)
+	}
+
+	if _, err := m.RegisterCallback(func(_ context.Context, o metric.Observer) error {
+		s.mu.Lock()
+		defer s.mu.Unlock()
+		o.ObserveInt64(fdsOpen, int64(len(s.fds)))
+		for host, p := range s.peers {
+			o.ObserveInt64(peerDelay, int64(p.delayMs), metric.WithAttributes(attribute.String("host", host)))
+		}
+		return nil
+	}, fdsOpen, peerDelay); err != nil {
+		return fmt.Errorf("register glutton observable callback: %w", err)
+	}
+
+	return nil
+}