atelet: emit per-actor usage events from the stats sweep (#1206)

Part of #896 (Phase 1 of #550): the events channel — per-actor usage
samples as structured log events. With this, both halves of #174's
cardinality split exist: template-level metrics in the TSDB (#961), and
everything carrying actor/atespace identity here, in the log store.

## The events

One JSON record per executing actor per sweep on atelet's stdout, riding
the poller's existing probe — no extra RPC load, and one knob governs
both channels (`--actor-stats-poll-interval 0` disables the subsystem).
Idle workers emit nothing: an idle fleet is silent by design.

Records use the same label vocabulary as actorlog's lifecycle events
(`ateattr.ActorLogLabels`, including the GCE
`logging.googleapis.com/labels` spelling) — so one Cloud Logging filter
on `labels."ate.actor.uid"` returns an actor's container logs, lifecycle
transitions, and usage samples interleaved. Identity comes solely from
each sample's echo, per the stats RPCs' attribution contract.
Measurements ride as payload fields (`kind`, class, source, the four
numbers, `observed_at_unix_nano`); the `kind` field (today always
`periodic`) lets future kinds join without reshaping the record.

Events also carry the `ate.workerpool.namespace/name` pair the metric
labels already carry, resolved by the sweep's own pod list, so a
pool-level metric spike can pivot to the actors behind it — pool
membership lives on the worker pod and is unrecoverable from logs once
the pod is gone, so it must be stamped at emission. An unresolved pod
emits without the pair, following the metric channel's rule.

## Scope

Earlier revisions also took first/final lifecycle samples from the
Run/Restore/Checkpoint handlers. Per the review discussion on
suspend/resume latency, those are descoped from this PR: it now touches
no lifecycle path at all. The bracket design (including taking the final
sample inside ateom's CheckpointWorkload and echoing it in the response)
moves to a follow-up under #896.

## Isolation

The poller dials its own short-lived connection per probe
(`dialAteomStats`) and never touches the lifecycle RPCs' cached clients
— after review on #961 the isolation is structural.

## Validated live on ate-dev

Periodic events read back from Cloud Logging via
`labels."ate.actor.name"`, one per executing actor per minute, with
`ate.actor.uid` confirmed promoted into `LogEntry.labels` (filterable) —
the claim only production could prove.

Consumer note: aggregate freely in the log store, but a log-based metric
built over these events must label only by the bounded set (template,
class, source, pool) — promoting actor identity into a metric label
would reintroduce exactly the cardinality #174 keeps out of the TSDB.

Part of #896. Part of #550.

cmd/atelet/main.go

@@ -35,6 +35,7 @@ import (
 	"sync"
 
 	"github.com/agent-substrate/substrate/cmd/atelet/internal/ategcs"
+	"github.com/agent-substrate/substrate/internal/actorlog"
 	"github.com/agent-substrate/substrate/internal/ateapiauth"
 	"github.com/agent-substrate/substrate/internal/ateattr"
 	"github.com/agent-substrate/substrate/internal/ateerrors"
@@ -118,7 +119,13 @@ func main() {
 		return
 	}
 	ctx := context.Background()
-	serverboot.InitLogger()
+	// One synchronized writer in front of stdout, shared by the runtime
+	// logger and the usage-event drain (see startStatsPoller): uncoordinated
+	// writers stay tear-free only while every record fits a pipe's
+	// atomic-write size -- an accident of field sizes, not a contract. Same
+	// pattern as the ateoms' actor-log forwarders.
+	logSink := actorlog.NewSyncedWriter(os.Stdout)
+	serverboot.InitLoggerWithWriter(logSink)
 	if err := serverboot.SetLogLevel(*logLevelFlag); err != nil {
 		serverboot.Fatal(ctx, "Invalid --log-level", err)
 	}
@@ -257,7 +264,7 @@ func main() {
 			// crash-looping every actor operation on the node.
 			slog.ErrorContext(ctx, "Actor stats sampling disabled: failed to create instruments", slog.Any("err", err))
 		} else {
-			startStatsPoller(ctx, interval, statsInst, k8sClient)
+			startStatsPoller(ctx, interval, statsInst, k8sClient, logSink)
 		}
 	}
 
@@ -294,6 +301,7 @@ func main() {
 		csiDriverConfigLister,
 		clusterTrustBundleLister,
 	)
+
 	// Pre-download sandbox assets as SandboxConfigs appear/change so the first
 	// Run/Restore on this node hits the cache. Best-effort: on failure the
 	// on-demand fetch in ensureSandboxAssets still covers correctness.

cmd/atelet/statsevents.go

@@ -0,0 +1,184 @@
+// 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 main
+
+import (
+	"context"
+	"io"
+	"log/slog"
+	"sync"
+	"sync/atomic"
+
+	"cloud.google.com/go/compute/metadata"
+
+	"github.com/agent-substrate/substrate/internal/actorlog"
+	"github.com/agent-substrate/substrate/internal/ateattr"
+	"github.com/agent-substrate/substrate/internal/contextlogging"
+	"github.com/agent-substrate/substrate/internal/proto/ateompb"
+	"github.com/agent-substrate/substrate/internal/resources"
+)
+
+// The events channel is the per-actor half of the usage telemetry split: the
+// metrics (statspoller.go) aggregate to the bounded template-level label set,
+// and everything with actor or atespace identity travels here instead, as
+// structured log events -- the same stream and label vocabulary as actorlog's
+// lifecycle events, never a TSDB series.
+
+// usageSampleMsg is the message every usage event carries; consumers filter on
+// it plus the "kind" field.
+const usageSampleMsg = "Actor usage sample"
+
+// eventKindPeriodic marks the events riding the poller's sweep -- today the
+// only kind, carried on the wire so future kinds (lifecycle brackets, say)
+// can join without reshaping the record.
+const eventKindPeriodic = "periodic"
+
+// defaultLabelsKey resolves the actor-identity label group's spelling --
+// actorlog's, so usage events and lifecycle events promote into Cloud
+// Logging labels the same way. metadata.OnGCE probes the metadata server, so
+// startStatsPoller warms this off the boot path on a throwaway goroutine;
+// sync.OnceValue makes an emit that arrives first wait for the in-flight
+// probe. That wait is first-tick-only and bounded: milliseconds on GCE, the
+// transport's 2s dial timeout off GCE, a 5s cap on the one pathological
+// branch (SMBIOS says GCE, probes disagree) -- and no request path ever
+// touches it.
+var defaultLabelsKey = sync.OnceValue(func() string {
+	return actorlog.LabelsKey(metadata.OnGCE())
+})
+
+// statsEventEmitter writes per-actor usage events to the process log stream.
+// A nil emitter is a valid no-op, so call sites need no guard.
+type statsEventEmitter struct {
+	log       *slog.Logger
+	labelsKey func() string
+}
+
+// newStatsEventEmitter builds an emitter over its own fixed-level handler on
+// w rather than the serverboot logger: these records are a data feed, not
+// leveled diagnostics, and quieting the node with --log-level=warn must not
+// silently sever them -- the subsystem's one off-switch is
+// --actor-stats-poll-interval=0. The records still carry level INFO on the
+// wire, so nothing downstream changes. In production w is an asyncWriter
+// over the process's synchronized stdout writer (see startStatsPoller).
+func newStatsEventEmitter(w io.Writer, labelsKey func() string) *statsEventEmitter {
+	return &statsEventEmitter{
+		log:       slog.New(contextlogging.NewHandler(slog.NewJSONHandler(w, nil))),
+		labelsKey: labelsKey,
+	}
+}
+
+// usageEventQueueDepth is the asyncWriter's buffer, in records: about one
+// tick of a packed node. A healthy drain outruns the RPC-paced producers by
+// orders of magnitude, so the queue only fills once the pipe is dead --
+// where any finite depth drops, and the choice is cosmetic.
+const usageEventQueueDepth = 256
+
+// asyncWriter decouples event emission from the log consumer's health. A
+// write to a full stdout pipe blocks forever, and emit runs inside the
+// sweep's errgroup, so one stalled log consumer (wedged rotation, disk-full
+// fallout) would otherwise park a probe, wedge the sweep, and silently
+// freeze the metrics channel that shares it. Writes land in a bounded queue
+// drained by one goroutine; a full queue drops the record and counts it.
+// Dropping is safe: the samples are point-in-time readings the next healthy
+// tick repairs, and when stdout is dead the events are lost either way --
+// the choice is whether the metrics die with them.
+type asyncWriter struct {
+	w  io.Writer
+	ch chan []byte
+
+	// report carries the drop warning over the same fixed-level pipeline as
+	// the records: the loss signal is part of the feed's integrity, so it
+	// must be exactly as unkillable by --log-level as the feed itself.
+	report *slog.Logger
+
+	dropped atomic.Int64
+}
+
+func newAsyncWriter(ctx context.Context, w io.Writer, depth int) *asyncWriter {
+	aw := &asyncWriter{
+		w:      w,
+		ch:     make(chan []byte, depth),
+		report: slog.New(contextlogging.NewHandler(slog.NewJSONHandler(w, nil))),
+	}
+	go aw.run(ctx)
+	return aw
+}
+
+func (aw *asyncWriter) run(ctx context.Context) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case b := <-aw.ch:
+			// Best-effort: a failed write has nobody to report to.
+			_, _ = aw.w.Write(b)
+			// Report drops only after a successful write, when the stream
+			// can carry the warning -- written directly, so the drain does
+			// not queue behind itself.
+			if n := aw.dropped.Swap(0); n > 0 {
+				aw.report.WarnContext(ctx, "Usage events dropped while the log stream stalled", slog.Int64("count", n))
+			}
+		}
+	}
+}
+
+// Write queues one record without ever blocking. It reports full success
+// even on a drop: the emitter has no recovery to offer, and the drop is
+// already counted for the writer goroutine to report.
+func (aw *asyncWriter) Write(p []byte) (int, error) {
+	// The slog handler reuses its buffer after Write returns, so the queue
+	// must own a copy.
+	b := append([]byte(nil), p...)
+	select {
+	case aw.ch <- b:
+	default:
+		aw.dropped.Add(1)
+	}
+	return len(p), nil
+}
+
+// emit writes one usage event. The identity comes solely from the sample's
+// echo, per the stats RPCs' attribution contract. pool is the caller's
+// pod-to-pool resolution -- the same enrichment the metric labels carry, so a
+// pool-level metric spike can pivot to the actors behind it. A zero-valued
+// pool omits the label pair rather than emitting empty strings, following the
+// metric channel's rule.
+func (e *statsEventEmitter) emit(ctx context.Context, kind string, s *ateompb.WorkloadStatsSample, pool workerPoolRef) {
+	if e == nil || s == nil {
+		return
+	}
+	a := resources.ActorAttribution{
+		Ref:              resources.ActorRef{Atespace: s.GetAtespace(), Name: s.GetActorName()},
+		UID:              s.GetActorUid(),
+		TemplateAtespace: s.GetActorTemplateAtespace(),
+		TemplateName:     s.GetActorTemplateName(),
+	}
+	labels := ateattr.ActorLogLabels(a, "")
+	if pool != (workerPoolRef{}) {
+		labels[string(ateattr.WorkerPoolNamespaceKey)] = pool.namespace
+		labels[string(ateattr.WorkerPoolNameKey)] = pool.name
+	}
+	e.log.LogAttrs(ctx, slog.LevelInfo, usageSampleMsg,
+		slog.Any(e.labelsKey(), labels),
+		slog.String("kind", kind),
+		slog.String("sandbox_class", sandboxClassLabel(s.GetSandboxClass())),
+		slog.String("source", statsSourceLabel(s.GetSource())),
+		slog.Uint64("memory_current_bytes", s.GetMemoryCurrentBytes()),
+		slog.Uint64("memory_peak_bytes", s.GetMemoryPeakBytes()),
+		slog.Uint64("memory_working_set_bytes", s.GetMemoryWorkingSetBytes()),
+		slog.Uint64("cpu_usage_usec", s.GetCpuUsageUsec()),
+		slog.Int64("observed_at_unix_nano", s.GetObservedAtUnixNano()),
+	)
+}

