Add opt-in guest metrics for selected microVM actors

cmd/atelet/actorguestmetrics.go

@@ -0,0 +1,144 @@
+// Copyright 2026 The Agent Substrate Authors
+//
+// 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 (
+	"errors"
+	"fmt"
+	"sync/atomic"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"go.ate.dev/substrate/internal/proto/ateworkerpb"
+	"go.ate.dev/substrate/internal/resources"
+	"k8s.io/apimachinery/pkg/util/validation/field"
+)
+
+const maxDebugActorUIDs = 16
+
+var (
+	guestMetricLabels     = []string{"actor_uid", "actor_name", "actor_atespace"}
+	guestCPUCores         = prometheus.NewDesc("ate_debug_actor_guest_cpu_usage_cores", "Mean guest CPU cores used between two samples in the same activation.", guestMetricLabels, nil)
+	guestMemoryUsage      = prometheus.NewDesc("ate_debug_actor_guest_memory_usage_bytes", "Guest memory usage including reclaimable page cache.", guestMetricLabels, nil)
+	guestMemoryWorkingSet = prometheus.NewDesc("ate_debug_actor_guest_memory_working_set_bytes", "Guest memory working set.", guestMetricLabels, nil)
+)
+
+type actorGuestReading struct {
+	sample *ateworkerpb.WorkloadStatsSample
+	cores  float64
+	hasCPU bool
+}
+
+// actorGuestMetrics exposes samples for the operator's allowlist.
+// The stats poller publishes a complete snapshot after each sweep.
+type actorGuestMetrics struct {
+	allowed  map[string]struct{}
+	interval time.Duration
+	latest   atomic.Pointer[map[string]actorGuestReading]
+}
+
+func newActorGuestMetrics(uids []string, interval time.Duration) (*actorGuestMetrics, error) {
+	if len(uids) == 0 {
+		return nil, nil
+	}
+	if interval <= 0 {
+		return nil, errors.New("actor stats polling must be enabled")
+	}
+	if len(uids) > maxDebugActorUIDs {
+		return nil, fmt.Errorf("at most %d actor UIDs are allowed", maxDebugActorUIDs)
+	}
+	m := &actorGuestMetrics{allowed: make(map[string]struct{}, len(uids)), interval: interval}
+	for _, uid := range uids {
+		if errs := resources.ValidateUUID(uid, field.NewPath("uid")); len(errs) > 0 {
+			return nil, errs.ToAggregate()
+		}
+		m.allowed[uid] = struct{}{}
+	}
+	return m, nil
+}
+
+func (m *actorGuestMetrics) selectSample(samples map[string]*ateworkerpb.WorkloadStatsSample, sample *ateworkerpb.WorkloadStatsSample) {
+	uid := sample.GetActorUid()
+	if _, ok := m.allowed[uid]; !ok || sample.GetEpochUnixNano() <= 0 {
+		return
+	}
+	switch sample.GetSource() {
+	case ateworkerpb.StatsSource_STATS_SOURCE_UNSPECIFIED, ateworkerpb.StatsSource_STATS_SOURCE_CGROUP:
+		// A restore can supersede the guest reading before the new activation
+		// is measured, or move the actor to gVisor.
+	case ateworkerpb.StatsSource_STATS_SOURCE_GUEST_AGENT:
+		if sample.GetObservedAtUnixNano() <= 0 {
+			return
+		}
+	default:
+		return
+	}
+	old := samples[uid]
+	if sample.GetEpochUnixNano() > old.GetEpochUnixNano() ||
+		(sample.GetEpochUnixNano() == old.GetEpochUnixNano() && sample.GetObservedAtUnixNano() > old.GetObservedAtUnixNano()) {
+		samples[uid] = sample
+	}
+}
+
+func (m *actorGuestMetrics) publish(samples map[string]*ateworkerpb.WorkloadStatsSample) {
+	previous := m.latest.Load()
+	next := make(map[string]actorGuestReading, len(samples))
+	for uid, sample := range samples {
+		if sample.GetSource() != ateworkerpb.StatsSource_STATS_SOURCE_GUEST_AGENT {
+			continue
+		}
+		reading := actorGuestReading{sample: sample}
+		if previous != nil {
+			last := (*previous)[uid]
+			if last.sample != nil && last.sample.GetEpochUnixNano() == sample.GetEpochUnixNano() {
+				elapsed := sample.GetObservedAtUnixNano() - last.sample.GetObservedAtUnixNano()
+				if elapsed == 0 {
+					reading = last
+				} else if elapsed > 0 && sample.GetCpuUsageUsec() >= last.sample.GetCpuUsageUsec() {
+					reading.cores = 1000 * float64(sample.GetCpuUsageUsec()-last.sample.GetCpuUsageUsec()) / float64(elapsed)
+					reading.hasCPU = true
+				}
+			}
+		}
+		next[uid] = reading
+	}
+	m.latest.Store(&next)
+}
+
+func (m *actorGuestMetrics) Describe(ch chan<- *prometheus.Desc) {
+	ch <- guestCPUCores
+	ch <- guestMemoryUsage
+	ch <- guestMemoryWorkingSet
+}
+
+func (m *actorGuestMetrics) Collect(ch chan<- prometheus.Metric) {
+	snapshot := m.latest.Load()
+	if snapshot == nil {
+		return
+	}
+	cutoff := time.Now().Add(-m.interval).Add(-m.interval).UnixNano()
+	for uid, reading := range *snapshot {
+		sample := reading.sample
+		if sample.GetObservedAtUnixNano() < cutoff {
+			continue
+		}
+		labels := []string{uid, sample.GetActorName(), sample.GetAtespace()}
+		ch <- prometheus.MustNewConstMetric(guestMemoryUsage, prometheus.GaugeValue, float64(sample.GetMemoryCurrentBytes()), labels...)
+		ch <- prometheus.MustNewConstMetric(guestMemoryWorkingSet, prometheus.GaugeValue, float64(sample.GetMemoryWorkingSetBytes()), labels...)
+		if reading.hasCPU {
+			ch <- prometheus.MustNewConstMetric(guestCPUCores, prometheus.GaugeValue, reading.cores, labels...)
+		}
+	}
+}

