Instrument both workercache and valkey store with metrics.

cmd/ateapi/internal/controlapi/functional_test.go

@@ -253,7 +253,11 @@ func setupTest(t *testing.T, ns string) *testContext {
 	rdb := redis.NewClusterClient(&redis.ClusterOptions{
 		Addrs: []string{mr.Addr()},
 	})
-	persistence := ateredis.NewPersistence(rdb)
+	persistence, err := ateredis.NewPersistence(rdb)
+	if err != nil {
+		mr.Close()
+		t.Fatalf("ateredis.NewPersistence: %v", err)
+	}
 
 	// 2. Initialize Clientsets using global cfg
 	k8sClient, err := kubernetes.NewForConfig(cfg)
@@ -291,7 +295,12 @@ func setupTest(t *testing.T, ns string) *testContext {
 	substrateInformerFactory.WaitForCacheSync(ctx.Done())
 
 	// 4. Initialize Service
-	wc := workercache.New(persistence, 5*time.Minute)
+	wc, err := workercache.New(persistence, 5*time.Minute)
+	if err != nil {
+		cancel()
+		mr.Close()
+		t.Fatalf("workercache.New: %v", err)
+	}
 	if err := wc.Start(ctx); err != nil {
 		cancel()
 		mr.Close()

cmd/ateapi/internal/store/ateredis/ateredis.go

@@ -56,6 +56,9 @@ import (
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store"
 	"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
 	"github.com/redis/go-redis/v9"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
 	"google.golang.org/protobuf/encoding/protojson"
 	"google.golang.org/protobuf/proto"
 )
@@ -74,16 +77,25 @@ type redisClient interface {
 
 // Persistence is a service that stores information about applications in Redis.
 type Persistence struct {
-	rdb redisClient
+	rdb                        redisClient
+	metricWorkerPubsubMessages metric.Int64Counter
 }
 
 var _ store.Interface = (*Persistence)(nil)
 
 // NewPersistence creates a new Persistence.
-func NewPersistence(redisClient *redis.ClusterClient) *Persistence {
-	return &Persistence{
-		rdb: redisClient,
+func NewPersistence(redisClient *redis.ClusterClient) (*Persistence, error) {
+	publishMessages, err := otel.Meter("ateredis").Int64Counter(
+		"persistence.worker.pubsub.messages",
+		metric.WithUnit("{message}"),
+		metric.WithDescription("Total worker-changes pub/sub PUBLISH attempts."))
+	if err != nil {
+		return nil, fmt.Errorf("create persistence.worker.pubsub.messages counter: %w", err)
 	}
+	return &Persistence{
+		rdb:                        redisClient,
+		metricWorkerPubsubMessages: publishMessages,
+	}, nil
 }
 
 func actorDBKey(id string) string {
@@ -128,6 +140,9 @@ func (s *Persistence) publishWorkerEvent(ctx context.Context, eventType store.Wo
 	}
 	if err := s.rdb.Publish(ctx, workerPubSubChannel, payload).Err(); err != nil {
 		slog.ErrorContext(ctx, "worker event publish failed", slog.Any("err", err))
+		s.metricWorkerPubsubMessages.Add(ctx, 1, metric.WithAttributes(attribute.String("error.type", "_OTHER")))
+	} else {
+		s.metricWorkerPubsubMessages.Add(ctx, 1)
 	}
 }
 

cmd/ateapi/internal/store/ateredis/ateredis_test.go

@@ -42,7 +42,11 @@ func setupTest(t *testing.T) (*miniredis.Miniredis, *Persistence, context.Contex
 	rdb := redis.NewClusterClient(&redis.ClusterOptions{
 		Addrs: []string{mr.Addr()},
 	})
-	return mr, &Persistence{rdb: rdb}, context.Background()
+	p, err := NewPersistence(rdb)
+	if err != nil {
+		t.Fatalf("NewPersistence: %v", err)
+	}
+	return mr, p, context.Background()
 }
 
 func TestGetActor_NotFound(t *testing.T) {

cmd/ateapi/internal/store/storetest/storetest.go

@@ -37,7 +37,10 @@ func SetupTestStore(t *testing.T) (store.Interface, func()) {
 		Addrs: []string{mr.Addr()},
 	})
 
-	persistence := ateredis.NewPersistence(rdb)
+	persistence, err := ateredis.NewPersistence(rdb)
+	if err != nil {
+		t.Fatalf("ateredis.NewPersistence: %v", err)
+	}
 
 	cleanup := func() {
 		mr.Close()

cmd/ateapi/internal/workercache/workercache.go

@@ -28,13 +28,13 @@ import (
 
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store"
 	"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
 	"k8s.io/apimachinery/pkg/util/wait"
 )
 
 // Cache maintains an in-memory snapshot of all workers.
-//
-// TODO: add metrics — at minimum a gauge for worker count, a counter for
-// resync events, and a counter for failed PUBLISH operations (in ateredis).
 type Cache struct {
 	store          store.Interface
 	relistInterval time.Duration
@@ -43,17 +43,53 @@ type Cache struct {
 	workers map[string]*ateapipb.Worker
 
 	ready atomic.Bool
+
+	metricWorkerCount      metric.Int64Gauge
+	metricNotReadyDuration metric.Float64Counter
+	metricResyncs          metric.Int64Counter
+	metricRelists          metric.Int64Counter
 }
 
 // New creates a Cache backed by a given store. relistInterval controls how
 // often the cache performs a full ListWorkers to recover from state drifts
 // caused by missing WorkerWatch events.
-func New(store store.Interface, relistInterval time.Duration) *Cache {
-	return &Cache{
-		store:          store,
+func New(s store.Interface, relistInterval time.Duration) (*Cache, error) {
+	c := &Cache{
+		store:          s,
 		relistInterval: relistInterval,
 		workers:        make(map[string]*ateapipb.Worker),
 	}
+	m := otel.Meter("workercache")
+	var err error
+	c.metricWorkerCount, err = m.Int64Gauge(
+		"cache.worker.count",
+		metric.WithUnit("{worker}"),
+		metric.WithDescription("Current number of workers in the cache."))
+	if err != nil {
+		return nil, fmt.Errorf("create cache.worker.count gauge failed: %w", err)
+	}
+	c.metricNotReadyDuration, err = m.Float64Counter(
+		"cache.not_ready.duration",
+		metric.WithUnit("s"),
+		metric.WithDescription("Total time the worker cache spent not ready, between a watch disconnect and a successful resync."))
+	if err != nil {
+		return nil, fmt.Errorf("create cache.not_ready.duration counter failed: %w", err)
+	}
+	c.metricResyncs, err = m.Int64Counter(
+		"cache.resyncs",
+		metric.WithUnit("{resync}"),
+		metric.WithDescription("Total full resyncs triggered by watch disconnects."))
+	if err != nil {
+		return nil, fmt.Errorf("create cache.resyncs counter failed: %w", err)
+	}
+	c.metricRelists, err = m.Int64Counter(
+		"cache.relists",
+		metric.WithUnit("{relist}"),
+		metric.WithDescription("Total relists (initial, resync, and periodic)."))
+	if err != nil {
+		return nil, fmt.Errorf("create cache.relists counter failed: %w", err)
+	}
+	return c, nil
 }
 
 // Start syncs the cache synchronously, then spawns a background goroutine
@@ -101,6 +137,7 @@ func (c *Cache) sync(ctx context.Context) (*store.WorkerWatch, error) {
 func (c *Cache) relist(ctx context.Context) error {
 	workers, err := c.store.ListWorkers(ctx)
 	if err != nil {
+		c.metricRelists.Add(ctx, 1, metric.WithAttributes(attribute.String("error.type", "_OTHER")))
 		return fmt.Errorf("ListWorkers: %w", err)
 	}
 	newMap := make(map[string]*ateapipb.Worker, len(workers))
@@ -109,8 +146,11 @@ func (c *Cache) relist(ctx context.Context) error {
 	}
 	c.mu.Lock()
 	c.workers = newMap
+	count := int64(len(newMap))
 	c.mu.Unlock()
-	slog.InfoContext(ctx, "worker cache synced", slog.Int("count", len(newMap)))
+	c.metricWorkerCount.Record(ctx, count)
+	c.metricRelists.Add(ctx, 1)
+	slog.InfoContext(ctx, "worker cache synced", slog.Int("count", int(count)))
 	return nil
 }
 
@@ -122,6 +162,7 @@ func (c *Cache) watchEvents(ctx context.Context, watch *store.WorkerWatch) {
 		case event, ok := <-watch.Events:
 			if !ok {
 				c.ready.Store(false)
+				notReadySince := time.Now()
 				watch.Close()
 				if ctx.Err() != nil {
 					return
@@ -132,8 +173,9 @@ func (c *Cache) watchEvents(ctx context.Context, watch *store.WorkerWatch) {
 					return // context cancelled
 				}
 				c.ready.Store(true)
+				c.metricNotReadyDuration.Add(ctx, time.Since(notReadySince).Seconds())
 			} else {
-				c.applyEvent(event)
+				c.applyEvent(ctx, event)
 			}
 		case <-ticker.C:
 			if err := c.relist(ctx); err != nil {
@@ -148,6 +190,7 @@ func (c *Cache) watchEvents(ctx context.Context, watch *store.WorkerWatch) {
 }
 
 func (c *Cache) resync(ctx context.Context) *store.WorkerWatch {
+	c.metricResyncs.Add(ctx, 1)
 	backoff := wait.Backoff{
 		Duration: time.Second,
 		Factor:   2.0,
@@ -167,10 +210,9 @@ func (c *Cache) resync(ctx context.Context) *store.WorkerWatch {
 	return watch
 }
 
-func (c *Cache) applyEvent(event store.WorkerEvent) {
+func (c *Cache) applyEvent(ctx context.Context, event store.WorkerEvent) {
 	key := workerKey(event.Worker)
 	c.mu.Lock()
-	defer c.mu.Unlock()
 	switch event.Type {
 	case store.WorkerEventDeleted:
 		delete(c.workers, key)
@@ -180,6 +222,9 @@ func (c *Cache) applyEvent(event store.WorkerEvent) {
 			c.workers[key] = event.Worker
 		}
 	}
+	count := int64(len(c.workers))
+	c.mu.Unlock()
+	c.metricWorkerCount.Record(ctx, count)
 }
 
 func workerKey(w *ateapipb.Worker) string {