cmd/atelet/statsevents_test.go

@@ -0,0 +1,307 @@
+// 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 main
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/agent-substrate/substrate/internal/actorlog"
+	"github.com/agent-substrate/substrate/internal/proto/ateompb"
+	"github.com/agent-substrate/substrate/internal/serverboot"
+)
+
+// eventSample is the fully-populated sample the event tests emit; distinct
+// values per field so a crossed wire is visible in the JSON.
+func eventSample() *ateompb.WorkloadStatsSample {
+	return &ateompb.WorkloadStatsSample{
+		Atespace:              "space-a",
+		ActorName:             "actor-a",
+		ActorUid:              "uid-a",
+		ActorTemplateAtespace: "ns-a",
+		ActorTemplateName:     "template-a",
+		SandboxClass:          ateompb.SandboxClass_SANDBOX_CLASS_MICROVM,
+		Source:                ateompb.StatsSource_STATS_SOURCE_GUEST_AGENT,
+		MemoryCurrentBytes:    1000,
+		MemoryPeakBytes:       2000,
+		MemoryWorkingSetBytes: 700,
+		CpuUsageUsec:          1234,
+		ObservedAtUnixNano:    42,
+	}
+}
+
+// syncBuffer is a mutex-guarded buffer, so tests stay valid if an emitter
+// call site ever moves onto a goroutine.
+type syncBuffer struct {
+	mu  sync.Mutex
+	buf bytes.Buffer
+}
+
+func (b *syncBuffer) Write(p []byte) (int, error) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.buf.Write(p)
+}
+
+func (b *syncBuffer) Bytes() []byte {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return append([]byte(nil), b.buf.Bytes()...)
+}
+
+func (b *syncBuffer) Len() int {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.buf.Len()
+}
+
+func (b *syncBuffer) String() string { return string(b.Bytes()) }
+
+// newBufferEmitter returns an emitter writing JSON records into buf. The
+// labels-key closure stands in for defaultLabelsKey, whose metadata-server
+// probe has no place in a unit test.
+func newBufferEmitter(buf *syncBuffer, isOnGCE bool) *statsEventEmitter {
+	return newStatsEventEmitter(buf, func() string { return actorlog.LabelsKey(isOnGCE) })
+}
+
+func TestStatsEventEmitterEmit(t *testing.T) {
+	var buf syncBuffer
+	e := newBufferEmitter(&buf, false)
+
+	e.emit(context.Background(), eventKindPeriodic, eventSample(), workerPoolRef{})
+
+	var rec map[string]any
+	if err := json.Unmarshal(buf.Bytes(), &rec); err != nil {
+		t.Fatalf("event is not one JSON record: %v (%q)", err, buf.String())
+	}
+	if got := rec["msg"]; got != usageSampleMsg {
+		t.Errorf("msg = %v, want %q", got, usageSampleMsg)
+	}
+	if got := rec["kind"]; got != "periodic" {
+		t.Errorf("kind = %v, want periodic", got)
+	}
+	// The identity travels as the actorlog-style label group, so usage events
+	// join lifecycle events and container logs under the same keys.
+	labels, ok := rec["labels"].(map[string]any)
+	if !ok {
+		t.Fatalf("labels group missing or wrong shape: %v", rec["labels"])
+	}
+	for key, want := range map[string]string{
+		"ate.atespace":          "space-a",
+		"ate.actor.name":        "actor-a",
+		"ate.actor.uid":         "uid-a",
+		"ate.template.atespace": "ns-a",
+		"ate.template.name":     "template-a",
+	} {
+		if got := labels[key]; got != want {
+			t.Errorf("labels[%q] = %v, want %q", key, got, want)
+		}
+	}
+	for key, want := range map[string]float64{
+		"memory_current_bytes":     1000,
+		"memory_peak_bytes":        2000,
+		"memory_working_set_bytes": 700,
+		"cpu_usage_usec":           1234,
+		"observed_at_unix_nano":    42,
+	} {
+		if got := rec[key]; got != want {
+			t.Errorf("%s = %v, want %v", key, got, want)
+		}
+	}
+	if got := rec["sandbox_class"]; got != "microvm" {
+		t.Errorf("sandbox_class = %v, want microvm", got)
+	}
+	if got := rec["source"]; got != "guest-agent" {
+		t.Errorf("source = %v, want guest-agent", got)
+	}
+	// An unresolved pool omits the pair rather than emitting empty strings.
+	for _, key := range []string{"ate.workerpool.namespace", "ate.workerpool.name"} {
+		if _, ok := labels[key]; ok {
+			t.Errorf("zero pool ref emitted label %q: %v", key, labels[key])
+		}
+	}
+}
+
+// TestStatsEventEmitterPoolLabels pins the pool enrichment on events: a
+// resolved pod's events carry the same ate.workerpool label pair as the
+// metric channel, inside the promoted label group.
+func TestStatsEventEmitterPoolLabels(t *testing.T) {
+	var buf syncBuffer
+	e := newBufferEmitter(&buf, false)
+
+	e.emit(context.Background(), eventKindPeriodic, eventSample(), workerPoolRef{namespace: "pool-ns", name: "pool-a"})
+
+	var rec map[string]any
+	if err := json.Unmarshal(buf.Bytes(), &rec); err != nil {
+		t.Fatalf("unmarshal: %v", err)
+	}
+	labels, ok := rec["labels"].(map[string]any)
+	if !ok {
+		t.Fatalf("labels group missing or wrong shape: %v", rec["labels"])
+	}
+	if got := labels["ate.workerpool.namespace"]; got != "pool-ns" {
+		t.Errorf("labels[ate.workerpool.namespace] = %v, want pool-ns", got)
+	}
+	if got := labels["ate.workerpool.name"]; got != "pool-a" {
+		t.Errorf("labels[ate.workerpool.name] = %v, want pool-a", got)
+	}
+}
+
+// TestStatsEventEmitterGCELabelsKey pins the label-group spelling split: on
+// GCE the group must sit under the key Cloud Logging promotes.
+func TestStatsEventEmitterGCELabelsKey(t *testing.T) {
+	var buf syncBuffer
+	e := newBufferEmitter(&buf, true)
+
+	e.emit(context.Background(), eventKindPeriodic, eventSample(), workerPoolRef{})
+
+	var rec map[string]any
+	if err := json.Unmarshal(buf.Bytes(), &rec); err != nil {
+		t.Fatalf("unmarshal: %v", err)
+	}
+	if _, ok := rec["logging.googleapis.com/labels"]; !ok {
+		t.Errorf("GCE emitter did not use the promoted labels key; record keys: %v", buf.String())
+	}
+}
+
+// TestStatsEventEmitterIgnoresLogLevel pins the emitter's independence from
+// the serverboot verbosity knob: usage events are a data feed, and quieting
+// the node with --log-level=warn (or error) must not silently sever them.
+func TestStatsEventEmitterIgnoresLogLevel(t *testing.T) {
+	if err := serverboot.SetLogLevel("error"); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		if err := serverboot.SetLogLevel("info"); err != nil {
+			t.Fatal(err)
+		}
+	})
+
+	var buf syncBuffer
+	newBufferEmitter(&buf, false).emit(context.Background(), eventKindPeriodic, eventSample(), workerPoolRef{})
+
+	if buf.Len() == 0 {
+		t.Error("raising the serverboot log level suppressed a usage event")
+	}
+}
+
+// TestStatsEventEmitterNil: a nil emitter and a nil sample are both valid
+// no-ops, so call sites stay unconditional.
+func TestStatsEventEmitterNil(t *testing.T) {
+	var e *statsEventEmitter
+	e.emit(context.Background(), eventKindPeriodic, eventSample(), workerPoolRef{}) // must not panic
+
+	var buf syncBuffer
+	newBufferEmitter(&buf, false).emit(context.Background(), eventKindPeriodic, nil, workerPoolRef{})
+	if buf.Len() != 0 {
+		t.Errorf("nil sample emitted a record: %q", buf.String())
+	}
+}
+
+// stallableWriter blocks every Write until released -- the stalled log
+// consumer the asyncWriter exists for.
+type stallableWriter struct {
+	gate  chan struct{}
+	wrote syncBuffer
+}
+
+func (w *stallableWriter) Write(p []byte) (int, error) {
+	<-w.gate
+	return w.wrote.Write(p)
+}
+
+// TestAsyncWriterNeverBlocks pins the contract: with the underlying
+// writer wedged, every Write returns immediately -- overflow drops and
+// counts instead of blocking -- and releasing the wedge drains what the
+// queue held.
+func TestAsyncWriterNeverBlocks(t *testing.T) {
+	under := &stallableWriter{gate: make(chan struct{})}
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+	const depth = 4
+	aw := newAsyncWriter(ctx, under, depth)
+
+	// The writer goroutine dequeues one record and wedges on it; the queue
+	// holds depth more. Everything past that must drop, not block. The
+	// in-flight handoff is asynchronous, so tolerate one record of slack.
+	const writes = depth + 8
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		for i := 0; i < writes; i++ {
+			if _, err := aw.Write([]byte("x")); err != nil {
+				t.Errorf("Write returned %v, want nil", err)
+			}
+		}
+	}()
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		t.Fatal("Write blocked on a wedged log stream")
+	}
+	dropped := int(aw.dropped.Load())
+	if dropped < writes-depth-1 || dropped > writes-depth {
+		t.Errorf("dropped = %d, want %d or %d", dropped, writes-depth-1, writes-depth)
+	}
+
+	// Conservation: everything not dropped -- the wedged in-flight record
+	// plus the queue -- drains once the stream recovers, and the drop report
+	// arrives over the same sink, unkillable by any verbosity knob.
+	close(under.gate)
+	wantWritten := writes - dropped
+	waitFor(t, func() bool { return bytes.Count(under.wrote.Bytes(), []byte("x")) >= wantWritten })
+	if got := bytes.Count(under.wrote.Bytes(), []byte("x")); got != wantWritten {
+		t.Errorf("drained %d records, want %d", got, wantWritten)
+	}
+	waitFor(t, func() bool { return bytes.Contains(under.wrote.Bytes(), []byte("Usage events dropped")) })
+}
+
+// TestAsyncWriterCopies: the slog handler reuses its buffer after
+// Write returns, so the queue must hold copies, not aliases.
+func TestAsyncWriterCopies(t *testing.T) {
+	under := &stallableWriter{gate: make(chan struct{})}
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+	aw := newAsyncWriter(ctx, under, 4)
+
+	p := []byte("original")
+	if _, err := aw.Write(p); err != nil {
+		t.Fatal(err)
+	}
+	copy(p, "clobber!")
+
+	close(under.gate)
+	waitFor(t, func() bool { return under.wrote.Len() > 0 })
+	if got := under.wrote.String(); got != "original" {
+		t.Errorf("record = %q, want the copy taken at Write time", got)
+	}
+}
+
+// waitFor polls until cond returns true or the deadline passes.
+func waitFor(t *testing.T, cond func() bool) {
+	t.Helper()
+	deadline := time.Now().Add(5 * time.Second)
+	for time.Now().Before(deadline) {
+		if cond() {
+			return
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	t.Fatal("condition not reached before deadline")
+}

cmd/atelet/statspoller.go

@@ -111,25 +111,28 @@ type statsPoller struct {
 	ateomsDir string
 
 	// dial returns a stats client for one ateom plus the closer that releases
-	// its connection; the probe closes it before returning, so a connection
-	// lives exactly one probe. Deliberately NOT the lifecycle RPCs' cached
-	// AteomDialer: at one probe per ateom per minute over a local unix socket
-	// a cache saves nothing, and sweeping the node's stale sockets through a
-	// shared cache would let telemetry evict connections the lifecycle RPCs
-	// are using.
+	// its connection; a connection lives exactly one probe. Deliberately NOT
+	// the lifecycle RPCs' cached AteomDialer: at one probe per ateom per
+	// minute over a local unix socket a cache saves nothing, and sweeping the
+	// node's stale sockets through a shared cache would let telemetry evict
+	// connections the lifecycle RPCs are using.
 	dial func(ctx context.Context, podUID string) (activeStatsClient, io.Closer, error)
 
 	// workerPools resolves this node's worker pod UIDs to the pool that owns
 	// them, called once per sweep. Nil (or a nil map, or a missing entry)
-	// degrades to samples grouped without pool labels rather than dropped: the
-	// pool is enrichment, the sample is the point. The real resolver lists the
-	// node's pods by the ate.dev/worker-pool label the pool controller stamps
-	// on every worker (see workerpool_apply.go); the ateom directory name IS
+	// degrades to samples grouped without pool labels rather than dropped:
+	// the pool is enrichment, the sample is the point. The real resolver
+	// lists the node's pods by workerPoolLabel; the ateom directory name IS
 	// the worker pod UID, which is the join key.
 	workerPools func(ctx context.Context) map[string]workerPoolRef
 
 	inst *statsInstruments
 
+	// eventEmitter receives one usage event per sample per sweep -- the
+	// per-actor channel the aggregates deliberately erase identity from.
+	// Nil disables emission; emit is nil-safe.
+	eventEmitter *statsEventEmitter
+
 	// lastCPU is the previous sweep's cpu_usage_usec per actor uid, the
 	// baseline the next sweep's deltas are computed against. Only the sweep
 	// loop touches it (under collect's mutex), and entries for actors a sweep
@@ -140,10 +143,10 @@ type statsPoller struct {
 	lastCPU map[string]uint64
 }
 
-// templateAggregate is one tick's sums for one templateKey group: the bounded
-// label set #174 permits on a TSDB series. Actor and
-// atespace identity deliberately never reach a metric label; per-actor detail
-// is the events channel's job, not this one's.
+// templateAggregate is one tick's sums for one templateKey group: the
+// bounded label set #174 permits on a TSDB series. Actor and atespace
+// identity deliberately never reach a metric label; per-actor detail is the
+// events channel's job, not this one's.
 //
 // The memory fields are point-in-time sums the gauges observe. cpuDeltaUsec is
 // different: cpu_usage_usec is a cumulative per-epoch counter per actor, so
@@ -198,7 +201,7 @@ func (k templateKey) attrs() metric.MeasurementOption {
 	return metric.WithAttributes(attrs...)
 }
 
-// run polls until ctx is cancelled. The caller has already validated and
+// run polls until ctx is canceled. The caller has already validated and
 // clamped interval.
 func (p *statsPoller) run(ctx context.Context) {
 	slog.InfoContext(ctx, "Actor stats poller starting", slog.Duration("interval", p.interval))
@@ -286,6 +289,7 @@ func (p *statsPoller) collect(ctx context.Context) map[templateKey]*templateAggr
 				// add.
 				return nil
 			}
+			p.eventEmitter.emit(ctx, eventKindPeriodic, sample, pools[podUID])
 
 			key := templateKey{
 				templateNamespace: sample.GetActorTemplateAtespace(),
@@ -401,9 +405,17 @@ func nodeWorkerPools(client kubernetes.Interface, nodeName string) func(ctx cont
 		}
 		pools := make(map[string]workerPoolRef, len(pods.Items))
 		for _, pod := range pods.Items {
+			name := pod.Labels[workerPoolLabel]
+			if name == "" {
+				// The existence selector also matches empty-valued labels
+				// (a bare key in YAML parses to ""), and half a pair names
+				// no pool: skip it, so absent and unresolvable are the same
+				// unlabeled answer downstream.
+				continue
+			}
 			pools[string(pod.UID)] = workerPoolRef{
 				namespace: pod.Namespace,
-				name:      pod.Labels[workerPoolLabel],
+				name:      name,
 			}
 		}
 		return pools
@@ -516,36 +528,41 @@ func (i *statsInstruments) addCPU(ctx context.Context, aggs map[templateKey]*tem
 		return
 	}
 	for key, agg := range aggs {
-		// The wire carries microseconds; the metric is seconds, the base unit
-		// CPU time is exported in everywhere else (cAdvisor's
-		// container_cpu_usage_seconds_total, OTel's *.cpu.time), so the
+		// The wire carries microseconds; the metric is seconds -- the base
+		// unit CPU time is exported in everywhere else (cAdvisor's
+		// container_cpu_usage_seconds_total, OTel's *.cpu.time) -- so the
 		// existing rate() idioms read directly as cores.
 		i.cpuUsage.Add(ctx, float64(agg.cpuDeltaUsec)/1e6, key.attrs())
 	}
 }
 
-// startStatsPoller assembles the poller and starts it. Split from main's boot
-// sequence so the sampling subsystem has one obvious entry point.
+// startStatsPoller assembles the sampling subsystem -- the metrics poller and
+// the periodic events channel -- and starts the poller. Split from main's
+// boot sequence so the subsystem has one obvious entry point.
 //
-// The poller dials its own per-probe connections (see statsPoller.dial) and
-// takes no AteomDialer: the isolation from the lifecycle RPCs' connection
-// cache is structural, not just behavioral.
-func startStatsPoller(ctx context.Context, interval time.Duration, inst *statsInstruments, k8sClient kubernetes.Interface) {
+// The poller dials its own short-lived connection per probe (see
+// dialAteomStats) and takes no AteomDialer: the isolation from the lifecycle
+// RPCs' connection cache is structural, not just behavioral.
+// logSink is the process's synchronized stdout writer, shared with the
+// runtime logger so the event drain and slog can never tear each other's
+// records.
+func startStatsPoller(ctx context.Context, interval time.Duration, inst *statsInstruments, k8sClient kubernetes.Interface, logSink io.Writer) {
+	// Warm the labels-key resolution so the first emit does not pay the
+	// metadata probe either; see defaultLabelsKey.
+	go defaultLabelsKey()
+
 	poller := &statsPoller{
 		interval:  interval,
 		ateomsDir: ateompath.AteomsDir(),
 		dial: func(_ context.Context, podUID string) (activeStatsClient, io.Closer, error) {
-			conn, err := grpc.NewClient(
-				"unix://"+ateompath.AteomSocketPath(podUID),
-				grpc.WithTransportCredentials(insecure.NewCredentials()),
-				grpc.WithStatsHandler(otelgrpc.NewClientHandler()),
-			)
+			conn, closer, err := dialAteomStats(podUID)
 			if err != nil {
 				return nil, nil, err
 			}
-			return ateompb.NewAteomClient(conn), conn, nil
+			return ateompb.NewAteomClient(conn), closer, nil
 		},
-		inst: inst,
+		inst:         inst,
+		eventEmitter: newStatsEventEmitter(newAsyncWriter(ctx, logSink, usageEventQueueDepth), defaultLabelsKey),
 	}
 	// NODE_NAME comes from the Downward API; without it the samples still
 	// flow, just grouped without pool labels.
@@ -556,3 +573,19 @@ func startStatsPoller(ctx context.Context, interval time.Duration, inst *statsIn
 	}
 	go poller.run(ctx)
 }
+
+// dialAteomStats opens the poller's short-lived connection: one per probe,
+// closed by the caller, never the lifecycle RPCs' cached AteomDialer.
+// grpc.NewClient is lazy, so this cannot block; the caller's deadline bounds
+// the actual connect inside the RPC.
+func dialAteomStats(podUID string) (*grpc.ClientConn, io.Closer, error) {
+	conn, err := grpc.NewClient(
+		"unix://"+ateompath.AteomSocketPath(podUID),
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithStatsHandler(otelgrpc.NewClientHandler()),
+	)
+	if err != nil {
+		return nil, nil, err
+	}
+	return conn, conn, nil
+}

cmd/atelet/statspoller_test.go

@@ -15,7 +15,9 @@
 package main
 
 import (
+	"bytes"
 	"context"
+	"encoding/json"
 	"errors"
 	"io"
 	"math"
@@ -30,6 +32,10 @@ import (
 	"go.opentelemetry.io/otel/sdk/metric/metricdata"
 	"google.golang.org/grpc"
 
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	k8sfake "k8s.io/client-go/kubernetes/fake"
+
 	"github.com/agent-substrate/substrate/internal/proto/ateompb"
 )
 
@@ -359,6 +365,43 @@ func TestStatsPollerWorkerPoolLabels(t *testing.T) {
 	}
 }
 
+// TestStatsPollerPeriodicEvents pins the events channel: one event per
+// executing sample per sweep, none for idle or mid-boot ateoms, identity
+// taken from the echo, pool labels from the sweep's own resolution.
+func TestStatsPollerPeriodicEvents(t *testing.T) {
+	fakes := map[string]*fakeStatsAteom{
+		"uid-1": {resp: executingResponse("ns-a", "tmpl-a", ateompb.SandboxClass_SANDBOX_CLASS_GVISOR, ateompb.StatsSource_STATS_SOURCE_CGROUP, 1000, 700)},
+		"uid-2": {resp: noSampleResponse(ateompb.NoSampleReason_NO_SAMPLE_REASON_NO_WORKLOAD)},
+	}
+	p, _ := newPollerFixture(t, fakes)
+	var buf syncBuffer
+	p.eventEmitter = newBufferEmitter(&buf, false)
+	p.workerPools = func(context.Context) map[string]workerPoolRef {
+		return map[string]workerPoolRef{"uid-1": {namespace: "pool-ns", name: "pool-a"}}
+	}
+
+	p.collect(context.Background())
+
+	lines := bytes.Count(bytes.TrimSpace(buf.Bytes()), []byte("\n")) + 1
+	if buf.Len() == 0 {
+		t.Fatal("no periodic event emitted for the executing ateom")
+	}
+	if lines != 1 {
+		t.Fatalf("emitted %d events, want 1 (idle ateoms emit nothing): %q", lines, buf.String())
+	}
+	var rec map[string]any
+	if err := json.Unmarshal(bytes.TrimSpace(buf.Bytes()), &rec); err != nil {
+		t.Fatalf("unmarshal: %v", err)
+	}
+	if got := rec["kind"]; got != "periodic" {
+		t.Errorf("kind = %v, want periodic", got)
+	}
+	labels, _ := rec["labels"].(map[string]any)
+	if got := labels["ate.workerpool.name"]; got != "pool-a" {
+		t.Errorf("labels[ate.workerpool.name] = %v, want pool-a", got)
+	}
+}
+
 func TestAddSat(t *testing.T) {
 	tests := []struct {
 		name string
@@ -426,3 +469,31 @@ func TestStatsPollerCPUDeltaSaturatesCorruptCounter(t *testing.T) {
 		t.Errorf("corrupt-counter sweep delta = %d, want pinned at MaxInt64", got)
 	}
 }
+
+// TestNodeWorkerPools pins the resolver's ingestion rules: a labeled worker
+// maps by pod UID, an empty label value names no pool and never enters the
+// map (the presence-only selector matches it anyway), and unlabeled pods are
+// not workers at all. The fake clientset honors label selectors but not the
+// spec.nodeName field selector, so node scoping is not assertable here.
+func TestNodeWorkerPools(t *testing.T) {
+	client := k8sfake.NewSimpleClientset(
+		&corev1.Pod{ObjectMeta: metav1.ObjectMeta{
+			Name: "worker-a", Namespace: "pool-ns", UID: "uid-a",
+			Labels: map[string]string{workerPoolLabel: "pool-a"},
+		}},
+		&corev1.Pod{ObjectMeta: metav1.ObjectMeta{
+			Name: "worker-empty", Namespace: "pool-ns", UID: "uid-empty",
+			Labels: map[string]string{workerPoolLabel: ""},
+		}},
+		&corev1.Pod{ObjectMeta: metav1.ObjectMeta{
+			Name: "bystander", Namespace: "other-ns", UID: "uid-bystander",
+		}},
+	)
+
+	got := nodeWorkerPools(client, "node-1")(context.Background())
+
+	want := map[string]workerPoolRef{"uid-a": {namespace: "pool-ns", name: "pool-a"}}
+	if diff := cmp.Diff(want, got, cmp.AllowUnexported(workerPoolRef{})); diff != "" {
+		t.Errorf("nodeWorkerPools mismatch (-want +got):\n%s", diff)
+	}
+}

internal/actorlog/logger.go

@@ -73,15 +73,23 @@ const (
 	labelsKeyGCE   = "logging.googleapis.com/labels"
 )
 
+// LabelsKey returns the label group's spelling for this environment. Every
+// emitter of the actor-identity label group (container logs, lifecycle
+// events, usage events) must pick its key here, so they all promote into
+// Cloud Logging the same way -- and so going vendor-neutral later means
+// changing one function.
+func LabelsKey(isOnGCE bool) string {
+	if isOnGCE {
+		return labelsKeyGCE
+	}
+	return labelsKeyPlain
+}
+
 // NewActorLogger creates a new ActorLogger wrapping the provided destination writer.
 func NewActorLogger(w io.Writer, isOnGCE bool) *ActorLogger {
-	labelsKey := labelsKeyPlain
-	if isOnGCE {
-		labelsKey = labelsKeyGCE
-	}
 	return &ActorLogger{
 		writer:    w,
-		labelsKey: labelsKey,
+		labelsKey: LabelsKey(isOnGCE),
 	}
 }
 

internal/actorlog/logger_test.go

@@ -450,3 +450,14 @@ func mustSpanID(t *testing.T, s string) trace.SpanID {
 	}
 	return id
 }
+
+// TestLabelsKey pins the one place the label group's spelling is chosen: the
+// GCE spelling is the key Cloud Logging promotes into LogEntry.labels.
+func TestLabelsKey(t *testing.T) {
+	if got := LabelsKey(false); got != "labels" {
+		t.Errorf("LabelsKey(false) = %q, want labels", got)
+	}
+	if got := LabelsKey(true); got != "logging.googleapis.com/labels" {
+		t.Errorf("LabelsKey(true) = %q, want logging.googleapis.com/labels", got)
+	}
+}