cmd/atelet/main.go

@@ -31,6 +31,7 @@ import (
 	"time"
 
 	"github.com/google/go-containerregistry/pkg/authn"
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/spf13/pflag"
 	"github.com/spiffe/go-spiffe/v2/spiffeid"
 	"github.com/spiffe/go-spiffe/v2/svid/x509svid"
@@ -128,6 +129,7 @@ var (
 	otlpRelaySocket = pflag.String("otlp-relay-socket", nodepath.AteletOTLPSocketPath(), "Unix socket to serve the OTLP relay on, which forwards the node's worker telemetry to OTEL_EXPORTER_OTLP_ENDPOINT so worker pods need no network path to the collector. Empty disables the relay.")
 
 	actorStatsPollInterval = pflag.Duration("actor-stats-poll-interval", time.Minute, fmt.Sprintf("Actor resource utilization sampling frequency. 0 disables the sampling entirely; minimum accepted value is %v.", minActorStatsPollInterval))
+	actorGuestMetricsUIDs  = pflag.StringSlice("debug-actor-guest-metrics-uids", nil, "Actor UIDs whose guest metrics to expose on /metrics. Empty disables per-actor metrics.")
 
 	drainDelay   = pflag.Duration("drain-delay", 0, "How long to keep accepting new RPCs after SIGTERM before starting the gRPC drain.")
 	drainTimeout = pflag.Duration("drain-timeout", 5*time.Minute, "Deadline for the graceful gRPC drain on shutdown. In-flight RPCs still running past it are forcefully cancelled.")
@@ -212,6 +214,16 @@ func main() {
 	readiness := &serverboot.Readiness{}
 	serving := make(chan struct{})
 	readiness.AddCheck("serving", servingCheck(serving))
+	statsInterval := clampActorStatsPollInterval(ctx, *actorStatsPollInterval)
+	guestMetrics, err := newActorGuestMetrics(*actorGuestMetricsUIDs, statsInterval)
+	if err != nil {
+		serverboot.Fatal(ctx, "Invalid actor guest debug metrics configuration", err)
+	}
+	if guestMetrics != nil {
+		if err := prometheus.Register(guestMetrics); err != nil {
+			serverboot.Fatal(ctx, "Failed to register actor guest debug metrics", err)
+		}
+	}
 	go serverboot.StartMetricsServer(ctx, *metricsListenAddr)
 	go serverboot.StartHealthServer(ctx, *healthListenAddr, readiness)
 
@@ -329,7 +341,7 @@ func main() {
 		serverboot.Fatal(ctx, "Failed to create Kubernetes clients", err)
 	}
 
-	if interval := clampActorStatsPollInterval(ctx, *actorStatsPollInterval); interval > 0 {
+	if statsInterval > 0 {
 		if statsInst, err := newStatsInstruments(otel.Meter("atelet")); err != nil {
 			// Telemetry must not take the node's lifecycle daemon down with
 			// it. Instrument creation only fails on programmer error
@@ -339,7 +351,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, statsInterval, statsInst, k8sClient, guestMetrics)
 		}
 	}
 

cmd/atelet/statspoller.go

@@ -123,7 +123,8 @@ type statsPoller struct {
 	// which is the join key.
 	fetchWorkerPools func(ctx context.Context) map[string]workerPoolRef
 
-	inst *statsInstruments
+	inst         *statsInstruments
+	guestMetrics *actorGuestMetrics
 
 	// cachedPools carries pool resolutions across sweeps, so one failed pod
 	// list cannot re-home a tick's samples -- and the CPU counter's
@@ -259,6 +260,9 @@ func (p *statsPoller) collect(ctx context.Context) map[templateKey]*templateAggr
 		// A node with no workers directory yet has no workers to measure; the
 		// first RunWorkload dispatch creates it.
 		slog.DebugContext(ctx, "Actor stats sweep: no workers directory", slog.Any("err", err))
+		if p.guestMetrics != nil {
+			p.guestMetrics.publish(nil)
+		}
 		return nil
 	}
 
@@ -271,6 +275,10 @@ func (p *statsPoller) collect(ctx context.Context) map[templateKey]*templateAggr
 		}
 	}
 	pools := p.resolveWorkerPools(ctx, podUIDs)
+	var guestSamples map[string]*ateworkerpb.WorkloadStatsSample
+	if p.guestMetrics != nil {
+		guestSamples = make(map[string]*ateworkerpb.WorkloadStatsSample)
+	}
 
 	var (
 		mu      sync.Mutex
@@ -311,6 +319,11 @@ func (p *statsPoller) collect(ctx context.Context) map[templateKey]*templateAggr
 			// One entry per workload the worker is hosting; empty when it is
 			// available.
 			for _, sample := range resp.GetSamples() {
+				if p.guestMetrics != nil {
+					mu.Lock()
+					p.guestMetrics.selectSample(guestSamples, sample)
+					mu.Unlock()
+				}
 				cpuAt := cpuKey{actorUID: sample.GetActorUid(), epoch: sample.GetEpochUnixNano()}
 				if sample.GetSource() == ateworkerpb.StatsSource_STATS_SOURCE_UNSPECIFIED {
 					// Pending: hosted but not measured, so it adds nothing
@@ -363,6 +376,9 @@ func (p *statsPoller) collect(ctx context.Context) map[templateKey]*templateAggr
 	// The tasks only ever return nil: a probe that fails is "not a target this
 	// tick", never a failed sweep.
 	_ = g.Wait()
+	if p.guestMetrics != nil {
+		p.guestMetrics.publish(guestSamples)
+	}
 	// A worker that did not answer keeps its activations' baselines: dropping
 	// one would charge the activation in full again once the worker answers.
 	// Everything else this sweep did not see is dropped, so lastCPU does not
@@ -582,7 +598,7 @@ func (i *statsInstruments) addCPU(ctx context.Context, aggs map[templateKey]*tem
 // The poller dials its own short-lived connection per probe (see
 // dialWorkerStats) and takes no WorkerDialer: 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) {
+func startStatsPoller(ctx context.Context, interval time.Duration, inst *statsInstruments, k8sClient kubernetes.Interface, guestMetrics *actorGuestMetrics) {
 	poller := &statsPoller{
 		interval:   interval,
 		workersDir: nodepath.WorkersDir(),
@@ -593,8 +609,9 @@ func startStatsPoller(ctx context.Context, interval time.Duration, inst *statsIn
 			}
 			return ateworkerpb.NewWorkerClient(conn), closer, nil
 		},
-		inst:      inst,
-		startedAt: time.Now().UnixNano(),
+		inst:         inst,
+		guestMetrics: guestMetrics,
+		startedAt:    time.Now().UnixNano(),
 	}
 	// NODE_NAME comes from the Downward API; without it the samples still
 	// flow, just grouped without pool labels.

cmd/atelet/statspoller_test.go

@@ -19,13 +19,18 @@ import (
 	"errors"
 	"io"
 	"math"
+	"net/http"
+	"net/http/httptest"
 	"os"
 	"path/filepath"
+	"strings"
 	"sync"
 	"testing"
 	"time"
 
 	"github.com/google/go-cmp/cmp"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
 	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
 	"go.opentelemetry.io/otel/sdk/metric/metricdata"
 	"google.golang.org/grpc"
@@ -188,6 +193,102 @@ func assertClosed(t *testing.T, fakes map[string]*fakeStatsWorker, closers map[s
 	}
 }
 
+func TestActorGuestDebugMetrics(t *testing.T) {
+	const uid = "c7f3fb50-ef2d-4075-b59a-4a8e192e48a0"
+	const other = "c7f3fb50-ef2d-4075-b59a-4a8e192e48a1"
+	const unselected = "c7f3fb50-ef2d-4075-b59a-4a8e192e48a2"
+	const labels = `{actor_atespace="team",actor_name="my-agent",actor_uid="` + uid + `"}`
+	const cpu = "ate_debug_actor_guest_cpu_usage_cores" + labels + " 1"
+	const memory = "ate_debug_actor_guest_memory_usage_bytes" + labels + " 4096"
+	const workingSet = "ate_debug_actor_guest_memory_working_set_bytes" + labels + " 2048"
+	now := time.Now()
+	start := now.Add(-10 * time.Second)
+	guest := func(actorUID string, epoch, observed time.Time, cpuUsec uint64) *ateworkerpb.WorkloadStatsSample {
+		sample := atEpoch(measuredSample(actorUID, "ns", "template", 4096, 2048, cpuUsec), epoch.UnixNano())
+		sample.Source = ateworkerpb.StatsSource_STATS_SOURCE_GUEST_AGENT
+		sample.SandboxClass = "microvm"
+		sample.Atespace = "team"
+		sample.ActorName = "my-agent"
+		sample.ObservedAtUnixNano = observed.UnixNano()
+		return sample
+	}
+
+	metrics, err := newActorGuestMetrics([]string{uid, other}, time.Minute)
+	if err != nil {
+		t.Fatal(err)
+	}
+	worker := &fakeStatsWorker{}
+	poller, _ := newPollerFixture(t, map[string]*fakeStatsWorker{"worker": worker})
+	poller.guestMetrics = metrics
+	if err := prometheus.Register(metrics); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { prometheus.Unregister(metrics) })
+	handler := promhttp.Handler()
+	scrape := func(samples ...*ateworkerpb.WorkloadStatsSample) string {
+		t.Helper()
+		worker.resp = &ateworkerpb.GetActiveWorkloadStatsResponse{Samples: samples}
+		poller.collect(t.Context())
+		res := httptest.NewRecorder()
+		req := httptest.NewRequest(http.MethodGet, "/metrics", nil)
+		handler.ServeHTTP(res, req)
+		if res.Code != http.StatusOK {
+			t.Fatalf("scrape returned %d: %s", res.Code, res.Body)
+		}
+		return res.Body.String()
+	}
+
+	first := guest(uid, start, start, 100_000)
+	if got := scrape(first, guest(unselected, start, start, 1),
+		measuredSample(other, "ns", "template", 4096, 2048, 1)); !strings.Contains(got, memory) ||
+		!strings.Contains(got, workingSet) || strings.Contains(got, "cpu_usage_cores") ||
+		strings.Contains(got, other) || strings.Contains(got, unselected) {
+		t.Fatalf("first scrape should contain only the selected guest memory: %s", got)
+	}
+	second := guest(uid, start, start.Add(time.Second), 1_100_000)
+	for _, sample := range []*ateworkerpb.WorkloadStatsSample{second, second} {
+		if got := scrape(sample, first); !strings.Contains(got, cpu+"\n") {
+			t.Fatalf("new or cached reading should report one CPU core: %s", got)
+		}
+	}
+	if got := scrape(second, guest(uid, now, now, 2_100_000)); !strings.Contains(got, memory) ||
+		strings.Contains(got, "cpu_usage_cores") {
+		t.Fatalf("a new activation should start with a fresh CPU baseline: %s", got)
+	}
+	if got := scrape(guest(other, now, now, 2_200_000)); !strings.Contains(got, other) ||
+		strings.Contains(got, uid) || strings.Contains(got, "cpu_usage_cores") {
+		t.Fatalf("a recreated actor should have its own series and CPU baseline: %s", got)
+	}
+	scrape(second)
+	pending := atEpoch(pendingSample(uid, "ns", "template"), now.UnixNano())
+	if got := scrape(second, pending); strings.Contains(got, uid) {
+		t.Fatalf("pending restore exposed the previous activation's series: %s", got)
+	}
+	expired := now.Add(-3 * time.Minute)
+	if got := scrape(guest(uid, expired, expired, 1)); strings.Contains(got, uid) {
+		t.Fatalf("expired guest sample still had a series: %s", got)
+	}
+}
+
+func TestActorGuestDebugMetricsConfig(t *testing.T) {
+	const uid = "c7f3fb50-ef2d-4075-b59a-4a8e192e48a0"
+	if metrics, err := newActorGuestMetrics(nil, 0); metrics != nil || err != nil {
+		t.Fatalf("empty allowlist should disable per-actor metrics, got %v, %v", metrics, err)
+	}
+	for _, tc := range []struct {
+		uids     []string
+		interval time.Duration
+	}{
+		{[]string{uid}, 0},
+		{[]string{"*"}, time.Minute},
+		{make([]string, maxDebugActorUIDs+1), time.Minute},
+	} {
+		if _, err := newActorGuestMetrics(tc.uids, tc.interval); err == nil {
+			t.Errorf("accepted invalid config with UIDs %q and interval %v", tc.uids, tc.interval)
+		}
+	}
+}
+
 func TestStatsPollerCollectAggregates(t *testing.T) {
 	// Two actors of the same template on this node, one of another, one idle
 	// worker, one mid-boot: the same-template pair sums, the others contribute