diff --git a/cmd/agent/main.go b/cmd/agent/main.go index 3aa24ff..eb7c085 100644 --- a/cmd/agent/main.go +++ b/cmd/agent/main.go @@ -34,6 +34,7 @@ import ( "github.com/RebuildStackCo/runtime-agent/internal/inventory" "github.com/RebuildStackCo/runtime-agent/internal/journal" "github.com/RebuildStackCo/runtime-agent/internal/metadata" + "github.com/RebuildStackCo/runtime-agent/internal/metrics" "github.com/RebuildStackCo/runtime-agent/internal/model" "github.com/RebuildStackCo/runtime-agent/internal/nodeauth" "github.com/RebuildStackCo/runtime-agent/internal/nodeintake" @@ -828,7 +829,40 @@ func run(ctx context.Context, logger *slog.Logger, clientset kubernetes.Interfac } return true, "" } - tasks["health"] = health.New(addr, live, ready, logger).Run + // The metrics endpoint joins this listener as a third path, as ADR 0069 + // §1 reserved. Its numbers come from the accessors above rather than + // from counters of its own (ADR 0070 §2). + sources := controllerSources{ + beat: beat, + sources: podWatcher.SourceHealths, + kubelet: usagePoller.PathReads, + filter: filter.Snapshot, + placement: podWatcher.PlacementDrops, + nodes: nodeWatcher.Drops, + spool: func() sink.Counters { return spool.Counters() }, + } + sources.intake = func() (model.IntakeRejections, bool) { + if intakeRejections == nil { + return model.IntakeRejections{}, false + } + return intakeRejections(), true + } + if goStore != nil { + sources.inventory = func() (inventory.Counters, inventory.ScanCoverage) { + return goStore.Counters(), goStore.ScanCoverage() + } + sources.ebpf = func() (inventory.ProfileCoverage, uint64, uint64) { + return goStore.ProfileCoverage(), profilesReceived.Load(), profilesUnjoined.Load() + } + } + if prober != nil { + sources.probe = prober.Snapshot + } + if puller != nil { + sources.pull = puller.Snapshot + } + tasks["health"] = health.New(addr, live, ready, + metrics.Handler(sources.gather, logger), logger).Run } // The node-intake receiver is optional (only the ebpf/node profile ships a diff --git a/cmd/agent/metrics.go b/cmd/agent/metrics.go new file mode 100644 index 0000000..fc7ef5a --- /dev/null +++ b/cmd/agent/metrics.go @@ -0,0 +1,106 @@ +package main + +import ( + dto "github.com/prometheus/client_model/go" + + "github.com/RebuildStackCo/runtime-agent/internal/collector" + "github.com/RebuildStackCo/runtime-agent/internal/health" + "github.com/RebuildStackCo/runtime-agent/internal/inventory" + "github.com/RebuildStackCo/runtime-agent/internal/metrics" + "github.com/RebuildStackCo/runtime-agent/internal/model" + "github.com/RebuildStackCo/runtime-agent/internal/nodescan" + "github.com/RebuildStackCo/runtime-agent/internal/pprofprobe" + "github.com/RebuildStackCo/runtime-agent/internal/pprofpull" + "github.com/RebuildStackCo/runtime-agent/internal/sink" +) + +// controllerSources is where the controller's metrics come from: the same +// accessors `logCoverage` reads to build `collection_coverage`, held as +// functions so the two are one set of counters read twice rather than two sets +// (ADR 0070 §2). A nil function is a feature this installation does not have. +type controllerSources struct { + beat *health.Heartbeat + sources func() []model.SourceHealth + kubelet func() []collector.PathReads + filter func() model.Coverage + placement func() model.PlacementDrops + nodes func() model.NodeDrops + spool func() sink.Counters + + // intake is late-bound: the receiver is wired after this listener is, so + // the presence of one is a question asked per scrape, not at startup. + intake func() (model.IntakeRejections, bool) + inventory func() (inventory.Counters, inventory.ScanCoverage) + ebpf func() (inventory.ProfileCoverage, uint64, uint64) + probe func() pprofprobe.Coverage + pull func() pprofpull.Coverage +} + +func (c controllerSources) gather() ([]*dto.MetricFamily, error) { + m := metrics.Controller{ + Process: metrics.Process{ + Role: metrics.RoleController, + Version: version, + PassStart: c.beat.Last(), + Deadline: c.beat.Deadline(), + }, + Sources: c.sources(), + Kubelet: c.kubelet(), + Filter: c.filter(), + Placement: c.placement(), + Nodes: c.nodes(), + Spool: c.spool(), + } + if c.intake != nil { + if r, ok := c.intake(); ok { + m.Intake = &r + } + } + if c.inventory != nil { + counters, scan := c.inventory() + m.Inventory, m.Scan = &counters, &scan + } + if c.ebpf != nil { + // Present only once a node has said what its profiler did, exactly as + // in the coverage payload: before that there is no fleet to describe. + if cov, received, unjoined := c.ebpf(); cov.Nodes > 0 { + m.EBPF, m.ProfilesReceived, m.ProfilesUnjoined = &cov, received, unjoined + } + } + if c.probe != nil { + p := c.probe() + m.Probe = &p + } + if c.pull != nil { + p := c.pull() + m.Pull = &p + } + return metrics.CollectController(m).Families() +} + +// nodeSources is the node role's equivalent. +type nodeSources struct { + beat *health.Heartbeat + scan func() (nodescan.Counters, int, bool) + profiling func() nodescan.ProfilingCoverage + shipped func() (uint64, uint64) +} + +func (n nodeSources) gather() ([]*dto.MetricFamily, error) { + m := metrics.Node{ + Process: metrics.Process{ + Role: metrics.RoleNode, + Version: version, + PassStart: n.beat.Last(), + Deadline: n.beat.Deadline(), + }, + Profiling: n.profiling(), + } + // Before the first pass finishes there is nothing to say about a scan, and + // zeroes would read as a node that walked an empty process table. + if counters, inScope, ok := n.scan(); ok { + m.Scanned, m.Scan, m.PodsInScope = true, counters, inScope + } + m.ReportsShipped, m.ReportFailures = n.shipped() + return metrics.CollectNode(m).Families() +} diff --git a/cmd/agent/metrics_test.go b/cmd/agent/metrics_test.go new file mode 100644 index 0000000..536f241 --- /dev/null +++ b/cmd/agent/metrics_test.go @@ -0,0 +1,286 @@ +package main + +import ( + "context" + "io" + "log/slog" + "net/http" + "strings" + "testing" + "time" + + dto "github.com/prometheus/client_model/go" + "github.com/prometheus/common/expfmt" + "github.com/prometheus/common/model" + "k8s.io/client-go/kubernetes/fake" + "k8s.io/client-go/rest" + + "github.com/RebuildStackCo/runtime-agent/internal/config" + "github.com/RebuildStackCo/runtime-agent/internal/health" + "github.com/RebuildStackCo/runtime-agent/internal/metrics" + "github.com/RebuildStackCo/runtime-agent/internal/sink" +) + +// scrape fetches and parses the exposition, with the same package's decoder the +// cAdvisor reader uses — so what this asserts is that a Prometheus can read it, +// not that some bytes were produced. +func scrape(t *testing.T, addr string) map[string]*dto.MetricFamily { + t.Helper() + req, err := http.NewRequestWithContext(context.Background(), http.MethodGet, + "http://"+addr+health.MetricsPath, nil) + if err != nil { + t.Fatal(err) + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("scraping: %v", err) + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode != http.StatusOK { + t.Fatalf("GET %s = %d", health.MetricsPath, resp.StatusCode) + } + parser := expfmt.NewTextParser(model.UTF8Validation) + families, err := parser.TextToMetricFamilies(resp.Body) + if err != nil { + t.Fatalf("parsing the exposition: %v", err) + } + return families +} + +// TestTheControllerExposesItsOwnWork is the whole slice end to end: a running +// controller, scraped over the health listener's third path, parsed back. +func TestTheControllerExposesItsOwnWork(t *testing.T) { + addr := freeAddress(t) + cfg := config.Config{Health: config.Health{ListenAddress: addr}} + cfg.Spool.Dir = t.TempDir() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { + done <- run(ctx, slog.New(slog.NewTextHandler(io.Discard, nil)), + fake.NewClientset(), &rest.Config{}, cfg, config.Shape{}, time.Now()) + }() + waitFor(t, addr, health.ReadyPath, http.StatusOK) + + families := scrape(t, addr) + + // Every payload kind has a series before anything has shipped, because the + // spool's counters are seeded from the registry (ADR 0022, ADR 0070 §3). + kinds := map[string]bool{} + for _, m := range families[metrics.Prefix+"spool_payloads_written_total"].GetMetric() { + for _, l := range m.GetLabel() { + if l.GetName() == "kind" { + kinds[l.GetValue()] = true + } + } + } + for _, row := range sink.Registry() { + if !kinds[row.Kind] { + t.Errorf("no series for payload kind %q", row.Kind) + } + } + + for _, required := range []string{ + "build_info", "pass_start_timestamp_seconds", "pass_deadline_seconds", + "source_synced", "kubelet_requests_total", "pods_observed_total", + "spool_bytes", "spool_files", "spool_evicted_total", + } { + if families[metrics.Prefix+required] == nil { + t.Errorf("%s%s is not exposed", metrics.Prefix, required) + } + } + + // And nothing it serves carries a label name outside the permitted set — + // asserted against what actually crossed the wire, not against the builder. + for name, fam := range families { + if !strings.HasPrefix(name, metrics.Prefix) { + t.Errorf("metric %q does not carry the agent's prefix", name) + } + for _, m := range fam.GetMetric() { + for _, l := range m.GetLabel() { + if !metrics.Allowed[metrics.LabelName(l.GetName())] { + t.Errorf("metric %q carries label %q", name, l.GetName()) + } + } + } + } + + cancel() + select { + case err := <-done: + if err != nil { + t.Fatalf("run returned %v", err) + } + case <-time.After(10 * time.Second): + t.Fatal("run did not stop") + } +} + +// TestALogOnlyControllerStillExposesItself: no spool means nothing to collect +// from, and the metrics still say so — a scrape is how an operator finds out +// the agent is in that mode, so it must not be the mode with no scrape. +func TestALogOnlyControllerStillExposesItself(t *testing.T) { + addr := freeAddress(t) + cfg := config.Config{Health: config.Health{ListenAddress: addr}} + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { + done <- run(ctx, slog.New(slog.NewTextHandler(io.Discard, nil)), + fake.NewClientset(), &rest.Config{}, cfg, config.Shape{}, time.Now()) + }() + waitFor(t, addr, health.LivePath, http.StatusOK) + + families := scrape(t, addr) + if families[metrics.Prefix+"spool_files"] == nil { + t.Error("a log-only controller exposes no spool metrics at all") + } + if fam := families[metrics.Prefix+"spool_payloads_written_total"]; fam == nil { + t.Error("a log-only controller names no payload kinds") + } else if len(fam.GetMetric()) != len(sink.Registry()) { + t.Errorf("%d kind series, want %d", len(fam.GetMetric()), len(sink.Registry())) + } + cancel() + <-done +} + +// TestTheScrapePathChangesNothing: it is a read, like the two probes beside it +// (CLAUDE.md invariant 1). +func TestTheScrapePathChangesNothing(t *testing.T) { + addr := freeAddress(t) + cfg := config.Config{Health: config.Health{ListenAddress: addr}} + cfg.Spool.Dir = t.TempDir() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { + done <- run(ctx, slog.New(slog.NewTextHandler(io.Discard, nil)), + fake.NewClientset(), &rest.Config{}, cfg, config.Shape{}, time.Now()) + }() + waitFor(t, addr, health.ReadyPath, http.StatusOK) + + for _, method := range []string{http.MethodPost, http.MethodPut, http.MethodDelete} { + req, err := http.NewRequestWithContext(context.Background(), method, + "http://"+addr+health.MetricsPath, strings.NewReader("scrape_me=1")) + if err != nil { + t.Fatal(err) + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + _, _ = io.Copy(io.Discard, resp.Body) + _ = resp.Body.Close() + if resp.StatusCode != http.StatusMethodNotAllowed { + t.Errorf("%s %s = %d, want 405", method, health.MetricsPath, resp.StatusCode) + } + } + cancel() + <-done +} + +// TestTheNodeExposesItsOwnPass. The coverage payload is aggregate over the +// fleet by design (ADR 0054 §4), so which node went quiet is a question only a +// per-node scrape answers — and the node names itself nowhere in the reply: the +// asking is done by the customer's own service discovery (ADR 0070 §4). +func TestTheNodeExposesItsOwnPass(t *testing.T) { + t.Setenv("NODE_NAME", "node-1") + addr := freeAddress(t) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { + done <- runNode(ctx, slog.New(slog.NewTextHandler(io.Discard, nil)), []string{ + "-proc", t.TempDir(), + "-interval", "1s", + "-health-address", addr, + }) + }() + waitFor(t, addr, health.ReadyPath, http.StatusOK) + + families := scrape(t, addr) + for _, required := range []string{ + "build_info", "pass_start_timestamp_seconds", "scan_processes", + "scan_pods_in_scope", "ebpf_state", "node_reports_shipped_total", + } { + if families[metrics.Prefix+required] == nil { + t.Errorf("%s%s is not exposed", metrics.Prefix, required) + } + } + // The node role is not the controller, and says so in the one label that + // distinguishes them. + for _, m := range families[metrics.Prefix+"build_info"].GetMetric() { + for _, l := range m.GetLabel() { + if l.GetName() == "role" && l.GetValue() != metrics.RoleNode { + t.Errorf("role is %q, want %q", l.GetValue(), metrics.RoleNode) + } + } + } + // Nothing here names the node, the cluster, or anything in it. + for name, fam := range families { + for _, m := range fam.GetMetric() { + for _, l := range m.GetLabel() { + if !metrics.Allowed[metrics.LabelName(l.GetName())] { + t.Errorf("metric %q carries label %q", name, l.GetName()) + } + if l.GetValue() == "node-1" { + t.Errorf("metric %q names the node it runs on", name) + } + } + } + } + + cancel() + select { + case err := <-done: + if err != nil { + t.Fatalf("runNode returned %v", err) + } + case <-time.After(10 * time.Second): + t.Fatal("runNode did not stop") + } +} + +// TestTheReceiverIsAskedAboutPerScrapeNotAtStartup. The node-intake receiver is +// wired after the health listener, so a presence check made when the listener +// was built would leave `node_reports_rejected_total` permanently absent on +// every installation that has a DaemonSet — the one signal that says whether +// the controller is the reason a node went quiet (ADR 0067). +func TestTheReceiverIsAskedAboutPerScrapeNotAtStartup(t *testing.T) { + addr := freeAddress(t) + cfg := config.Config{Health: config.Health{ListenAddress: addr}} + cfg.Spool.Dir = t.TempDir() + cfg.NodeIntake.Enabled = true + cfg.NodeIntake.ListenAddress = freeAddress(t) + cfg.NodeIntake.ExpectedSubject = "system:serviceaccount:runtime-agent:runtime-agent-node" + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { + done <- run(ctx, slog.New(slog.NewTextHandler(io.Discard, nil)), + fake.NewClientset(), &rest.Config{}, cfg, config.Shape{}, time.Now()) + }() + select { + case err := <-done: + t.Fatalf("run stopped early: %v", err) + case <-time.After(200 * time.Millisecond): + } + waitFor(t, addr, health.ReadyPath, http.StatusOK) + + families := scrape(t, addr) + fam := families[metrics.Prefix+"node_reports_rejected_total"] + if fam == nil { + t.Fatalf("%snode_reports_rejected_total is absent although the receiver is enabled", + metrics.Prefix) + } + if n := len(fam.GetMetric()); n != 3 { + t.Errorf("%d rejection reasons, want 3 (unauthorized, too_large, malformed)", n) + } + cancel() + <-done +} diff --git a/cmd/agent/node.go b/cmd/agent/node.go index fc34495..9f3dfbb 100644 --- a/cmd/agent/node.go +++ b/cmd/agent/node.go @@ -6,12 +6,14 @@ import ( "fmt" "log/slog" "os" + "sync" "sync/atomic" "time" "github.com/RebuildStackCo/runtime-agent/internal/config" "github.com/RebuildStackCo/runtime-agent/internal/ebpfgate" "github.com/RebuildStackCo/runtime-agent/internal/health" + "github.com/RebuildStackCo/runtime-agent/internal/metrics" "github.com/RebuildStackCo/runtime-agent/internal/nodeprofile" "github.com/RebuildStackCo/runtime-agent/internal/nodescan" ) @@ -154,6 +156,17 @@ func runNode(ctx context.Context, logger *slog.Logger, args []string) error { beat := health.NewHeartbeat(time.Now(), nodeLivenessDeadline(*interval)) var passed atomic.Bool + // The latest pass's counters, kept so the metrics endpoint can state what + // this node did without the controller aggregating it away (ADR 0070 §4). + // A pass replaces them wholesale, exactly as a node's report replaces its + // contribution on the controller (ADR 0052 §3). + var lastScan struct { + mu sync.Mutex + counters nodescan.Counters + podsInScope int + done bool + } + scanOnce := func() { beat.Beat(time.Now()) defer passed.Store(true) @@ -203,6 +216,10 @@ func runNode(ctx context.Context, logger *slog.Logger, args []string) error { "containers_with_known_modules", moduleIndex.Size(), ) + lastScan.mu.Lock() + lastScan.counters, lastScan.podsInScope, lastScan.done = res.Counters, scope.Size(), true + lastScan.mu.Unlock() + // What the profiler beside this scanner did, on the scanner's cadence // because it is the only one that fires when the profiler does not // (ADR 0060 §1). Reported always, logged only where it was asked for: @@ -255,9 +272,20 @@ func runNode(ctx context.Context, logger *slog.Logger, args []string) error { } return true, "" } + sources := nodeSources{ + beat: beat, + scan: func() (nodescan.Counters, int, bool) { + lastScan.mu.Lock() + defer lastScan.mu.Unlock() + return lastScan.counters, lastScan.podsInScope, lastScan.done + }, + profiling: ebpfMetrics.snapshot, + shipped: shipper.counts, + } healthCtx, stopHealth := context.WithCancel(ctx) defer stopHealth() - srv := health.New(*healthAddress, live, ready, logger) + srv := health.New(*healthAddress, live, ready, + metrics.Handler(sources.gather, logger), logger) go func() { healthErr <- srv.Run(healthCtx) }() } diff --git a/cmd/agent/nodeship.go b/cmd/agent/nodeship.go index 6e8bac6..2b3c67f 100644 --- a/cmd/agent/nodeship.go +++ b/cmd/agent/nodeship.go @@ -9,6 +9,7 @@ import ( "net/http" "os" "strings" + "sync/atomic" "time" "github.com/RebuildStackCo/runtime-agent/internal/nodeintake" @@ -25,6 +26,20 @@ type reportShipper struct { tokenPath string node string client *http.Client + // What became of each delivery. A node whose reports never land is + // otherwise visible only in its own log, in the customer's cluster, where + // nobody looks until something prompts them (ADR 0067, ADR 0070 §4). + shipped atomic.Uint64 + failures atomic.Uint64 +} + +// counts returns deliveries that succeeded and deliveries that failed. A nil +// shipper is the log-only mode and has neither. +func (s *reportShipper) counts() (uint64, uint64) { + if s == nil { + return 0, 0 + } + return s.shipped.Load(), s.failures.Load() } // newReportShipper builds a shipper for endpoint, or returns nil when endpoint @@ -49,7 +64,15 @@ func newReportShipper(endpoint, tokenPath, node string) *reportShipper { // best-effort by design: a failure is logged by the caller and the next scan // pass retries — the controller reconstructs inventory from re-scans, so a lost // report costs nothing (ADR 0010, loss-harmless). -func (s *reportShipper) ship(ctx context.Context, res nodescan.Result, prof nodescan.ProfilingCoverage) error { +func (s *reportShipper) ship(ctx context.Context, res nodescan.Result, prof nodescan.ProfilingCoverage) (err error) { + defer func() { + if err != nil { + s.failures.Add(1) + return + } + s.shipped.Add(1) + }() + token, err := s.readToken() if err != nil { return fmt.Errorf("reading controller token: %w", err) diff --git a/docs/adr/0013-observation-completeness.md b/docs/adr/0013-observation-completeness.md index 9f5362e..c748114 100644 --- a/docs/adr/0013-observation-completeness.md +++ b/docs/adr/0013-observation-completeness.md @@ -3,7 +3,7 @@ Date: 2026-08-20 Status: Accepted Amends: 0006 §3, 0012 -Amended by: 0022 +Amended by: 0022, 0070 ## Context diff --git a/docs/adr/0069-liveness-is-the-process-readiness-is-the-caches.md b/docs/adr/0069-liveness-is-the-process-readiness-is-the-caches.md index cd55e3e..80bc13b 100644 --- a/docs/adr/0069-liveness-is-the-process-readiness-is-the-caches.md +++ b/docs/adr/0069-liveness-is-the-process-readiness-is-the-caches.md @@ -2,6 +2,7 @@ Date: 2026-09-04 Status: Accepted +Amended by: 0070 ## Context diff --git a/docs/adr/0070-the-agent-says-what-it-did-in-the-format-the-cluster-already-reads.md b/docs/adr/0070-the-agent-says-what-it-did-in-the-format-the-cluster-already-reads.md new file mode 100644 index 0000000..ef14a9a --- /dev/null +++ b/docs/adr/0070-the-agent-says-what-it-did-in-the-format-the-cluster-already-reads.md @@ -0,0 +1,207 @@ +# 0070. The agent says what it did, in the format the cluster already reads + +Date: 2026-09-06 + +Status: Accepted + +Amends: 0013 §3, 0069 §1 + +Adds a `/metrics` path to the health listener of both roles, exposing the +counters the agent already keeps. No new collection, no new privilege, no new +port, no payload byte changed, and no counter that did not already exist except +the spool's own. + +## Context + +The agent measures a cluster and says nothing about itself. What it does say +goes to two places, and neither reaches the person who is on call for it. + +**A structured log line.** It travels the customer's log pipeline, where an +alert on it means writing a query against a text field, per counter, per +cluster. Nobody does that for a vendor's agent. + +**`collection_coverage`.** It carries the same numbers to the backend, which is +the right audience for "what does this report rest on" and the wrong one for +"page me when collection stops" — and today it reaches nobody at all, because +the transport is not built (ADR 0054, `security.md` §8 Transport **[planned]**). +So the one artifact that exists to distinguish an empty cluster from a blind +agent is itself written to a spool nothing reads. + +The result is that the customer's SRE has no way to alert on the failure this +repository has spent six decisions making *observable*: a watch that stopped +being fed (ADR 0035), a kubelet path that fails while the other keeps working +(ADR 0013), a node that reported and then went quiet (ADR 0067), a fleet whose +profiler never loaded (ADR 0060), a spool that is evicting payloads faster than +they can be shipped (ADR 0042). Every one of those is computed, in memory, right +now. + +And the port question was settled a slice ago. ADR 0069 §1 put the probes on +9090 rather than beside 8080 with the reasoning stated in advance: *"9090 rather +than a number beside 8080, because the metrics endpoint of a later slice joins +this listener as a path. One port means one container port, one probe target and +one hole in the network policy, now and after metrics land."* This is that +slice, and it changes no chart template. + +## Decision + +**1. The exposition is `prometheus/common/expfmt`, and no new dependency.** + +The obvious choice is `client_golang`, and it is the wrong one twice over. + +It is a new module in a build gated on reachable vulnerabilities (ADR 0038): +every future finding in it, and in the `procfs` and `expfmt` code it pulls +behind it, becomes a red build for a package whose entire job is to render +numbers this agent already holds. Writing the text format by hand is worse — a +standard format reimplemented for no reason, which is code with no reason. + +There is a third option, and it is the one taken: `github.com/prometheus/common/expfmt` +and `github.com/prometheus/client_model` are **already direct dependencies**. +`internal/collector/cadvisor.go` *decodes* this exact format with +`expfmt.NewDecoder` to read the kubelet. Encoding with `expfmt.NewEncoder` adds +no module to the build, gives the reachability gate nothing new to gate, and the +format is still the reference implementation's rather than this repository's. + +The second reason is the one that decided it. `client_golang`'s shape is a +process-wide registry of counter objects that call sites increment. Adopting it +invites `metrics.PodsExcluded.Inc()` beside the filter's own counter, and then +there are two answers to "how many pods were excluded" that drift silently — +which is precisely what §2 exists to prevent. `expfmt` has no registry to +increment: a metric family can only be **built from a snapshot at scrape time**. +The rule stops being a convention a reviewer must remember and becomes the only +thing the API permits. + +**2. One set of counters, read by two audiences.** + +Every number on the endpoint comes from the same accessor `collection_coverage` +is assembled from — `Filter.Snapshot`, `PodWatcher.SourceHealths`, +`Store.Counters`, `Prober.Snapshot`, `Puller.Snapshot`, the intake rejections, +the heartbeat that `/livez` reads. There is no second counter anywhere in this +change, and the one place where the payload and the metrics wanted different +shapes was resolved by making the *source* finer and the payload a sum of it, +never by adding a number beside one. + +That place is the kubelet paths. ADR 0013 §3 states that `/stats/summary` and +`/metrics/cadvisor` are separate requests because they fail independently, and +then counts them together, so the payload cannot say which one is failing. The +poller now counts per path and `Observation()` returns the sums: the payload's +bytes are unchanged, and the endpoint gains the distinction the ADR always +claimed. This is the pattern for every future divergence — refine the source, +derive the coarse view — and never the reverse. + +The endpoint is the SRE's, the payload is the backend's, and they cannot +disagree, because there is nothing for them to disagree about. + +**3. Label names come from a closed set of nine; no value names anything in the +cluster.** + +`role`, `version`, `source`, `path`, `reason`, `outcome`, `kind`, `state`, +`subject`. `TestOnlyTheseLabelNamesAreAllowed` pins the list, the builder +refuses a name outside it, and a second test walks the *actual* exposition of +both roles — not the builder's guard — so a metric added with a tenth label +fails whether or not its author read this. + +This is a cardinality budget and the continuation of invariant 6 at the same +time, and the second is why it is a rule rather than advice. A namespace label +would put the identity of a collected object on a port that any pod in the +cluster can reach (§5 below), through a channel no payload review covers. The +counts here are the same aggregates `collection_coverage` carries, and they name +nothing for the same reason it names nothing. + +One value does not originate here: a node's profiling state crosses the channel +from the DaemonSet. It is mapped through the enumeration in ADR 0060 §2 with +everything else bucketed to `other`, so a node running a newer build cannot open +a series in the controller's exposition. + +`kind` is not enumerated in this package at all. Its values are +`sink.Registry()`, and the spool's per-kind counters are seeded from the same +list, so every registered kind has a series at zero before it has ever shipped — +which is the point, because *"no `usage_window` has been written in an hour"* is +the alert, and an absent series cannot express it. `Spool.write` now takes the +kind, refuses one the registry does not hold, and every caller passes the same +string it puts in the payload's bytes. The registry stays the one list (ADR +0022), and it is now load-bearing at run time rather than only in the goldens. + +**4. Both roles, and the node names itself nowhere.** + +ADR 0054 §4 decided that the coverage payload is aggregate over the fleet and +never per node, so "how many processes did the fleet skip" is answerable and +"which node stopped" is not — ADR 0067 then closed half of that with +`asserted_at`, which says a node went quiet without saying which one. + +A per-node scrape closes the other half without touching a payload. The node +identity comes from the customer's own service discovery, which already knows +which pod it scraped; the agent contributes no label for it. What the node +exposes is its own latest pass, its own profiler's state, and whether its +reports are reaching the controller — the last of which was invisible on the +node until now, since a delivery failure only ever reached its own log. + +`pass_start_timestamp_seconds` and `pass_deadline_seconds` come from the same +`Heartbeat` `/livez` reads, so an alert written on them fires on exactly the +condition the kubelet restarts the pod for — and, unlike the probe, before the +restart erases the evidence. + +**5. No chart change, and the exposure that follows from that.** + +The path joins the listener ADR 0069 already opened: same container port, same +probe target, same NetworkPolicy rule, no `ServiceMonitor` and no Prometheus +Operator CRD in the chart — the chart installs an agent, not somebody else's +monitoring stack, and a CRD it does not own is a dependency on a component half +of its installations do not run. + +The consequence is stated rather than left to be found (ADR 0039): ADR 0069 §5's +rule admits the health port from **any** source, because the caller is the +kubelet and no `podSelector` can name it. So any pod in the cluster can scrape +these numbers. That is the second reason §3 is a rule: what such a pod learns is +how many pods this agent observed and how many its filters excluded, in +aggregate, and nothing about which. Discovery is left to the operator's own +scrape configuration rather than to `prometheus.io/*` annotations, which the +major operator-based stacks do not honour and which would turn a default-on +annotation into a promise this chart cannot keep. + +**6. The spool's own numbers are new, and stop at the endpoint for now.** + +Four counters did not exist: payloads written per kind, writes that failed, +evictions by which bound caused them, and the size and file count the last sweep +measured. ADR 0042 gave the spool hard ceilings and no way to see them act — a +cluster whose payloads are being evicted oldest-first looks exactly like one +whose payloads are being kept, which is the failure that ADR most needed to be +visible. + +They are exposed and **not** added to `collection_coverage` in this slice. That +would be a protocol change, and it belongs with the transport that gives it a +reader. Because they are single-source by construction, adding the block later +reads the same snapshot and cannot diverge from what the endpoint serves. + +## Consequences + +**Easier.** "The agent stopped collecting" becomes an alert an SRE writes in one +line against a timestamp, with the threshold supplied by the agent rather than +guessed. Five failures this repository already knew how to detect and could only +write to a log become alertable: a watch no longer fed, one kubelet path failing +while the other works, a node that reported and went quiet, a fleet whose +profiler never loaded, a spool evicting under its ceiling. The endpoint works on +an installation with no backend at all, which is every installation today. + +**Harder / given up.** A second consumer now depends on the shape of these +counters. Renaming a metric is a breaking change for whoever alerts on it, in a +way that renaming an internal field never was — the counters were free to move +while only a log line and an unshipped payload read them, and they are not any +more. That is the cost of being usable. + +The exposition is built per scrape, from locks the flush pass also holds. A +scrape is a handful of map reads and cannot block collection meaningfully, but +it is not free, and a scrape interval below the collection cadence buys nothing: +most of these numbers change once a pass. + +Any pod on the cluster network can read the agent's aggregate counters, as §5 +sets out. The alternative was a source-restricted rule that cannot be written in +a template that does not know the cluster's node CIDR — the same wall ADR 0069 +§5 hit, with the same answer. + +**Not changed.** No payload byte, no golden file, no RBAC rule, no capability, +no mount, no chart template, no collection. `internal/sink/registry.go` gains no +row; it gains a second reader. `docs/security.md` §5 gains one row because a +port answers one more question, not because the agent learns anything new. + +This ADR records decisions implemented in the same pull request, per the process +in [`README.md`](README.md). diff --git a/docs/adr/README.md b/docs/adr/README.md index f21809e..28d49cf 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -127,3 +127,4 @@ line to paste. - [0067. A record carries the instant its source last stated it](0067-a-record-carries-the-age-of-its-source.md) - [0068. The agent yields first, and two objects it does not ship](0068-the-agent-yields-first.md) - [0069. Liveness is the process; readiness is what it has collected](0069-liveness-is-the-process-readiness-is-the-caches.md) +- [0070. The agent says what it did, in the format the cluster already reads](0070-the-agent-says-what-it-did-in-the-format-the-cluster-already-reads.md) diff --git a/docs/prose_test.go b/docs/prose_test.go index 38b5184..447daec 100644 --- a/docs/prose_test.go +++ b/docs/prose_test.go @@ -34,7 +34,7 @@ var ceilings = []struct { lines int section int }{ - {"security.md", 1044, 360}, + {"security.md", 1045, 360}, {"backend-requirements.md", 679, 536}, } diff --git a/docs/security.md b/docs/security.md index c3eb78c..2191f12 100644 --- a/docs/security.md +++ b/docs/security.md @@ -211,6 +211,7 @@ collection stage rather than at the API server, and the counts in | kubelet stats, proxied | controller → API server | Poll `/stats/summary` and `/metrics/cadvisor` on every kubelet for usage counters (ADR 0006) | Goes through the API server proxy (`nodes/proxy`, §4) — the agent opens **no direct connection to kubelets**. A direct-kubelet transport for very large clusters would be a documented change here, not a silent one | | Backend egress **[planned]** | controller → one fixed domain | Ship aggregated rollups and filtered profiles | mTLS, pinned domain. The only cross-boundary connection in the system, and it does not exist yet: the controller writes payloads to its local spool and ships nothing. A NetworkPolicy restricting controller egress to this domain plus in-cluster targets ships with the chart **[planned]** — the chart exists ([`charts/runtime-agent`](../charts/runtime-agent)), the policy does not | | Health probes | kubelet → both roles, in-cluster only | Answer whether the agent is working and whether it may be depended on yet ([ADR 0069](adr/0069-liveness-is-the-process-readiness-is-the-caches.md)) | **Port 9090 on the controller pod and on every node pod, in every profile** — separate from the report receiver above. Two paths, `/livez` and `/readyz`; GET and HEAD only, and nothing about a request changes what the agent does (§9). The reply is a status code and one line about the agent itself — whether its own collection pass is still running, whether its caches have filled — and names **no** object in your cluster. Not exposed by any Service. Where a NetworkPolicy ships it carries a second rule admitting this port from **any** source, because the caller is your kubelet arriving from the node's address, which no pod selector can name | +| Agent metrics | your monitoring → both roles, in-cluster only | Alert on the agent itself: whether it is still collecting, which of its reads are failing, what its filters excluded, what its spool evicted ([ADR 0070](adr/0070-the-agent-says-what-it-did-in-the-format-the-cluster-already-reads.md)) | **`/metrics` on the same port 9090 as the probes above, in every profile** — GET and HEAD only, Prometheus text format, and no `ServiceMonitor` or other Prometheus Operator resource is installed. Every number is one the agent already keeps and already reports in `collection_coverage` ([§11](#11-your-controls-and-what-the-agent-says-about-itself)); the two cannot disagree, because they read the same counters. Labels come from a **closed set of nine names**, and no value is ever a namespace, pod, node, workload or image name — the counts are the same aggregates the coverage payload carries. Reached by the same NetworkPolicy rule as the probes, so where a policy is enforced any pod in the cluster may scrape it; what it learns is how many objects this agent observed and how many its filters excluded, never which | | Node → controller reports | node → controller, in-cluster only | Deliver on-node Go build-info findings and captured profiles for aggregation (ADR 0010, ADR 0011), and ask which pods on this node passed your filters (ADR 0015) | Plain HTTP on the cluster network; the node always initiates and authenticates with a projected controller-audience ServiceAccount token, validated locally (no `TokenReview`). The report endpoints answer ack/error only. The **scope** query answers from the controller's already-filtered pod index, so it can only narrow what the node scans — it cannot name a pod your filters exclude. The **profiling-target** query is the one reply the node acts on; see [§7.2](#72-the-ebpf-cpu-profiler-opt-in-ebpf-profile-adr-0011) for what bounds it. The receiver is not reachable from outside the cluster, and a NetworkPolicy shipped with the chart restricts it to the node DaemonSet (ADR 0040 — it was described as shipping for months before it did, ADR 0039). Read that policy for what it is: it is enforced by your CNI, so on a cluster whose CNI does not implement NetworkPolicy it is inert, and it restricts who may be *heard*, never what a caller may *say*. The DaemonSet has **no** external egress. Honest disclosure: the token and the (already-filtered) facts travel in cleartext in-cluster, so a party who can sniff pod traffic can replay a node's token until it expires. What that buys is now bounded to **the node it was taken from**: the token names the node its bearer runs on, every endpoint refuses a request that speaks for a different one, and a fact can only join to a pod on the reporting node (ADR 0040). So a replayed or compromised node token can post fabricated inventory facts and profiles about workloads on *that* node, and read which of *that* node's pods passed your filters. It cannot read or write anything about any other node, it cannot reach the API, and it controls nothing | --- @@ -1001,14 +1002,14 @@ one; it was never decided in an ADR and never existed in the configuration ### What the agent says about itself -The agent tallies pods observed and pods excluded by each filter, and those -counts leave in `collection_coverage`, so filtering behavior is verifiable from -what we receive and not only from what you configured. The same payload reports -which of the agent's own reads worked — measured from its caches, not read back -from the rules, so a grant defeated by a webhook reads as failing rather than as -granted. No startup self-audit via `SelfSubjectRulesReview` is performed: it is a -`create` call, and this agent makes none (ADR 0054 §3). A 403 still degrades with -a log line, never a crash-loop or retry storm. +The agent tallies pods observed and pods excluded by each filter. Those counts +leave in `collection_coverage`, and the same counters are readable inside your +cluster at `/metrics` ([§5](#5-network-access)) — so filtering behavior is +verifiable by you and from what we receive, not only from what you configured. +Both report which of the agent's own reads worked, measured from its caches +rather than read back from the rules, so a grant defeated by a webhook reads as +failing. No `SelfSubjectRulesReview` self-audit is performed: that is a `create` +call (ADR 0054 §3). A 403 degrades with a log line, never a crash-loop. **What those counts can and cannot tell us.** They say how much was excluded and under which of your four controls, never which object. One inference remains and diff --git a/go.mod b/go.mod index 8307518..9a61165 100644 --- a/go.mod +++ b/go.mod @@ -8,6 +8,7 @@ require ( github.com/prometheus/client_model v0.6.2 github.com/prometheus/common v0.70.1 go.opentelemetry.io/ebpf-profiler v0.0.202633 + google.golang.org/protobuf v1.36.12 helm.sh/helm/v4 v4.2.4 k8s.io/api v0.37.0 k8s.io/apimachinery v0.37.0 @@ -147,7 +148,6 @@ require ( golang.org/x/vuln v1.7.0 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20260720211330-0afa2a65878a // indirect google.golang.org/grpc v1.83.1 // indirect - google.golang.org/protobuf v1.36.12 // indirect gopkg.in/evanphx/json-patch.v4 v4.13.0 // indirect gopkg.in/inf.v0 v0.9.1 // indirect k8s.io/apiextensions-apiserver v0.36.1 // indirect diff --git a/internal/collector/usage.go b/internal/collector/usage.go index d0610f0..e258e10 100644 --- a/internal/collector/usage.go +++ b/internal/collector/usage.go @@ -84,10 +84,33 @@ type UsagePoller struct { // (runtime probing, ADR 0006) and how the polling itself is going. Written // by the poll goroutine, read by whoever ships or logs a payload, so it is // the one part of the poller that is synchronized. - obsMu sync.Mutex - signals map[string]bool - pollsAttempted int64 - pollsFailed int64 + obsMu sync.Mutex + signals map[string]bool + polls map[KubeletPath]*pathCounts +} + +// KubeletPath is one of the exactly two stats paths the agent requests +// (docs/security.md §4). The two are counted apart because they fail +// independently, which ADR 0013 §3 states and the payload's totals could not +// show; a reader of the metrics endpoint sees which of them is failing. +type KubeletPath string + +// The two paths, as the metrics endpoint's `path` label carries them. +const ( + PathStatsSummary KubeletPath = "stats_summary" + PathMetricsCadvisor KubeletPath = "metrics_cadvisor" +) + +// KubeletPaths is the closed set, in the order they are read. +var KubeletPaths = []KubeletPath{PathStatsSummary, PathMetricsCadvisor} + +type pathCounts struct{ attempted, failed int64 } + +// PathReads is one path's request counts, cumulative since the agent started. +type PathReads struct { + Path KubeletPath + Attempted int64 + Failed int64 } type trackerKey struct { @@ -148,6 +171,7 @@ func NewUsagePoller( onError: onError, acc: rollup.NewAccumulator(UsageWindowLength), netAcc: rollup.NewNetworkAccumulator(UsageWindowLength), + polls: map[KubeletPath]*pathCounts{PathStatsSummary: {}, PathMetricsCadvisor: {}}, tracker: make(map[trackerKey]*counterState), throttle: make(map[trackerKey]*throttleState), netTracker: make(map[types.UID]*netCounterState), @@ -201,12 +225,29 @@ func (p *UsagePoller) signalsLocked() []string { func (p *UsagePoller) Observation() model.Observation { p.obsMu.Lock() defer p.obsMu.Unlock() - return model.Observation{ + obs := model.Observation{ PollIntervalSeconds: int64(usagePollInterval / time.Second), - PollsAttempted: p.pollsAttempted, - PollsFailed: p.pollsFailed, Signals: p.signalsLocked(), } + // The payload states the channel, the per-path counts state which half of + // it is failing; summing here keeps one set of counters behind both. + for _, c := range p.polls { + obs.PollsAttempted += c.attempted + obs.PollsFailed += c.failed + } + return obs +} + +// PathReads returns each kubelet path's request counts, in a stable order. +func (p *UsagePoller) PathReads() []PathReads { + p.obsMu.Lock() + defer p.obsMu.Unlock() + out := make([]PathReads, 0, len(KubeletPaths)) + for _, path := range KubeletPaths { + c := p.polls[path] + out = append(out, PathReads{Path: path, Attempted: c.attempted, Failed: c.failed}) + } + return out } func (p *UsagePoller) markSignal(name string) { @@ -215,15 +256,15 @@ func (p *UsagePoller) markSignal(name string) { p.obsMu.Unlock() } -// countPoll records one kubelet request and whether it failed. Both kubelet -// paths (/stats/summary and /metrics/cadvisor) count as requests of their own: -// they fail independently, and which one failed is resolved per record by the -// per-signal sample counts, not here. -func (p *UsagePoller) countPoll(failed bool) { +// countPoll records one request on one kubelet path and whether it failed. +// Which path lost a given record is still resolved by that record's per-signal +// sample counts, not by these totals (ADR 0013 §3). +func (p *UsagePoller) countPoll(path KubeletPath, failed bool) { p.obsMu.Lock() - p.pollsAttempted++ + c := p.polls[path] + c.attempted++ if failed { - p.pollsFailed++ + c.failed++ } p.obsMu.Unlock() } @@ -248,13 +289,13 @@ func (p *UsagePoller) pollOnce(ctx context.Context, now time.Time) { } for i := range reads { r := &reads[i] - p.countPoll(r.summaryErr != nil) + p.countPoll(PathStatsSummary, r.summaryErr != nil) if r.summaryErr != nil { p.reportError(r.node, r.summaryErr) } else { p.ingest(r.summary, now) } - p.countPoll(r.cadvisorErr != nil) + p.countPoll(PathMetricsCadvisor, r.cadvisorErr != nil) if r.cadvisorErr != nil { p.reportError(r.node, r.cadvisorErr) } else { diff --git a/internal/collector/usage_test.go b/internal/collector/usage_test.go index fe960ba..a516900 100644 --- a/internal/collector/usage_test.go +++ b/internal/collector/usage_test.go @@ -567,3 +567,66 @@ func TestNoNetworkBlockRaisesNoSignal(t *testing.T) { } } } + +// TestTheTwoKubeletPathsAreCountedApartAndTheirSumIsThePayload. ADR 0013 §3 +// states that the two paths are separate requests because they fail +// independently, and the payload's totals could not show which one did. The +// per-path counts are what the metrics endpoint reads; the payload keeps the +// same two numbers it always had, from the same counters (ADR 0070 §2). +func TestTheTwoKubeletPathsAreCountedApartAndTheirSumIsThePayload(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // Only the cadvisor path fails: the case a cluster-wide total hides. + if strings.HasSuffix(r.URL.Path, "metrics/cadvisor") { + w.WriteHeader(500) + return + } + w.WriteHeader(200) + _, _ = io.WriteString(w, `{"pods":[]}`) + })) + defer server.Close() + clientset, err := kubernetes.NewForConfig(&rest.Config{Host: server.URL}) + if err != nil { + t.Fatal(err) + } + p := NewUsagePoller(clientset, func() []string { return []string{"node-1", "node-2"} }, + webResolver(), nil, nil, nil, func(string, error) {}) + p.pollOnce(context.Background(), usageTestStart) + + want := map[KubeletPath]PathReads{ + PathStatsSummary: {Path: PathStatsSummary, Attempted: 2, Failed: 0}, + PathMetricsCadvisor: {Path: PathMetricsCadvisor, Attempted: 2, Failed: 2}, + } + reads := p.PathReads() + if len(reads) != len(want) { + t.Fatalf("got %d paths, want %d", len(reads), len(want)) + } + var attempted, failed int64 + for _, got := range reads { + if got != want[got.Path] { + t.Errorf("%s: %+v, want %+v", got.Path, got, want[got.Path]) + } + attempted += got.Attempted + failed += got.Failed + } + + obs := p.Observation() + if obs.PollsAttempted != attempted || obs.PollsFailed != failed { + t.Errorf("the payload says %d/%d and the paths sum to %d/%d; the two must be one set of counters", + obs.PollsAttempted, obs.PollsFailed, attempted, failed) + } + if obs.PollsAttempted != 4 || obs.PollsFailed != 2 { + t.Errorf("payload totals %d/%d, want 4/2 — two nodes, two paths each", obs.PollsAttempted, obs.PollsFailed) + } +} + +// TestPathReadsIsStablyOrdered so the exposition does not reorder series +// between scrapes. +func TestPathReadsIsStablyOrdered(t *testing.T) { + p := NewUsagePoller(nil, func() []string { return nil }, webResolver(), nil, nil, nil, func(string, error) {}) + for range 5 { + reads := p.PathReads() + if len(reads) != 2 || reads[0].Path != PathStatsSummary || reads[1].Path != PathMetricsCadvisor { + t.Fatalf("paths came back as %+v", reads) + } + } +} diff --git a/internal/health/heartbeat.go b/internal/health/heartbeat.go index 4dfee7d..e8b2c20 100644 --- a/internal/health/heartbeat.go +++ b/internal/health/heartbeat.go @@ -38,3 +38,16 @@ func (h *Heartbeat) Alive(now time.Time) (bool, time.Duration) { age := now.Sub(h.last) return age <= h.deadline, age } + +// Last is when the loop last reached the top of a pass, and Deadline how stale +// that may become. The metrics endpoint exposes both, so an alert on the stamp +// uses the agent's own threshold rather than one an operator guessed +// (ADR 0070 §4). +func (h *Heartbeat) Last() time.Time { + h.mu.Lock() + defer h.mu.Unlock() + return h.last +} + +// Deadline is how old a stamp may be before the role is no longer alive. +func (h *Heartbeat) Deadline() time.Duration { return h.deadline } diff --git a/internal/health/heartbeat_test.go b/internal/health/heartbeat_test.go index 098f5e5..83e10ac 100644 --- a/internal/health/heartbeat_test.go +++ b/internal/health/heartbeat_test.go @@ -29,3 +29,29 @@ func TestAStampOlderThanTheDeadlineIsNotAlive(t *testing.T) { t.Error("a beat did not revive the heartbeat") } } + +// TestTheStampAndTheDeadlineAreReadableForTheMetrics. An alert on the exposed +// stamp must fire on the same condition /livez answers no for, so both come +// from the heartbeat rather than from a second clock (ADR 0070 §4). +func TestTheStampAndTheDeadlineAreReadableForTheMetrics(t *testing.T) { + start := time.Unix(1_700_000_000, 0) + h := NewHeartbeat(start, 90*time.Second) + if got := h.Last(); !got.Equal(start) { + t.Errorf("Last() = %v, want %v", got, start) + } + if got := h.Deadline(); got != 90*time.Second { + t.Errorf("Deadline() = %v, want 90s", got) + } + + next := start.Add(time.Minute) + h.Beat(next) + if got := h.Last(); !got.Equal(next) { + t.Errorf("Last() = %v after a beat, want %v", got, next) + } + // The two agree by construction: the probe is not alive exactly when the + // stamp is older than the deadline the metric reports. + alive, _ := h.Alive(h.Last().Add(h.Deadline() + time.Second)) + if alive { + t.Error("a stamp past the exposed deadline still reads as alive") + } +} diff --git a/internal/health/server.go b/internal/health/server.go index e0fa2f2..750d168 100644 --- a/internal/health/server.go +++ b/internal/health/server.go @@ -1,8 +1,9 @@ // Package health serves the two questions the kubelet asks a pod about itself: -// is this process still working, and may it be depended on yet. +// is this process still working, and may it be depended on yet — and, on the +// same listener, what the agent has to say about its own work (ADR 0070). // -// It is one listener per role, on a port of its own, and it answers nothing -// else: no request body is read, no query is parsed, and no reply names a +// It is one listener per role, on a port of its own, and every path on it is +// read-only: no request body is read, no query is parsed, and no reply names a // cluster object (ADR 0069, CLAUDE.md invariants 1 and 6). package health @@ -16,11 +17,14 @@ import ( "time" ) -// The two paths, in the API server's vocabulary, because the distinction is the -// whole point: /livez is the process, /readyz is what it has collected. +// The probe paths, in the API server's vocabulary, because the distinction is +// the whole point: /livez is the process, /readyz is what it has collected. +// /metrics is the third, on this listener rather than a port of its own +// (ADR 0069 §1). const ( - LivePath = "/livez" - ReadyPath = "/readyz" + LivePath = "/livez" + ReadyPath = "/readyz" + MetricsPath = "/metrics" ) // Check answers one question. The string is why the answer is no, and it may @@ -36,11 +40,15 @@ type Server struct { logger *slog.Logger } -// New wires a listener on addr answering live at /livez and ready at /readyz. -func New(addr string, live, ready Check, logger *slog.Logger) *Server { +// New wires a listener on addr answering live at /livez, ready at /readyz, and +// — when metrics is non-nil — the agent's own numbers at /metrics. +func New(addr string, live, ready Check, metrics http.Handler, logger *slog.Logger) *Server { mux := http.NewServeMux() mux.Handle(LivePath, answer(live)) mux.Handle(ReadyPath, answer(ready)) + if metrics != nil { + mux.Handle(MetricsPath, metrics) + } return &Server{addr: addr, handler: mux, logger: logger} } @@ -84,7 +92,7 @@ func (s *Server) Run(ctx context.Context) error { serveErr := make(chan error, 1) go func() { s.logger.Info("health listener serving", "addr", s.addr, - "live", LivePath, "ready", ReadyPath) + "live", LivePath, "ready", ReadyPath, "metrics", MetricsPath) serveErr <- srv.ListenAndServe() }() diff --git a/internal/health/server_test.go b/internal/health/server_test.go index 48edae0..2531160 100644 --- a/internal/health/server_test.go +++ b/internal/health/server_test.go @@ -18,7 +18,7 @@ func testLogger() *slog.Logger { // serve returns a live server over the two checks, and the base URL to ask. func serve(t *testing.T, live, ready Check) string { t.Helper() - s := New("127.0.0.1:0", live, ready, testLogger()) + s := New("127.0.0.1:0", live, ready, nil, testLogger()) // httptest owns the listener so the test never picks a port; what is under // test is the handler wiring, which New built. srv := httptest.NewServer(s.handler) @@ -101,7 +101,7 @@ func TestAnUnknownPathIsNotAnAnswer(t *testing.T) { func TestTheListenerStopsWithItsContext(t *testing.T) { ok := func() (bool, string) { return true, "" } - s := New("127.0.0.1:0", ok, ok, testLogger()) + s := New("127.0.0.1:0", ok, ok, nil, testLogger()) ctx, cancel := context.WithCancel(context.Background()) done := make(chan error, 1) go func() { done <- s.Run(ctx) }() @@ -115,3 +115,49 @@ func TestTheListenerStopsWithItsContext(t *testing.T) { t.Fatal("Run did not return after its context was canceled") } } + +// TestMetricsJoinsTheListenerAsAThirdPath. ADR 0069 §1 chose one port so that +// there would be one container port, one probe target and one hole in the +// network policy after metrics landed; this is that path arriving. +func TestMetricsJoinsTheListenerAsAThirdPath(t *testing.T) { + ok := func() (bool, string) { return true, "" } + body := "# TYPE runtime_agent_spool_files gauge\nruntime_agent_spool_files 3\n" + metrics := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = io.WriteString(w, body) + }) + s := New("127.0.0.1:0", ok, ok, metrics, testLogger()) + srv := httptest.NewServer(s.handler) + t.Cleanup(srv.Close) + + resp, err := http.Get(srv.URL + MetricsPath) + if err != nil { + t.Fatal(err) + } + defer func() { _ = resp.Body.Close() }() + got, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatal(err) + } + if resp.StatusCode != http.StatusOK || string(got) != body { + t.Errorf("status %d body %q", resp.StatusCode, got) + } +} + +// TestWithoutAMetricsHandlerThePathIsNotThere: an agent run without metrics +// opens no path it was not asked for, the same way an empty listen address +// opens no listener (ADR 0069 §4). +func TestWithoutAMetricsHandlerThePathIsNotThere(t *testing.T) { + ok := func() (bool, string) { return true, "" } + s := New("127.0.0.1:0", ok, ok, nil, testLogger()) + srv := httptest.NewServer(s.handler) + t.Cleanup(srv.Close) + + resp, err := http.Get(srv.URL + MetricsPath) + if err != nil { + t.Fatal(err) + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode != http.StatusNotFound { + t.Errorf("status %d, want 404", resp.StatusCode) + } +} diff --git a/internal/metrics/collect.go b/internal/metrics/collect.go new file mode 100644 index 0000000..37967d4 --- /dev/null +++ b/internal/metrics/collect.go @@ -0,0 +1,305 @@ +package metrics + +import ( + "time" + + "github.com/RebuildStackCo/runtime-agent/internal/collector" + "github.com/RebuildStackCo/runtime-agent/internal/inventory" + "github.com/RebuildStackCo/runtime-agent/internal/model" + "github.com/RebuildStackCo/runtime-agent/internal/nodescan" + "github.com/RebuildStackCo/runtime-agent/internal/pprofprobe" + "github.com/RebuildStackCo/runtime-agent/internal/pprofpull" + "github.com/RebuildStackCo/runtime-agent/internal/sink" +) + +// The two roles, as the `role` label carries them. +const ( + RoleController = "controller" + RoleNode = "node" +) + +// ProfilingStates is the enumeration a node's profiling state is mapped +// through, from ADR 0060 §2. The state arrives over the wire from a node, so an +// unrecognized one becomes StateOther rather than a label value of its own: a +// node is not permitted to invent series in the controller's exposition. +var ProfilingStates = []string{ + "supported", "disabled", "program_load_failed", "capture_stopped", + "kernel_too_old", "btf_absent", "kernel_unknown", +} + +// StateOther is where a state outside the enumeration lands. +const StateOther = "other" + +func profilingState(state string) string { + for _, known := range ProfilingStates { + if state == known { + return state + } + } + return StateOther +} + +// Process is what both roles say about themselves: which build is running, and +// whether the loop that does the work is still turning. The stamp and its +// deadline are the ones /livez reads (ADR 0069 §2), so an alert on this metric +// fires on exactly the condition the kubelet restarts the pod for — and, unlike +// the probe, says so before the restart erases the evidence. +type Process struct { + Role string + Version string + PassStart time.Time + Deadline time.Duration +} + +func (s *Set) process(p Process) { + s.Gauge("build_info", "The agent build that is running, always 1.", 1, + L(LabelRole, p.Role), L(LabelVersion, p.Version)) + s.Gauge("pass_start_timestamp_seconds", + "When the role's own loop last reached the top of a pass.", + float64(p.PassStart.Unix())) + s.Gauge("pass_deadline_seconds", + "How stale that stamp may be before the role reports itself not alive.", + p.Deadline.Seconds()) +} + +// Controller is everything the controller exposes about its own work. Every +// field is a snapshot the coverage payload is also built from; nothing here is +// counted twice (ADR 0070 §2). +type Controller struct { + Process Process + Sources []model.SourceHealth + Kubelet []collector.PathReads + Filter model.Coverage + Placement model.PlacementDrops + Nodes model.NodeDrops + Spool sink.Counters + // Optional blocks, present when the feature they describe is installed. + Intake *model.IntakeRejections + Inventory *inventory.Counters + Scan *inventory.ScanCoverage + EBPF *inventory.ProfileCoverage + Probe *pprofprobe.Coverage + Pull *pprofpull.Coverage + ProfilesReceived uint64 + ProfilesUnjoined uint64 +} + +// CollectController builds the controller's exposition. +func CollectController(c Controller) *Set { + s := NewSet() + s.process(c.Process) + + for _, src := range c.Sources { + s.Gauge("source_synced", "Whether a watched source's cache ever filled.", + b2f(src.Synced), L(LabelSource, src.Name)) + s.Gauge("source_failing", "Whether a watched source's watch is erroring (ADR 0035).", + b2f(src.Failing), L(LabelSource, src.Name)) + } + + for _, r := range c.Kubelet { + s.Counter("kubelet_requests_total", "Kubelet stats requests attempted, by path.", + float64(r.Attempted), L(LabelPath, string(r.Path))) + s.Counter("kubelet_request_failures_total", "Kubelet stats requests that failed, by path.", + float64(r.Failed), L(LabelPath, string(r.Path))) + } + + f := c.Filter + s.Counter("pods_observed_total", "Pod appearances the filter considered.", float64(f.PodsObserved)) + excluded := "Pod appearances excluded from collection, by which control excluded them." + s.Counter("pods_excluded_total", excluded, float64(f.ExcludedNamespaceFilter), L(LabelReason, "namespace_filter")) + s.Counter("pods_excluded_total", excluded, float64(f.ExcludedNamespaceAnnotation), L(LabelReason, "namespace_annotation")) + s.Counter("pods_excluded_total", excluded, float64(f.ExcludedWorkloadAnnotation), L(LabelReason, "workload_annotation")) + s.Counter("pods_excluded_total", excluded, float64(f.ExcludedPodAnnotation), L(LabelReason, "pod_annotation")) + unchecked := "Pods collected without their workload's opt-out being checked (ADR 0028)." + s.Counter("pods_optout_unchecked_total", unchecked, float64(f.WorkloadUnknownKind), L(LabelReason, "unknown_kind")) + s.Counter("pods_optout_unchecked_total", unchecked, float64(f.WorkloadNotCached), L(LabelReason, "not_cached")) + + s.Counter("jobs_observed_total", "Finished Job runs the filter considered.", float64(f.JobsObserved)) + jobsExcluded := "Job runs excluded from collection, by which control excluded them." + s.Counter("jobs_excluded_total", jobsExcluded, float64(f.JobsExcludedNamespaceFilter), L(LabelReason, "namespace_filter")) + s.Counter("jobs_excluded_total", jobsExcluded, float64(f.JobsExcludedNamespaceAnnotation), L(LabelReason, "namespace_annotation")) + s.Counter("jobs_excluded_total", jobsExcluded, float64(f.JobsExcludedWorkloadAnnotation), L(LabelReason, "workload_annotation")) + s.Counter("jobs_excluded_total", jobsExcluded, float64(f.JobsExcludedAnnotation), L(LabelReason, "job_annotation")) + + drops := "Values the reductions refused to carry into a payload (ADR 0031, ADR 0064)." + s.Counter("reduction_drops_total", drops, float64(c.Placement.Values), L(LabelSubject, "placement_values")) + s.Counter("reduction_drops_total", drops, float64(c.Placement.Terms), L(LabelSubject, "placement_terms")) + s.Counter("reduction_drops_total", drops, float64(c.Nodes.Conditions), L(LabelSubject, "node_conditions")) + s.Counter("reduction_drops_total", drops, float64(c.Nodes.Devices), L(LabelSubject, "node_devices")) + s.Counter("reduction_drops_total", drops, float64(c.Nodes.Taints), L(LabelSubject, "node_taints")) + s.Counter("reduction_drops_total", drops, float64(c.Nodes.Values), L(LabelSubject, "node_values")) + + s.spool(c.Spool) + + if r := c.Intake; r != nil { + refused := "Node reports the receiver refused, by reason (ADR 0067)." + s.Counter("node_reports_rejected_total", refused, float64(r.Unauthorized), L(LabelReason, "unauthorized")) + s.Counter("node_reports_rejected_total", refused, float64(r.TooLarge), L(LabelReason, "too_large")) + s.Counter("node_reports_rejected_total", refused, float64(r.Malformed), L(LabelReason, "malformed")) + } + + if ic := c.Inventory; ic != nil { + s.Gauge("nodes_reporting", "Nodes that have delivered a scan report.", float64(ic.NodesReported)) + s.Gauge("inventory_records", "Distinct (namespace, workload, container) inventory records held.", + float64(ic.Records)) + facts := "On-node build facts received, and what became of each (ADR 0010 §5)." + s.Counter("inventory_facts_total", facts, float64(ic.FactsReceived), L(LabelOutcome, "received")) + s.Counter("inventory_facts_total", facts, float64(ic.FactsJoined), L(LabelOutcome, "joined")) + s.Counter("inventory_facts_total", facts, float64(ic.FactsUnjoined), L(LabelOutcome, "unjoined")) + s.Counter("inventory_facts_total", facts, float64(ic.FactsUndigested), L(LabelOutcome, "undigested")) + } + + if sc := c.Scan; sc != nil { + // A node that stops reporting keeps contributing its last pass, so this + // instant is what says one went quiet — the sums beside it cannot + // (ADR 0067). Zero before any node has reported. + if !sc.OldestAssertedAt.IsZero() { + s.Gauge("node_report_oldest_timestamp_seconds", + "When the stalest reporting node last stated its contribution.", + float64(sc.OldestAssertedAt.Unix())) + } + s.scanProcesses(nodescan.Counters{ + ProcessesScanned: sc.ProcessesScanned, + GoFound: sc.GoFound, + FilteredScope: sc.FilteredScope, + FilteredInfra: sc.FilteredInfra, + Unreadable: sc.Unreadable, + }) + } + + if pc := c.EBPF; pc != nil { + byState := map[string]int{} + for state, n := range pc.States { + byState[profilingState(state)] += n + } + for _, state := range append(append([]string{}, ProfilingStates...), StateOther) { + s.Gauge("ebpf_nodes", "Reporting nodes in each profiling state (ADR 0060 §2).", + float64(byState[state]), L(LabelState, state)) + } + s.ebpf(nodescan.ProfilingCoverage{ + Windows: pc.Windows, WindowsNoScope: pc.WindowsNoScope, + WindowsNoTargets: pc.WindowsNoTargets, WindowsNoSamples: pc.WindowsNoSamples, + ProfilesShipped: pc.ProfilesShipped, ProfilesInvalid: pc.ProfilesInvalid, + ProfilesUnshipped: pc.ProfilesUnshipped, SamplesOutOfScope: pc.SamplesOutOfScope, + ThirdPartyDropped: pc.ThirdPartyDropped, UnsymbolizedDropped: pc.UnsymbolizedDropped, + SamplesFiltered: pc.SamplesFiltered, + }) + // The controller's own two counts: a profile a node captured and this + // side could not join to a workload is lost here, after the work was done. + s.Counter("ebpf_profiles_received_total", "Captured profiles that reached the controller.", + float64(c.ProfilesReceived)) + s.Counter("ebpf_profiles_unjoined_total", "Captured profiles dropped for want of a pod to join them to.", + float64(c.ProfilesUnjoined)) + } + + if p := c.Probe; p != nil { + targets := "Workload endpoints by their latest answer to the pprof question (ADR 0057)." + s.Gauge("pprof_targets", targets, float64(p.Confirmed), L(LabelState, "confirmed")) + s.Gauge("pprof_targets", targets, float64(p.Absent), L(LabelState, "absent")) + s.Gauge("pprof_targets", targets, float64(p.Unreachable), L(LabelState, "unreachable")) + } + + if p := c.Pull; p != nil { + pulls := "Profile pulls by outcome; `refused` is a workload running its own profiler (ADR 0058 §3)." + s.Counter("pprof_pulls_total", pulls, float64(p.Shipped), L(LabelOutcome, "shipped")) + s.Counter("pprof_pulls_total", pulls, float64(p.Refused), L(LabelOutcome, "refused")) + s.Counter("pprof_pulls_total", pulls, float64(p.Unreachable), L(LabelOutcome, "unreachable")) + s.Counter("pprof_pulls_total", pulls, float64(p.Invalid), L(LabelOutcome, "invalid")) + } + return s +} + +func (s *Set) spool(c sink.Counters) { + for _, kind := range sink.SortedKeys(c.Written) { + s.Counter("spool_payloads_written_total", "Payloads written to the spool, by kind.", + float64(c.Written[kind]), L(LabelKind, kind)) + } + for _, kind := range sink.SortedKeys(c.WriteFailures) { + s.Counter("spool_write_failures_total", "Spool writes that failed, by payload kind.", + float64(c.WriteFailures[kind]), L(LabelKind, kind)) + } + for _, reason := range sink.SortedKeys(c.Evicted) { + s.Counter("spool_evicted_total", + "Payloads the sweep removed, by which bound removed them (ADR 0042).", + float64(c.Evicted[reason]), L(LabelReason, reason)) + } + s.Gauge("spool_bytes", "Bytes of payload the spool held at the last sweep.", float64(c.Bytes)) + s.Gauge("spool_files", "Payload files the spool held at the last sweep.", float64(c.Files)) + s.Gauge("spool_max_bytes", "The spool's byte ceiling.", float64(sink.DefaultMaxBytes)) + s.Gauge("spool_max_files", "The spool's file-count ceiling.", float64(sink.DefaultMaxFiles)) +} + +// scanProcesses is one scan pass's process counts — the fleet's latest passes +// summed on the controller, this node's latest on a node. A gauge either way: +// they describe one walk of a process table, not a running total (ADR 0054 §4). +func (s *Set) scanProcesses(c nodescan.Counters) { + walked := "Processes the latest scan pass walked, and what became of each." + s.Gauge("scan_processes", walked, float64(c.ProcessesScanned), L(LabelOutcome, "scanned")) + s.Gauge("scan_processes", walked, float64(c.GoFound), L(LabelOutcome, "go_found")) + s.Gauge("scan_processes", walked, float64(c.FilteredScope), L(LabelOutcome, "filtered_scope")) + s.Gauge("scan_processes", walked, float64(c.FilteredInfra), L(LabelOutcome, "filtered_infra")) + s.Gauge("scan_processes", walked, float64(c.Unreadable), L(LabelOutcome, "unreadable")) +} + +// ebpf is what a profiler did. Gauges rather than counters on the controller, +// because the sum is over the nodes reporting now and falls when one leaves; the +// same shape is used on the node, where the numbers reset with the process. +func (s *Set) ebpf(c nodescan.ProfilingCoverage) { + s.Gauge("ebpf_windows", "Capture windows cut.", float64(c.Windows)) + empty := "Capture windows that produced nothing, by reason (ADR 0060)." + s.Gauge("ebpf_windows_empty", empty, float64(c.WindowsNoScope), L(LabelReason, "no_scope")) + s.Gauge("ebpf_windows_empty", empty, float64(c.WindowsNoTargets), L(LabelReason, "no_targets")) + s.Gauge("ebpf_windows_empty", empty, float64(c.WindowsNoSamples), L(LabelReason, "no_samples")) + profiles := "Profiles built, by what became of each." + s.Gauge("ebpf_profiles", profiles, float64(c.ProfilesShipped), L(LabelOutcome, "shipped")) + s.Gauge("ebpf_profiles", profiles, float64(c.ProfilesInvalid), L(LabelOutcome, "invalid")) + s.Gauge("ebpf_profiles", profiles, float64(c.ProfilesUnshipped), L(LabelOutcome, "unshipped")) + dropped := "Samples and frames dropped, by reason; third_party rising means an allow-list that stopped matching (ADR 0059)." + s.Gauge("ebpf_samples_dropped", dropped, float64(c.SamplesOutOfScope), L(LabelReason, "out_of_scope")) + s.Gauge("ebpf_samples_dropped", dropped, float64(c.ThirdPartyDropped), L(LabelReason, "third_party")) + s.Gauge("ebpf_samples_dropped", dropped, float64(c.UnsymbolizedDropped), L(LabelReason, "unsymbolized")) + s.Gauge("ebpf_samples_dropped", dropped, float64(c.SamplesFiltered), L(LabelReason, "filtered")) +} + +// Node is what one node pod exposes. It answers the question the coverage +// payload deliberately cannot — which node — because the asking is done by the +// customer's own service discovery rather than by a label the agent attaches +// (ADR 0054 §4, ADR 0070 §4). +type Node struct { + Process Process + // Scanned is whether a pass has finished. Before one has, the scan counts + // are absent rather than zero: a zero is a claim, and "this node walked an + // empty process table" is not what a node that has not yet walked one means. + Scanned bool + Scan nodescan.Counters + PodsInScope int + Profiling nodescan.ProfilingCoverage + ReportsShipped uint64 + ReportFailures uint64 +} + +// CollectNode builds one node's exposition. +func CollectNode(n Node) *Set { + s := NewSet() + s.process(n.Process) + if n.Scanned { + s.scanProcesses(n.Scan) + s.Gauge("scan_pods_in_scope", "Pods the controller admitted this node to scan (ADR 0015).", + float64(n.PodsInScope)) + } + s.Gauge("ebpf_state", "This node's profiling state, always 1 for the state it is in.", 1, + L(LabelState, profilingState(n.Profiling.State))) + s.ebpf(n.Profiling) + s.Counter("node_reports_shipped_total", "Scan reports this node delivered to the controller.", + float64(n.ReportsShipped)) + s.Counter("node_report_failures_total", "Scan report deliveries that failed.", + float64(n.ReportFailures)) + return s +} + +func b2f(v bool) float64 { + if v { + return 1 + } + return 0 +} diff --git a/internal/metrics/collect_test.go b/internal/metrics/collect_test.go new file mode 100644 index 0000000..2345033 --- /dev/null +++ b/internal/metrics/collect_test.go @@ -0,0 +1,325 @@ +package metrics + +import ( + "sort" + "strings" + "testing" + "time" + + dto "github.com/prometheus/client_model/go" + + "github.com/RebuildStackCo/runtime-agent/internal/collector" + "github.com/RebuildStackCo/runtime-agent/internal/inventory" + "github.com/RebuildStackCo/runtime-agent/internal/model" + "github.com/RebuildStackCo/runtime-agent/internal/nodescan" + "github.com/RebuildStackCo/runtime-agent/internal/pprofprobe" + "github.com/RebuildStackCo/runtime-agent/internal/pprofpull" + "github.com/RebuildStackCo/runtime-agent/internal/sink" +) + +// fullController is every optional block present at once, with a value in every +// field, so the tests below walk the whole exposition rather than the half an +// unconfigured agent produces. +func fullController() Controller { + intake := model.IntakeRejections{Unauthorized: 1, TooLarge: 2, Malformed: 3} + counters := inventory.Counters{Records: 4, NodesReported: 2, FactsReceived: 9, FactsJoined: 8} + scan := inventory.ScanCoverage{ + Nodes: 2, OldestAssertedAt: time.Unix(1_700_000_000, 0), + ProcessesScanned: 100, GoFound: 10, FilteredScope: 1, FilteredInfra: 2, Unreadable: 3, + } + ebpf := inventory.ProfileCoverage{ + Nodes: 2, States: map[string]int{"supported": 1, "btf_absent": 1}, + Windows: 5, WindowsNoScope: 1, ProfilesShipped: 4, ThirdPartyDropped: 7, + } + probe := pprofprobe.Coverage{Confirmed: 3, Absent: 2, Unreachable: 1} + pull := pprofpull.Coverage{Shipped: 2, Refused: 1, Unreachable: 1, Invalid: 1} + spool := sink.Counters{ + Written: map[string]int64{}, + WriteFailures: map[string]int64{}, + Evicted: map[string]int64{}, + Bytes: 1024, Files: 12, + } + for _, k := range sink.Registry() { + spool.Written[k.Kind] = 1 + spool.WriteFailures[k.Kind] = 0 + } + for _, r := range sink.EvictionReasons { + spool.Evicted[r] = 1 + } + return Controller{ + Process: Process{Role: RoleController, Version: "1.2.3", + PassStart: time.Unix(1_700_000_100, 0), Deadline: 3 * time.Minute}, + Sources: []model.SourceHealth{{Name: "services", Synced: true}, {Name: "nodes", Failing: true}}, + Kubelet: []collector.PathReads{ + {Path: collector.PathStatsSummary, Attempted: 10, Failed: 1}, + {Path: collector.PathMetricsCadvisor, Attempted: 10, Failed: 4}, + }, + Filter: model.Coverage{PodsObserved: 412, ExcludedNamespaceFilter: 37, JobsObserved: 9}, + Placement: model.PlacementDrops{Values: 1, Terms: 2}, + Nodes: model.NodeDrops{Conditions: 1, Devices: 2, Taints: 3, Values: 4}, + Spool: spool, + Intake: &intake, + Inventory: &counters, + Scan: &scan, + EBPF: &ebpf, + Probe: &probe, + Pull: &pull, + ProfilesReceived: 6, + ProfilesUnjoined: 1, + } +} + +func fullNode() Node { + return Node{ + Process: Process{Role: RoleNode, Version: "1.2.3", + PassStart: time.Unix(1_700_000_100, 0), Deadline: 4 * time.Minute}, + Scanned: true, + Scan: nodescan.Counters{ProcessesScanned: 80, GoFound: 5, FilteredInfra: 3, Unreadable: 2}, + PodsInScope: 12, + Profiling: nodescan.ProfilingCoverage{ + State: "supported", Windows: 3, WindowsNoTargets: 1, + ProfilesShipped: 2, SamplesFiltered: 11, + }, + ReportsShipped: 7, + ReportFailures: 2, + } +} + +func families(t *testing.T, s *Set) []*dto.MetricFamily { + t.Helper() + out, err := s.Families() + if err != nil { + t.Fatalf("gathering: %v", err) + } + return out +} + +// TestTheRealExpositionCarriesOnlyPermittedLabels is the load-bearing one: it +// walks what the agent actually serves, not the builder's own guard, so a +// metric added with a label outside the set fails even if it never goes through +// a hand-written literal. +func TestTheRealExpositionCarriesOnlyPermittedLabels(t *testing.T) { + for _, tc := range []struct { + name string + set *Set + }{ + {"controller", CollectController(fullController())}, + {"node", CollectNode(fullNode())}, + } { + for _, fam := range families(t, tc.set) { + for _, m := range fam.GetMetric() { + for _, l := range m.GetLabel() { + if !Allowed[LabelName(l.GetName())] { + t.Errorf("%s: metric %q carries label %q, which is not permitted", + tc.name, fam.GetName(), l.GetName()) + } + } + } + } + } +} + +// TestNoLabelValueNamesAnythingFromTheCluster: the label names are closed, and +// their values are drawn from enumerations in the code — never from a string a +// cluster object supplied. This asserts it against a controller whose sources +// hold values that would be identities if any of them leaked through. +func TestNoLabelValueNamesAnythingFromTheCluster(t *testing.T) { + c := fullController() + // A node reporting a state this build does not know must not be able to + // invent a series in the controller's exposition (ADR 0070 §3). + c.EBPF.States = map[string]int{"prod-cluster-node-17": 1, "supported": 1} + + permitted := map[string]bool{} + for _, v := range ProfilingStates { + permitted[v] = true + } + for _, v := range []string{ + StateOther, RoleController, "1.2.3", "services", "nodes", + string(collector.PathStatsSummary), string(collector.PathMetricsCadvisor), + "namespace_filter", "namespace_annotation", "workload_annotation", "pod_annotation", + "job_annotation", "unknown_kind", "not_cached", + "placement_values", "placement_terms", "node_conditions", "node_devices", + "node_taints", "node_values", + "unauthorized", "too_large", "malformed", + "received", "joined", "unjoined", "undigested", + "scanned", "go_found", "filtered_scope", "filtered_infra", "unreadable", + "no_scope", "no_targets", "no_samples", "shipped", "invalid", "unshipped", + "out_of_scope", "third_party", "unsymbolized", "filtered", + "confirmed", "absent", "unreachable", "refused", + } { + permitted[v] = true + } + for _, k := range sink.Registry() { + permitted[k.Kind] = true + } + for _, r := range sink.EvictionReasons { + permitted[r] = true + } + + for _, fam := range families(t, CollectController(c)) { + for _, m := range fam.GetMetric() { + for _, l := range m.GetLabel() { + if !permitted[l.GetValue()] { + t.Errorf("metric %q carries label value %q, which comes from no enumeration in this package", + fam.GetName(), l.GetValue()) + } + } + } + } +} + +// TestThePayloadKindLabelIsTheRegistry: the label's values are the registry's +// rows, all of them, including the kinds this agent has never written. A kind +// that never ships reading as a zero is the whole point — "no usage_window has +// been written in an hour" is the alert (ADR 0070 §3). +func TestThePayloadKindLabelIsTheRegistry(t *testing.T) { + c := fullController() + // A spool that has written nothing at all still names every kind, because + // its counters are seeded from the registry. + c.Spool.Written = map[string]int64{} + for _, k := range sink.Registry() { + c.Spool.Written[k.Kind] = 0 + } + + var got []string + for _, fam := range families(t, CollectController(c)) { + if fam.GetName() != Prefix+"spool_payloads_written_total" { + continue + } + for _, m := range fam.GetMetric() { + for _, l := range m.GetLabel() { + if l.GetName() == string(LabelKind) { + got = append(got, l.GetValue()) + } + } + } + } + var want []string + for _, k := range sink.Registry() { + want = append(want, k.Kind) + } + sort.Strings(got) + sort.Strings(want) + if strings.Join(got, ",") != strings.Join(want, ",") { + t.Errorf("the kind label exposes\n %v\nand the registry holds\n %v\n"+ + "the registry is the one list (ADR 0022)", got, want) + } +} + +// TestAnUnconfiguredControllerOmitsWhatItDoesNotHave: a metrics-only install +// has no node channel, no inventory and no profiling, and must not report zeros +// for them — a zero is a claim, and "no DaemonSet" is not "no profiles". +func TestAnUnconfiguredControllerOmitsWhatItDoesNotHave(t *testing.T) { + bare := Controller{ + Process: Process{Role: RoleController, Version: "dev"}, + Spool: sink.Counters{Written: map[string]int64{}, WriteFailures: map[string]int64{}, Evicted: map[string]int64{}}, + } + present := map[string]bool{} + for _, fam := range families(t, CollectController(bare)) { + present[fam.GetName()] = true + } + for _, absent := range []string{ + "node_reports_rejected_total", "nodes_reporting", "inventory_records", + "inventory_facts_total", "scan_processes", "ebpf_nodes", "ebpf_windows", + "pprof_targets", "pprof_pulls_total", "node_report_oldest_timestamp_seconds", + } { + if present[Prefix+absent] { + t.Errorf("%s%s is exposed by an installation that has no such component", Prefix, absent) + } + } + for _, required := range []string{"build_info", "pass_start_timestamp_seconds", "spool_bytes"} { + if !present[Prefix+required] { + t.Errorf("%s%s is missing; every installation has it", Prefix, required) + } + } +} + +// TestTheLivenessStampIsTheOneTheProbeReads: the alert an operator writes on +// this metric must fire on the same condition the kubelet restarts the pod for, +// so both numbers come from the heartbeat rather than from a second clock. +func TestTheLivenessStampIsTheOneTheProbeReads(t *testing.T) { + c := fullController() + got := map[string]float64{} + for _, fam := range families(t, CollectController(c)) { + for _, m := range fam.GetMetric() { + got[fam.GetName()] = m.GetGauge().GetValue() + } + } + if want := float64(c.Process.PassStart.Unix()); got[Prefix+"pass_start_timestamp_seconds"] != want { + t.Errorf("pass stamp is %v, want %v", got[Prefix+"pass_start_timestamp_seconds"], want) + } + if want := c.Process.Deadline.Seconds(); got[Prefix+"pass_deadline_seconds"] != want { + t.Errorf("deadline is %v, want %v", got[Prefix+"pass_deadline_seconds"], want) + } +} + +// TestCountersAndGaugesAreTypedByWhetherTheyCanFall. Anything summed over the +// fleet's latest reports drops when a node leaves, so it is a gauge; only a +// number this process owns and never lowers is a counter, and only those carry +// the _total suffix. +func TestCountersAndGaugesAreTypedByWhetherTheyCanFall(t *testing.T) { + for _, set := range []*Set{CollectController(fullController()), CollectNode(fullNode())} { + for _, fam := range families(t, set) { + total := strings.HasSuffix(fam.GetName(), "_total") + counter := fam.GetType() == dto.MetricType_COUNTER + if total != counter { + t.Errorf("%q is a %s: the _total suffix and the counter type must agree", + fam.GetName(), fam.GetType()) + } + } + } +} + +// TestANodeStateOutsideTheEnumerationBecomesOther. The state crosses the +// channel from a node, and an unrecognized one must not open a label value of +// its own (ADR 0060 §2, ADR 0070 §3). +func TestANodeStateOutsideTheEnumerationBecomesOther(t *testing.T) { + n := fullNode() + n.Profiling.State = "something-a-newer-node-invented" + for _, fam := range families(t, CollectNode(n)) { + if fam.GetName() != Prefix+"ebpf_state" { + continue + } + for _, m := range fam.GetMetric() { + if v := m.GetLabel()[0].GetValue(); v != StateOther { + t.Errorf("unknown state exposed as %q, want %q", v, StateOther) + } + } + } + + c := fullController() + c.EBPF.States = map[string]int{"invented": 2, "supported": 1} + for _, fam := range families(t, CollectController(c)) { + if fam.GetName() != Prefix+"ebpf_nodes" { + continue + } + for _, m := range fam.GetMetric() { + if m.GetLabel()[0].GetValue() == StateOther && m.GetGauge().GetValue() != 2 { + t.Errorf("the two nodes in an unknown state did not land in %q", StateOther) + } + } + } +} + +// TestANodeSaysNothingAboutAScanItHasNotRun: zeroes before the first pass would +// read as a node that walked an empty process table. +func TestANodeSaysNothingAboutAScanItHasNotRun(t *testing.T) { + n := Node{Process: Process{Role: RoleNode, Version: "dev"}} + for _, fam := range families(t, CollectNode(n)) { + switch fam.GetName() { + case Prefix + "scan_processes", Prefix + "scan_pods_in_scope": + t.Errorf("%s is exposed before the first pass finished", fam.GetName()) + } + } + // What it does say is that it is alive and how it is configured, which is + // exactly the state a node in its first pass is in. + present := map[string]bool{} + for _, fam := range families(t, CollectNode(n)) { + present[fam.GetName()] = true + } + for _, required := range []string{"build_info", "pass_start_timestamp_seconds", "ebpf_state"} { + if !present[Prefix+required] { + t.Errorf("%s%s is missing", Prefix, required) + } + } +} diff --git a/internal/metrics/handler.go b/internal/metrics/handler.go new file mode 100644 index 0000000..1c3a5ab --- /dev/null +++ b/internal/metrics/handler.go @@ -0,0 +1,50 @@ +package metrics + +import ( + "log/slog" + "net/http" + + dto "github.com/prometheus/client_model/go" + "github.com/prometheus/common/expfmt" +) + +// Gatherer builds the families for one scrape. It is called per request, from +// the accessors the coverage payload reads, so there is nothing to keep in sync. +type Gatherer func() ([]*dto.MetricFamily, error) + +// Handler serves the text exposition. Like the probes beside it, it reads +// nothing about the request but its method: no body, no query, nothing a caller +// sends changes what the agent does (CLAUDE.md invariant 1). +// +// The encoder is prometheus/common's own — the same package the cAdvisor reader +// decodes with — so the format is the reference implementation's rather than +// this repository's (ADR 0070 §1). +func Handler(gather Gatherer, logger *slog.Logger) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet && r.Method != http.MethodHead { + w.Header().Set("Allow", "GET, HEAD") + http.Error(w, "read-only endpoint", http.StatusMethodNotAllowed) + return + } + families, err := gather() + if err != nil { + // A label name outside the permitted set is the only way here, and + // a test makes it unreachable. Serving nothing is the safe half of + // that: the reason is in the agent's log, not in the reply. + logger.Error("gathering agent metrics", "error", err) + http.Error(w, "metrics unavailable", http.StatusInternalServerError) + return + } + format := expfmt.NewFormat(expfmt.TypeTextPlain) + w.Header().Set("Content-Type", string(format)) + enc := expfmt.NewEncoder(w, format) + for _, fam := range families { + if err := enc.Encode(fam); err != nil { + // The response is already partly written; there is nobody left + // to tell but the log. + logger.Error("encoding agent metrics", "error", err, "metric", fam.GetName()) + return + } + } + }) +} diff --git a/internal/metrics/handler_test.go b/internal/metrics/handler_test.go new file mode 100644 index 0000000..e7c53e4 --- /dev/null +++ b/internal/metrics/handler_test.go @@ -0,0 +1,88 @@ +package metrics + +import ( + "io" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +func discardLogger() *slog.Logger { + return slog.New(slog.NewTextHandler(io.Discard, nil)) +} + +func ask(t *testing.T, h http.Handler, method string) *httptest.ResponseRecorder { + t.Helper() + rec := httptest.NewRecorder() + h.ServeHTTP(rec, httptest.NewRequest(method, MetricsPathForTest, nil)) + return rec +} + +// MetricsPathForTest keeps this package from importing the health package for +// one string; the path itself is health's to define. +const MetricsPathForTest = "/metrics" + +func TestTheExpositionIsTheTextFormat(t *testing.T) { + s := NewSet() + s.Counter("pods_observed_total", "Pod appearances the filter considered.", 412) + s.Gauge("spool_files", "Payload files held.", 12, L(LabelRole, RoleController)) + h := Handler(s.Families, discardLogger()) + + rec := ask(t, h, http.MethodGet) + if rec.Code != http.StatusOK { + t.Fatalf("status %d", rec.Code) + } + body := rec.Body.String() + for _, want := range []string{ + "# HELP runtime_agent_pods_observed_total Pod appearances the filter considered.", + "# TYPE runtime_agent_pods_observed_total counter", + "runtime_agent_pods_observed_total 412", + "# TYPE runtime_agent_spool_files gauge", + `runtime_agent_spool_files{role="controller"} 12`, + } { + if !strings.Contains(body, want) { + t.Errorf("exposition is missing %q:\n%s", want, body) + } + } + if ct := rec.Header().Get("Content-Type"); !strings.HasPrefix(ct, "text/plain") { + t.Errorf("content type is %q", ct) + } +} + +// TestOnlyGETAndHEADAreAnswered: nothing a caller sends changes what the agent +// does, and this endpoint is no exception (CLAUDE.md invariant 1). +func TestOnlyGETAndHEADAreAnswered(t *testing.T) { + h := Handler(NewSet().Families, discardLogger()) + for _, method := range []string{http.MethodPost, http.MethodPut, http.MethodDelete, http.MethodPatch} { + rec := ask(t, h, method) + if rec.Code != http.StatusMethodNotAllowed { + t.Errorf("%s answered with %d, want 405", method, rec.Code) + } + if allow := rec.Header().Get("Allow"); allow != "GET, HEAD" { + t.Errorf("%s: Allow is %q", method, allow) + } + } + for _, method := range []string{http.MethodGet, http.MethodHead} { + if rec := ask(t, h, method); rec.Code != http.StatusOK { + t.Errorf("%s answered with %d, want 200", method, rec.Code) + } + } +} + +// TestAGatherFailureServesNothing: a refused label name must not reach a +// scraper, and the reason belongs in the agent's log rather than in the reply. +func TestAGatherFailureServesNothing(t *testing.T) { + // The one way here: a label name outside the permitted set. + bad := NewSet() + bad.Gauge("pods_observed", "help", 1, L(LabelName("namespace"), "kube-system")) + h := Handler(bad.Families, discardLogger()) + rec := ask(t, h, http.MethodGet) + if rec.Code != http.StatusInternalServerError { + t.Fatalf("status %d, want 500", rec.Code) + } + if body := rec.Body.String(); strings.Contains(body, "namespace") || strings.Contains(body, "kube-system") { + t.Errorf("the refusal repeated what it refused: %q", body) + } +} diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go new file mode 100644 index 0000000..feb7269 --- /dev/null +++ b/internal/metrics/metrics.go @@ -0,0 +1,149 @@ +// Package metrics exposes what the agent did about its own work, in the +// Prometheus text format, for the customer's own monitoring. +// +// Two rules shape the whole package (ADR 0070). Every number here is read from +// the same accessor `collection_coverage` is built from, so the two audiences +// cannot be given different answers. And a label name comes from a closed set, +// so the endpoint can neither grow unbounded series nor become a second channel +// for identities the payloads do not carry (CLAUDE.md invariant 6). +package metrics + +import ( + "fmt" + "sort" + + dto "github.com/prometheus/client_model/go" + "google.golang.org/protobuf/proto" +) + +// Prefix every metric this agent exposes carries. It names the component being +// observed, which is what a prefix is for, and matches the chart and image. +const Prefix = "runtime_agent_" + +// LabelName is a label this agent may attach. The constants below are the whole +// permitted set; Allowed is what enforces it, and TestOnlyTheseLabelNamesAreAllowed +// pins the set so widening it is a decision in a diff. +// +// No value of any of these ever holds a namespace, pod, node, workload or image +// name: the closed set is a cardinality budget and a disclosure boundary at once. +type LabelName string + +const ( + // LabelRole is "controller" or "node", and LabelVersion the agent build. + // Both name the agent, never the cluster, and each takes one value per + // process lifetime. + LabelRole LabelName = "role" + // LabelVersion is the agent build. + LabelVersion LabelName = "version" + // LabelSource is a watched resource class ("services", "nodes"), as + // model.SourceHealth already reports it. + LabelSource LabelName = "source" + // LabelPath is one of the two kubelet stats paths (ADR 0013 §3). + LabelPath LabelName = "path" + // LabelReason is why something was excluded, evicted or refused. + LabelReason LabelName = "reason" + // LabelOutcome is which of several ends a piece of work reached. + LabelOutcome LabelName = "outcome" + // LabelKind is a payload kind, and its values come from + // internal/sink.Registry() rather than from a list kept here. + LabelKind LabelName = "kind" + // LabelState is a node's profiling state, mapped through the enumeration in + // ADR 0060 §2 because it arrives over the wire, or a pprof target's answer. + LabelState LabelName = "state" + // LabelSubject is which reduction dropped a value (ADR 0031, ADR 0064 §4). + LabelSubject LabelName = "subject" +) + +// Allowed is the closed set of label names, as a set for the check below and +// for the test that pins it. +var Allowed = map[LabelName]bool{ + LabelRole: true, LabelVersion: true, LabelSource: true, LabelPath: true, + LabelReason: true, LabelOutcome: true, LabelKind: true, LabelState: true, + LabelSubject: true, +} + +// Label is one dimension of one series. +type Label struct { + Name LabelName + Value string +} + +// L builds a label. It exists so a call site reads as one argument per +// dimension rather than as a struct literal per dimension. +func L(name LabelName, value string) Label { return Label{Name: name, Value: value} } + +// Set is metric families under construction. It is filled per scrape from +// snapshots and thrown away — there is deliberately no long-lived registry of +// counter objects to increment, because that is the shape in which a second +// counter appears beside the one the payload reads (ADR 0070 §2). +type Set struct { + order []string + families map[string]*dto.MetricFamily + err error +} + +// NewSet returns an empty set. +func NewSet() *Set { + return &Set{families: make(map[string]*dto.MetricFamily)} +} + +// Gauge adds one gauge series. A gauge is the right type for anything that can +// fall — including every count summed over the fleet's latest node reports, +// which drops when a node leaves (ADR 0067). +func (s *Set) Gauge(name, help string, value float64, labels ...Label) { + s.add(name, help, dto.MetricType_GAUGE, &dto.Metric{Gauge: &dto.Gauge{Value: proto.Float64(value)}}, labels) +} + +// Counter adds one counter series. Only for a number this process increments +// and never lowers; the name must end in _total. +func (s *Set) Counter(name, help string, value float64, labels ...Label) { + s.add(name, help, dto.MetricType_COUNTER, &dto.Metric{Counter: &dto.Counter{Value: proto.Float64(value)}}, labels) +} + +func (s *Set) add(name, help string, typ dto.MetricType, m *dto.Metric, labels []Label) { + full := Prefix + name + for _, l := range labels { + if !Allowed[l.Name] { + s.fail(fmt.Errorf("metric %s: label name %q is not in the permitted set (ADR 0070 §3)", full, l.Name)) + return + } + m.Label = append(m.Label, &dto.LabelPair{ + Name: proto.String(string(l.Name)), + Value: proto.String(l.Value), + }) + } + sort.Slice(m.Label, func(i, j int) bool { return m.Label[i].GetName() < m.Label[j].GetName() }) + + fam, ok := s.families[full] + if !ok { + fam = &dto.MetricFamily{Name: proto.String(full), Help: proto.String(help), Type: typ.Enum()} + s.families[full] = fam + s.order = append(s.order, full) + } else if fam.GetType() != typ { + s.fail(fmt.Errorf("metric %s is exposed as both %s and %s", full, fam.GetType(), typ)) + return + } + fam.Metric = append(fam.Metric, m) +} + +func (s *Set) fail(err error) { + if s.err == nil { + s.err = err + } +} + +// Families returns the set in a stable order, or the first refusal. A refused +// label name yields no exposition at all rather than a partial one: the caller +// serves an error, and the test that pins the permitted set is what keeps this +// unreachable. +func (s *Set) Families() ([]*dto.MetricFamily, error) { + if s.err != nil { + return nil, s.err + } + sort.Strings(s.order) + out := make([]*dto.MetricFamily, 0, len(s.order)) + for _, name := range s.order { + out = append(out, s.families[name]) + } + return out, nil +} diff --git a/internal/metrics/metrics_test.go b/internal/metrics/metrics_test.go new file mode 100644 index 0000000..11d7a9f --- /dev/null +++ b/internal/metrics/metrics_test.go @@ -0,0 +1,122 @@ +package metrics + +import ( + "sort" + "strings" + "testing" +) + +// TestOnlyTheseLabelNamesAreAllowed pins the closed set. A label name added to +// the package without a line here fails, which is the point: the set is a +// cardinality budget and the continuation of "identities do not leave by a +// second channel" at once (CLAUDE.md invariant 6, ADR 0070 §3). +func TestOnlyTheseLabelNamesAreAllowed(t *testing.T) { + want := []string{ + "kind", "outcome", "path", "reason", "role", "source", "state", + "subject", "version", + } + var got []string + for name := range Allowed { + got = append(got, string(name)) + } + sort.Strings(got) + if strings.Join(got, ",") != strings.Join(want, ",") { + t.Errorf("permitted label names are %v, pinned as %v; widening the set is a decision, "+ + "and one that must not admit a namespace, pod, node, workload or image name", got, want) + } +} + +// TestALabelOutsideTheSetYieldsNoExposition is the runtime half. The test above +// is what fails a build; this is what the endpoint does if one ever slips past +// it — serve nothing rather than a series naming something it should not. +func TestALabelOutsideTheSetYieldsNoExposition(t *testing.T) { + for _, name := range []LabelName{"namespace", "pod", "node", "workload", "image", ""} { + s := NewSet() + s.Gauge("pods_observed", "help", 1, L(name, "anything")) + s.Gauge("spool_files", "help", 3) + families, err := s.Families() + if err == nil { + t.Errorf("label %q was accepted", name) + } + if families != nil { + t.Errorf("label %q left %d families exposed; a refusal exposes nothing", name, len(families)) + } + } +} + +func TestEveryNameCarriesThePrefix(t *testing.T) { + s := NewSet() + s.Gauge("spool_files", "help", 1) + s.Counter("pods_observed_total", "help", 2) + families, err := s.Families() + if err != nil { + t.Fatal(err) + } + for _, f := range families { + if !strings.HasPrefix(f.GetName(), Prefix) { + t.Errorf("metric %q does not carry the %q prefix", f.GetName(), Prefix) + } + } +} + +// TestOneNameIsOneFamily: repeated adds under one name become one family with +// many series, which is what the text format requires. +func TestOneNameIsOneFamily(t *testing.T) { + s := NewSet() + s.Counter("pods_excluded_total", "help", 1, L(LabelReason, "namespace_filter")) + s.Counter("pods_excluded_total", "help", 2, L(LabelReason, "pod_annotation")) + families, err := s.Families() + if err != nil { + t.Fatal(err) + } + if len(families) != 1 { + t.Fatalf("got %d families, want 1", len(families)) + } + if n := len(families[0].GetMetric()); n != 2 { + t.Errorf("got %d series, want 2", n) + } +} + +// TestOneNameCannotBeTwoTypes: a name exposed as both a counter and a gauge is +// an exposition no scraper can read, so it is refused rather than served. +func TestOneNameCannotBeTwoTypes(t *testing.T) { + s := NewSet() + s.Counter("spool_evicted_total", "help", 1, L(LabelReason, "age")) + s.Gauge("spool_evicted_total", "help", 1, L(LabelReason, "bytes")) + if _, err := s.Families(); err == nil { + t.Error("a name exposed as two types was accepted") + } +} + +func TestFamiliesAreOrderedStably(t *testing.T) { + s := NewSet() + s.Gauge("spool_files", "help", 1) + s.Gauge("build_info", "help", 1, L(LabelRole, RoleNode), L(LabelVersion, "dev")) + s.Gauge("nodes_reporting", "help", 1) + families, err := s.Families() + if err != nil { + t.Fatal(err) + } + var names []string + for _, f := range families { + names = append(names, f.GetName()) + } + if !sort.StringsAreSorted(names) { + t.Errorf("families are not in a stable order: %v", names) + } +} + +// TestLabelsAreOrderedWithinASeries keeps the bytes of one scrape comparable to +// the next, which a map's iteration order would not. +func TestLabelsAreOrderedWithinASeries(t *testing.T) { + s := NewSet() + s.Gauge("build_info", "help", 1, L(LabelVersion, "1.2.3"), L(LabelRole, RoleController)) + families, err := s.Families() + if err != nil { + t.Fatal(err) + } + labels := families[0].GetMetric()[0].GetLabel() + if len(labels) != 2 || labels[0].GetName() != "role" || labels[1].GetName() != "version" { + t.Errorf("labels are not sorted by name: %v", labels) + } +} diff --git a/internal/sink/counters.go b/internal/sink/counters.go new file mode 100644 index 0000000..fcba0e1 --- /dev/null +++ b/internal/sink/counters.go @@ -0,0 +1,118 @@ +package sink + +import "sort" + +// Why a payload left the spool. The age cutoff and the orphaned temp file are +// unconditional; bytes and files are the ceilings acting, and those two rising +// mean the backend is missing payloads that were written (ADR 0042). +const ( + EvictedAge = "age" + EvictedOrphanTemp = "orphan_temp" + EvictedBytes = "bytes" + EvictedFiles = "files" +) + +// EvictionReasons is the closed set, for whoever renders it. +var EvictionReasons = []string{EvictedAge, EvictedOrphanTemp, EvictedBytes, EvictedFiles} + +// Counters is what the spool did, cumulative since the process started, plus +// what the last sweep measured. Nothing here reaches a payload today: it is the +// agent's own operational state, and the shape of a spool block in +// `collection_coverage` is a protocol decision of its own (ADR 0070 §5). +type Counters struct { + // Written and WriteFailures are per payload kind, and hold a zero entry for + // every kind in the registry — "no usage_window has ever been written" is + // the reading that matters, and it needs the series to exist. + Written map[string]int64 + WriteFailures map[string]int64 + // Evicted is per reason, with a zero entry for each. + Evicted map[string]int64 + // Bytes and Files are what the last sweep counted after evicting, against + // DefaultMaxBytes and DefaultMaxFiles. Zero before the first sweep runs. + Bytes int64 + Files int64 +} + +// countWrite records one payload written, or one write that failed. +func (s *Spool) countWrite(kind string, failed bool) { + s.countMu.Lock() + defer s.countMu.Unlock() + if failed { + s.writeFailures[kind]++ + return + } + s.written[kind]++ +} + +func (s *Spool) countEviction(reason string) { + s.countMu.Lock() + defer s.countMu.Unlock() + s.evicted[reason]++ +} + +func (s *Spool) recordSweep(bytes int64, files int) { + s.countMu.Lock() + defer s.countMu.Unlock() + s.bytes, s.files = bytes, int64(files) +} + +// Counters returns a copy of the spool's own counters, safe to read while the +// flush goroutine writes. A nil spool is the log-only mode: it has written +// nothing, and says so with every registered kind at zero rather than with +// silence. +func (s *Spool) Counters() Counters { + if s == nil { + return Counters{ + Written: newKindCounts(), + WriteFailures: newKindCounts(), + Evicted: newReasonCounts(), + } + } + s.countMu.Lock() + defer s.countMu.Unlock() + return Counters{ + Written: copyCounts(s.written), + WriteFailures: copyCounts(s.writeFailures), + Evicted: copyCounts(s.evicted), + Bytes: s.bytes, + Files: s.files, + } +} + +func copyCounts(in map[string]int64) map[string]int64 { + out := make(map[string]int64, len(in)) + for k, v := range in { + out[k] = v + } + return out +} + +// newKindCounts seeds a per-kind map from the registry, so the list of payload +// kinds has exactly one home (ADR 0022) and a kind that never ships is visible +// as a zero rather than as an absence. +func newKindCounts() map[string]int64 { + out := make(map[string]int64, len(registry)) + for _, k := range registry { + out[k.Kind] = 0 + } + return out +} + +func newReasonCounts() map[string]int64 { + out := make(map[string]int64, len(EvictionReasons)) + for _, r := range EvictionReasons { + out[r] = 0 + } + return out +} + +// SortedKeys returns a map's keys in a stable order, so an exposition built +// from one of these maps does not reorder between scrapes. +func SortedKeys(in map[string]int64) []string { + out := make([]string, 0, len(in)) + for k := range in { + out = append(out, k) + } + sort.Strings(out) + return out +} diff --git a/internal/sink/counters_test.go b/internal/sink/counters_test.go new file mode 100644 index 0000000..c71253b --- /dev/null +++ b/internal/sink/counters_test.go @@ -0,0 +1,154 @@ +package sink + +import ( + "os" + "path/filepath" + "testing" + "time" + + "github.com/RebuildStackCo/runtime-agent/internal/model" +) + +// TestEveryRegisteredKindHasASeriesBeforeItShips. A kind that has never been +// written must read as a zero rather than as an absence: "no usage_window has +// been written in an hour" is the reading an operator alerts on, and it needs +// the series to exist (ADR 0070 §3). +func TestEveryRegisteredKindHasASeriesBeforeItShips(t *testing.T) { + s, _ := newTestSpool(t) + c := s.Counters() + for _, kind := range Registry() { + if _, ok := c.Written[kind.Kind]; !ok { + t.Errorf("kind %q has no written counter", kind.Kind) + } + if _, ok := c.WriteFailures[kind.Kind]; !ok { + t.Errorf("kind %q has no failure counter", kind.Kind) + } + } + if len(c.Written) != len(Registry()) { + t.Errorf("the spool counts %d kinds and the registry holds %d; the registry is the one list (ADR 0022)", + len(c.Written), len(Registry())) + } + for _, reason := range EvictionReasons { + if _, ok := c.Evicted[reason]; !ok { + t.Errorf("eviction reason %q has no counter", reason) + } + } +} + +// TestAKindWithNoRegistryRowIsRefused makes the registry load-bearing at run +// time, not only in the golden tests: a payload kind added without a row does +// not reach the spool at all. +func TestAKindWithNoRegistryRowIsRefused(t *testing.T) { + s, dir := newTestSpool(t) + err := s.write("usage_forecast", "usage-forecast.json", struct{}{}) + if err == nil { + t.Fatal("a kind with no registry row was written") + } + entries, readErr := os.ReadDir(dir) + if readErr != nil { + t.Fatal(readErr) + } + if len(entries) != 0 { + t.Errorf("the refused write left %d entries behind", len(entries)) + } + if _, ok := s.Counters().Written["usage_forecast"]; ok { + t.Error("the refused kind opened a counter of its own") + } +} + +// TestAWrittenPayloadIsCountedUnderTheKindInItsBytes: the label and the +// discriminator are the same string, so the two cannot disagree. +func TestAWrittenPayloadIsCountedUnderTheKindInItsBytes(t *testing.T) { + s, _ := newTestSpool(t) + if err := s.WriteCollectionCoverage(time.Now(), time.Now(), AgentInfo{}, nil, + model.Coverage{}, model.PlacementDrops{}, model.NodeDrops{}, + nil, nil, nil, nil, nil, nil); err != nil { + t.Fatal(err) + } + if got := s.Counters().Written["collection_coverage"]; got != 1 { + t.Errorf("collection_coverage counted %d times, want 1", got) + } + for kind, n := range s.Counters().Written { + if kind != "collection_coverage" && n != 0 { + t.Errorf("%s counted %d times; only one payload was written", kind, n) + } + } +} + +// TestASweepCountsWhatItRemovedAndWhatItLeft. Nothing kept these numbers +// before: the spool enforced its bounds silently, so a cluster whose payloads +// were being evicted looked identical to one whose payloads were being kept +// (ADR 0042, ADR 0070 §5). +func TestASweepCountsWhatItRemovedAndWhatItLeft(t *testing.T) { + s, dir := newTestSpool(t) + now := time.Now() + + // Two payloads past the age cutoff, one orphaned temp file, one current. + old := now.Add(-2 * DefaultMaxAge) + for _, f := range []struct { + name string + mod time.Time + }{ + {"expired-a.json", old}, + {"expired-b.json", old}, + {"orphan.json.tmp", now.Add(-2 * time.Minute)}, + {"current.json", now}, + } { + path := filepath.Join(dir, f.name) + if err := os.WriteFile(path, []byte("{}\n"), 0o600); err != nil { + t.Fatal(err) + } + if err := os.Chtimes(path, f.mod, f.mod); err != nil { + t.Fatal(err) + } + } + + if err := s.Sweep(now); err != nil { + t.Fatal(err) + } + c := s.Counters() + if c.Evicted[EvictedAge] != 2 { + t.Errorf("age evictions: %d, want 2", c.Evicted[EvictedAge]) + } + if c.Evicted[EvictedOrphanTemp] != 1 { + t.Errorf("orphan temp files: %d, want 1", c.Evicted[EvictedOrphanTemp]) + } + if c.Files != 1 { + t.Errorf("files left: %d, want 1", c.Files) + } + if c.Bytes != 3 { + t.Errorf("bytes left: %d, want 3", c.Bytes) + } +} + +// TestTheFileCeilingSaysWhichBoundEvicted: bytes and files answer different +// questions — more payload than the volume holds, against more payloads than +// the count allows — so the reason distinguishes them (ADR 0042 §2). +func TestTheFileCeilingSaysWhichBoundEvicted(t *testing.T) { + s, dir := newTestSpool(t) + s.maxFiles = 2 + now := time.Now() + for i, name := range []string{"a.json", "b.json", "c.json", "d.json"} { + path := filepath.Join(dir, name) + if err := os.WriteFile(path, []byte("{}\n"), 0o600); err != nil { + t.Fatal(err) + } + mod := now.Add(time.Duration(i) * time.Second) + if err := os.Chtimes(path, mod, mod); err != nil { + t.Fatal(err) + } + } + if err := s.Sweep(now.Add(time.Minute)); err != nil { + t.Fatal(err) + } + c := s.Counters() + if c.Evicted[EvictedFiles] != 2 { + t.Errorf("file-ceiling evictions: %d, want 2", c.Evicted[EvictedFiles]) + } + if c.Evicted[EvictedBytes] != 0 { + t.Errorf("byte-ceiling evictions: %d, want 0", c.Evicted[EvictedBytes]) + } + if c.Files != 2 { + t.Errorf("files left: %d, want 2", c.Files) + } +} diff --git a/internal/sink/spool.go b/internal/sink/spool.go index 0727de8..873021a 100644 --- a/internal/sink/spool.go +++ b/internal/sink/spool.go @@ -14,6 +14,7 @@ import ( "path/filepath" "sort" "strings" + "sync" "time" "github.com/RebuildStackCo/runtime-agent/internal/config" @@ -97,6 +98,16 @@ type Spool struct { maxAge time.Duration maxBytes int64 maxFiles int + + // The spool's own counters, and the only lock it holds. It guards these + // numbers and nothing else: writing a payload is still unsynchronized, and + // two writers of one key still race for its temp file (ADR 0070 §5). + countMu sync.Mutex + written map[string]int64 + writeFailures map[string]int64 + evicted map[string]int64 + bytes int64 + files int64 } // NewSpool opens (creating if needed) the spool directory. maxAge ≤ 0 @@ -108,7 +119,12 @@ func NewSpool(dir string, maxAge time.Duration) (*Spool, error) { if err := os.MkdirAll(dir, 0o700); err != nil { return nil, fmt.Errorf("creating spool directory: %w", err) } - return &Spool{dir: dir, maxAge: maxAge, maxBytes: DefaultMaxBytes, maxFiles: DefaultMaxFiles}, nil + return &Spool{ + dir: dir, maxAge: maxAge, maxBytes: DefaultMaxBytes, maxFiles: DefaultMaxFiles, + written: newKindCounts(), + writeFailures: newKindCounts(), + evicted: newReasonCounts(), + }, nil } // usagePayload is one shippable batch: every record of one wall-clock @@ -549,7 +565,7 @@ func (s *Spool) WriteUsageSnapshot(records []*rollup.Record, obs model.Observati Observation: obs, Records: group, } - if err := s.write(k.name()+".snapshot.json", payload); err != nil { + if err := s.write(payload.Kind, k.name()+".snapshot.json", payload); err != nil { return err } } @@ -569,7 +585,7 @@ func (s *Spool) WriteClosedWindows(records []*rollup.Record, obs model.Observati Observation: obs, Records: group, } - if err := s.write(k.name()+".json", payload); err != nil { + if err := s.write(payload.Kind, k.name()+".json", payload); err != nil { return err } if err := os.Remove(filepath.Join(s.dir, k.name()+".snapshot.json")); err != nil && !os.IsNotExist(err) { @@ -596,7 +612,7 @@ func (s *Spool) WriteNetworkWindows(records []*rollup.NetworkRecord, obs model.O Observation: obs, Records: group, } - if err := s.write(fmt.Sprintf("network-%d-%d.json", k.start.Unix(), k.seconds), payload); err != nil { + if err := s.write(payload.Kind, fmt.Sprintf("network-%d-%d.json", k.start.Unix(), k.seconds), payload); err != nil { return err } } @@ -625,7 +641,7 @@ func (s *Spool) WriteCollectionCoverage(capturedAt, since time.Time, agent Agent Intake: intake, PprofPull: pull, } - return s.write("collection-coverage.json", payload) + return s.write(payload.Kind, "collection-coverage.json", payload) } // WriteContainerRestarts writes the restart records of each window they belong @@ -650,7 +666,7 @@ func (s *Spool) WriteContainerRestarts(records []journal.RestartRecord) error { WindowSeconds: k.seconds, Records: group, } - if err := s.write(fmt.Sprintf("restarts-%d-%d.json", k.start.Unix(), k.seconds), payload); err != nil { + if err := s.write(payload.Kind, fmt.Sprintf("restarts-%d-%d.json", k.start.Unix(), k.seconds), payload); err != nil { return err } } @@ -679,7 +695,7 @@ func (s *Spool) WritePodDisruptions(records []journal.DisruptionRecord) error { WindowSeconds: k.seconds, Records: group, } - if err := s.write(fmt.Sprintf("disruptions-%d-%d.json", k.start.Unix(), k.seconds), payload); err != nil { + if err := s.write(payload.Kind, fmt.Sprintf("disruptions-%d-%d.json", k.start.Unix(), k.seconds), payload); err != nil { return err } } @@ -708,7 +724,7 @@ func (s *Spool) WriteNodeLifecycle(records []journal.NodeEventRecord) error { WindowSeconds: k.seconds, Records: group, } - if err := s.write(fmt.Sprintf("node-lifecycle-%d-%d.json", k.start.Unix(), k.seconds), payload); err != nil { + if err := s.write(payload.Kind, fmt.Sprintf("node-lifecycle-%d-%d.json", k.start.Unix(), k.seconds), payload); err != nil { return err } } @@ -721,7 +737,8 @@ func (s *Spool) WriteNodeLifecycle(records []journal.NodeEventRecord) error { func (s *Spool) WriteOOMKill(e model.OOMKill) error { name := fmt.Sprintf("oom-%d-%s-%s-%s-%d.json", e.FinishedAt.Unix(), fileToken(e.Namespace), fileToken(e.Pod), fileToken(e.Container), e.RestartCount) - return s.write(name, oomPayload{Kind: "oom_kill", Source: SourceJournal, Event: e}) + payload := oomPayload{Kind: "oom_kill", Source: SourceJournal, Event: e} + return s.write(payload.Kind, name, payload) } // WriteRestartCounters writes the current restart-counter reading of every @@ -741,7 +758,7 @@ func (s *Spool) WriteRestartCounters(capturedAt time.Time, records []model.Resta CapturedAt: capturedAt.UTC(), Records: records, } - return s.write("restart-counters.json", payload) + return s.write(payload.Kind, "restart-counters.json", payload) } // WriteJobRuns writes the finished Job runs of each window they belong to, one @@ -766,7 +783,7 @@ func (s *Spool) WriteJobRuns(records []journal.JobRunRecord) error { WindowSeconds: k.seconds, Records: group, } - if err := s.write(fmt.Sprintf("job-runs-%d-%d.json", k.start.Unix(), k.seconds), payload); err != nil { + if err := s.write(payload.Kind, fmt.Sprintf("job-runs-%d-%d.json", k.start.Unix(), k.seconds), payload); err != nil { return err } } @@ -787,7 +804,7 @@ func (s *Spool) WriteGoInventory(capturedAt time.Time, cov inventory.Coverage, r Coverage: cov, Records: records, } - return s.write("go-inventory.json", payload) + return s.write(payload.Kind, "go-inventory.json", payload) } // WriteProcessPeaks writes the current peak records as one superseding batch. @@ -800,7 +817,7 @@ func (s *Spool) WriteProcessPeaks(capturedAt time.Time, records []inventory.Peak CapturedAt: capturedAt.UTC(), Records: records, } - return s.write("process-peaks.json", payload) + return s.write(payload.Kind, "process-peaks.json", payload) } // WriteProcessCounters writes the current counter records as one superseding @@ -817,7 +834,7 @@ func (s *Spool) WriteProcessCounters(capturedAt time.Time, records []inventory.C CapturedAt: capturedAt.UTC(), Records: records, } - return s.write("process-counters.json", payload) + return s.write(payload.Kind, "process-counters.json", payload) } // WriteListeningPorts writes where every collected workload container accepts @@ -831,7 +848,7 @@ func (s *Spool) WriteListeningPorts(capturedAt time.Time, records []inventory.Po CapturedAt: capturedAt.UTC(), Records: records, } - return s.write("listening-ports.json", payload) + return s.write(payload.Kind, "listening-ports.json", payload) } // WriteGoBuild writes one build's facts. Unlike the superseding go-inventory, @@ -858,7 +875,7 @@ func (s *Spool) WriteGoBuild(b inventory.BuildFacts) error { GoDebug: b.GoDebug, PprofEndpoint: b.HasPprof, } - return s.write("go-build-"+digestFileToken(b.ImageDigest)+".json", payload) + return s.write(payload.Kind, "go-build-"+digestFileToken(b.ImageDigest)+".json", payload) } // digestFileToken turns an image digest into a filename-safe token. Digests are @@ -886,7 +903,7 @@ func (s *Spool) WriteWorkloadMetadata(capturedAt time.Time, records []metadata.R CapturedAt: capturedAt.UTC(), Records: records, } - return s.write("workload-metadata.json", payload) + return s.write(payload.Kind, "workload-metadata.json", payload) } // WriteNodeMetadata writes the current node inventory as one superseding batch. @@ -898,7 +915,7 @@ func (s *Spool) WriteNodeMetadata(capturedAt time.Time, nodes []model.NodeInfo) CapturedAt: capturedAt.UTC(), Nodes: nodes, } - return s.write("node-metadata.json", payload) + return s.write(payload.Kind, "node-metadata.json", payload) } // WriteWorkloadRevisions writes the current revision history as one @@ -912,7 +929,7 @@ func (s *Spool) WriteWorkloadRevisions(capturedAt time.Time, records []revisions CapturedAt: capturedAt.UTC(), Records: records, } - return s.write("workload-revisions.json", payload) + return s.write(payload.Kind, "workload-revisions.json", payload) } // WriteWorkloadPolicy writes the workload-policy snapshot. It supersedes its @@ -925,7 +942,7 @@ func (s *Spool) WriteWorkloadPolicy(capturedAt time.Time, records []model.Worklo UnavailableSources: unavailable, Records: records, } - return s.write("workload-policy.json", payload) + return s.write(payload.Kind, "workload-policy.json", payload) } // WriteClusterPolicy writes the cluster-policy snapshot. @@ -937,7 +954,7 @@ func (s *Spool) WriteClusterPolicy(capturedAt time.Time, policy model.ClusterPol UnavailableSources: unavailable, Policy: policy, } - return s.write("cluster-policy.json", payload) + return s.write(payload.Kind, "cluster-policy.json", payload) } // WriteProfile writes one captured eBPF CPU profile. Each capture is its own @@ -962,7 +979,7 @@ func (s *Spool) WriteProfile(key ProfileKey, pprof []byte) error { name := fmt.Sprintf("profile-%s-%s-%s-%s-%d-%d.json", fileToken(key.Namespace), fileToken(key.Workload), fileToken(key.Container), shortDigest(key.ImageDigest), key.CaptureStart.Unix(), key.CaptureEnd.Unix()) - return s.write(name, payload) + return s.write(payload.Kind, name, payload) } // WritePulledProfile writes one profile fetched from a workload's own @@ -986,7 +1003,7 @@ func (s *Spool) WritePulledProfile(key ProfileKey, pprof []byte, dropped Profile name := fmt.Sprintf("pprof-%s-%s-%s-%s-%d-%d.json", fileToken(key.Namespace), fileToken(key.Workload), fileToken(key.Container), shortDigest(key.ImageDigest), key.CaptureStart.Unix(), key.CaptureEnd.Unix()) - return s.write(name, payload) + return s.write(payload.Kind, name, payload) } // shortDigestLen is how much of a digest the filename carries. A digest is @@ -1070,15 +1087,31 @@ func fileToken(s string) string { return b.String() } -// write marshals the payload and lands it atomically: temp file in the same -// directory, then rename. A crash mid-write leaves a *.tmp file that the +// write records one payload of one kind and refuses a kind the registry does +// not hold. Every caller passes the same string it puts in the payload's bytes, +// so the registry is the one list behind the backend contract and the agent's +// own per-kind counters alike (ADR 0022, ADR 0070 §3). +func (s *Spool) write(kind, name string, payload any) error { + if _, ok := Lookup(kind); !ok { + return fmt.Errorf("refusing to write payload of kind %q: it has no row in the registry", kind) + } + if err := s.writePayload(name, payload); err != nil { + s.countWrite(kind, true) + return err + } + s.countWrite(kind, false) + return nil +} + +// writePayload marshals the payload and lands it atomically: temp file in the +// same directory, then rename. A crash mid-write leaves a *.tmp file that the // sweep collects; readers never observe a partial payload. // // It also refuses a name that is not a plain filename. Callers already build // their names from fileToken, so this catches nothing today — it is here for // the next caller, because every path this package writes goes through this one // function and that makes it the only place worth checking (ADR 0042). -func (s *Spool) write(name string, payload any) error { +func (s *Spool) writePayload(name string, payload any) error { if name != filepath.Base(name) || strings.ContainsAny(name, `/\`) || name == "." || name == ".." { return fmt.Errorf("refusing to write payload under %q: a spool payload name is a plain filename", name) } @@ -1134,6 +1167,11 @@ func (s *Spool) Sweep(now time.Time) error { if err := s.remove(entry.Name()); err != nil { return err } + if expired { + s.countEviction(EvictedAge) + } else { + s.countEviction(EvictedOrphanTemp) + } continue } kept = append(kept, file{name: entry.Name(), size: info.Size(), modTime: info.ModTime()}) @@ -1141,6 +1179,7 @@ func (s *Spool) Sweep(now time.Time) error { } if total <= s.maxBytes && len(kept) <= s.maxFiles { + s.recordSweep(total, len(kept)) return nil } @@ -1157,12 +1196,22 @@ func (s *Spool) Sweep(now time.Time) error { if total <= s.maxBytes && count <= s.maxFiles { break } + // Which ceiling is still exceeded is why this file goes, and the two + // answer different questions: bytes means the cluster produces more + // than the volume holds, files means it produces more payloads than + // the count allows (ADR 0042 §2). + reason := EvictedFiles + if total > s.maxBytes { + reason = EvictedBytes + } if err := s.remove(f.name); err != nil { return err } + s.countEviction(reason) total -= f.size count-- } + s.recordSweep(total, count) return nil } diff --git a/internal/sink/spool_test.go b/internal/sink/spool_test.go index 22adcd6..f8199d5 100644 --- a/internal/sink/spool_test.go +++ b/internal/sink/spool_test.go @@ -1792,7 +1792,7 @@ func TestAnOwnerReferenceNameCannotEscapeTheSpool(t *testing.T) { func TestWriteRefusesANameThatIsNotAPlainFilename(t *testing.T) { s, dir := newTestSpool(t) for _, name := range []string{"../escape.json", "sub/dir.json", `..\escape.json`, ".", ".."} { - if err := s.write(name, struct{}{}); err == nil { + if err := s.write("usage_window", name, struct{}{}); err == nil { t.Errorf("write(%q) was allowed", name) } }