diff --git a/cmd/atelet/main.go b/cmd/atelet/main.go index 0314faa82e..5ad4bced66 100644 --- a/cmd/atelet/main.go +++ b/cmd/atelet/main.go @@ -36,6 +36,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. diff --git a/cmd/atelet/statsevents.go b/cmd/atelet/statsevents.go new file mode 100644 index 0000000000..fd2672ec1e --- /dev/null +++ b/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()), + ) +} diff --git a/cmd/atelet/statsevents_test.go b/cmd/atelet/statsevents_test.go new file mode 100644 index 0000000000..cffcfd362c --- /dev/null +++ b/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") +} diff --git a/cmd/atelet/statspoller.go b/cmd/atelet/statspoller.go index 5224aff4ff..ea063b82c8 100644 --- a/cmd/atelet/statspoller.go +++ b/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 +} diff --git a/cmd/atelet/statspoller_test.go b/cmd/atelet/statspoller_test.go index c3ae0a4114..220e189114 100644 --- a/cmd/atelet/statspoller_test.go +++ b/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) + } +} diff --git a/internal/actorlog/logger.go b/internal/actorlog/logger.go index a80bea8c40..7ebc7f1a79 100644 --- a/internal/actorlog/logger.go +++ b/internal/actorlog/logger.go @@ -73,15 +73,23 @@ const ( labelsKeyGCE = "logging.googleapis.com/labels" ) -// NewActorLogger creates a new ActorLogger wrapping the provided destination writer. -func NewActorLogger(w io.Writer, isOnGCE bool) *ActorLogger { - labelsKey := labelsKeyPlain +// 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 { - labelsKey = labelsKeyGCE + return labelsKeyGCE } + return labelsKeyPlain +} + +// NewActorLogger creates a new ActorLogger wrapping the provided destination writer. +func NewActorLogger(w io.Writer, isOnGCE bool) *ActorLogger { return &ActorLogger{ writer: w, - labelsKey: labelsKey, + labelsKey: LabelsKey(isOnGCE), } } diff --git a/internal/actorlog/logger_test.go b/internal/actorlog/logger_test.go index 4bac344605..7b6bf9d9a8 100644 --- a/internal/actorlog/logger_test.go +++ b/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) + } +}