diff --git a/docs/adr/0054-coverage-says-how-much-was-hidden-never-what.md b/docs/adr/0054-coverage-says-how-much-was-hidden-never-what.md index a717d9a..1046f17 100644 --- a/docs/adr/0054-coverage-says-how-much-was-hidden-never-what.md +++ b/docs/adr/0054-coverage-says-how-much-was-hidden-never-what.md @@ -4,7 +4,7 @@ Date: 2026-08-30 Status: Accepted -Amended by: 0057, 0058, 0060 +Amended by: 0057, 0058, 0060, 0087 Amends: 0012 §2, 0033 diff --git a/docs/adr/0067-a-record-carries-the-age-of-its-source.md b/docs/adr/0067-a-record-carries-the-age-of-its-source.md index 31b21f8..d6ffdd0 100644 --- a/docs/adr/0067-a-record-carries-the-age-of-its-source.md +++ b/docs/adr/0067-a-record-carries-the-age-of-its-source.md @@ -4,6 +4,8 @@ Date: 2026-09-04 Status: Accepted +Amended by: 0087 + Amends: 0052, 0056, 0062 The node-fed payloads gain `asserted_at`, `go_inventory.coverage` gains the list diff --git a/docs/adr/0087-a-state-says-when-it-began.md b/docs/adr/0087-a-state-says-when-it-began.md new file mode 100644 index 0000000..caaf817 --- /dev/null +++ b/docs/adr/0087-a-state-says-when-it-began.md @@ -0,0 +1,101 @@ +# 0087. A state says when it began + +Date: 2026-09-18 + +Status: Accepted + +Amends: 0054, 0067 + +`sources[].failing` and `shipping.halted` gain a start instant beside them, and +the metrics endpoint gains one gauge. No new collection, no new privilege, no +new read: both instants are already held in memory and discarded at payload +time. + +## Context + +ADR 0054 made `collection_coverage` the payload the agent answers for itself +with, so that an empty report and a broken agent are not the same bytes. ADR +0067 established that a record carries the instant its source last stated it, +and applied it to the node-fed records — not to the agent's own state. + +Every number in that payload is a counter, cumulative from `since`, and two +fields are not numbers at all: + +``` +sources[].failing bool +shipping.halted bool +``` + +A boolean describes the present, and this kind supersedes — the backend upserts, +last write wins, and one row is all it is obliged to keep. So a state that ended +leaves nothing behind. A watch that failed for nineteen minutes and recovered +produces, on the next flush, bytes identical to a watch that never failed; an +outage shorter than the flush interval is never stated at all. The agent knew +and did not say. + +`halted` is the sharper case, and its own field comment already half-admitted +it: a halted agent ships nothing, so the payload carrying the fact never +travels. Its readers are whoever holds the spool and whoever scrapes the metrics +endpoint, and to both of them "shipping stopped" without "at 04:31 nine days +ago" is nearly the same as silence. + +The third fact is what makes this cheap. Both instants already exist. +`watchHealth` records where each run of failures began — that is what ADR 0035's +streak is measured from — and `SourceHealths` reduced it to a boolean on the way +out. The halt had no clock at all, because nothing had ever asked it for one. + +What is *not* in scope here is history. A state that ended is absent from the +next payload by design, and no field added below changes that. + +## Decision + +**1. Two states carry the instant they began.** `sources[].failing_since` and +`shipping.halted_since`. Absent is the claim that the agent is not in that +state, so a source that recovers drops the field rather than restating it. + +**2. The instant is where the run began, not where it last failed.** What +stopped being true is that the cache was fed, and it stopped when the run +started. The latest failure says only when the agent last looked, which is a +fact about the agent's schedule and not about the source. + +**3. The boolean stays for now, derived from the instant.** §6 of +`backend-requirements.md` requires both forms across a deprecation period, and +the schema gate enforces the same: `buf breaking` runs the `FILE` category, +which refuses the deletion of a field even with its number reserved — measured, +not assumed. Both booleans are therefore computed from the instant beside them +in one place each, so no payload can carry a state with no start or a start with +no state. + +**4. `synced` stays a boolean and is not amended.** It is a latch rather than a +state that ends: it goes true once and never back. The instant it would carry +for the false case is the payload's own `since` — the cache has not filled since +this process began counting — and for the true case it answers a question about +startup that nothing asks. + +**5. One gauge, not two.** `shipping_halted_since_seconds` joins +`shipping_halted`, because for a halt the metrics endpoint is the only channel +that works. `source_failing` gets no companion: its payload arrives on every +pass, and a scrape of the boolean is dated by the scrape. + +## Consequences + +**Easier.** "When did this stop?" becomes answerable for the two states where it +was not, and answerable from the payload rather than from a customer's logs. The +spool file a halted agent leaves behind stops being a fact about the file's own +timestamp and becomes a fact about the halt. + +**Harder / given up.** The payload carries two fields where it carried one, and +will until the booleans are removed — which needs either a released agent to +deprecate against or a decision about the `FILE` gate, and is neither taken nor +pre-judged here. Nothing else grows: both instants were already in memory. + +A state that ended is still invisible, and this ADR does not pretend otherwise. +Recovering that needs occupancy within a window, which is a different mechanism +with a different cost, and is not decided here. + +**Not changed.** What is collected, what is filtered, what is read, any cadence, +any natural key, or the one-way protocol. No identity of any object reaches +either new field: a source is a resource class, and a halt is about this agent. + +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 57dc6be..54fd5fb 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -144,3 +144,4 @@ line to paste. - [0084. What was excluded is named, inside the cluster and only there](0084-what-was-excluded-is-named-inside-the-cluster.md) - [0085. The image is built and scanned where the code is, and pinned to what it was built from](0085-the-image-is-built-and-scanned-by-ci.md) - [0086. A toolchain move is a review of what the toolchain reports](0086-a-toolchain-move-reviews-what-it-reports.md) +- [0087. A state says when it began](0087-a-state-says-when-it-began.md) diff --git a/docs/backend-requirements.md b/docs/backend-requirements.md index a1850b9..3b4d3f7 100644 --- a/docs/backend-requirements.md +++ b/docs/backend-requirements.md @@ -453,7 +453,13 @@ a hash of it; `agent.config.since` is when that configuration took effect, and i is what separates "the data changed" from "the filters changed". `sources[]` reports what the agent's reads actually did, so a source that is `synced: false` or `failing: true` bounds every finding resting on it — the backend MUST NOT -present such a finding as complete. The scan block is a fleet aggregate over the +present such a finding as complete. `failing_since` is when the run of failures +that left a cache unfed began, and `failing` is the same fact as a boolean, +derived from it, deprecated and retained for the window of section 6. An absent +instant means **not failing now**, never "never failed": a source that failed +and recovered between two flushes states nothing at all, so the backend MUST NOT +read a clean payload as evidence that these reads were healthy throughout the +period since `since` ([ADR 0087](adr/0087-a-state-says-when-it-began.md)). The scan block is a fleet aggregate over the latest pass of each reporting node and MUST NOT be summed across flushes. The `filter` counters count **decisions, not pods**. A customer who annotates a diff --git a/docs/prose_test.go b/docs/prose_test.go index 02a328c..2b0a49b 100644 --- a/docs/prose_test.go +++ b/docs/prose_test.go @@ -35,23 +35,23 @@ var ceilings = []struct { section int }{ // Each raise names the obligation that bought it: payloads leave the cluster - // at all (0075); a journal window states when it was written (0078); an - // endpoint confirmed once per image is now read once a minute per process - // (0080); twelve more GODEBUG names come out of the build (0081); every - // request names the agent that made it (0083); a listener answers inside the - // cluster what the filters excluded, by name (0084); and the image a customer - // trusts with a node states what it is made of, and what does not yet back - // it (0085). - {"security.md", 1135, 366}, + // at all (0075); a journal window states when it was written (0078); a + // confirmed endpoint is read once a minute per process (0080); twelve more + // GODEBUG names come out of the build (0081); a request names the agent that + // made it (0083); a listener names inside the cluster what the filters + // excluded (0084); the image states what it is made of and what does not yet + // back it (0085); a failing read says since when (0087), whose diff also + // corrects the startup-read list from five kinds to the six it reads. + {"security.md", 1138, 366}, // Each raise names its obligation here too. The body is encoded, so a - // payload's size is no longer the request's (ADR 0076); a window states - // whether it is finished (ADR 0078); gauge samples arrive, where adding - // them up is the wrong ingest (ADR 0080); `godebug` carries names where - // neither value is safe (ADR 0081); the filter counters count decisions - // (ADR 0054); a port's `pprof` is absent when nothing asked, not `absent` - // (ADR 0082); and a request may name an agent that states no version - // (ADR 0083). - {"backend-requirements.md", 823, 644}, + // payload's size is no longer the request's (0076); a window states whether + // it is finished (0078); gauge samples arrive, where adding them up is the + // wrong ingest (0080); `godebug` carries names where neither value is safe + // (0081); the filter counters count decisions (0054); a port's `pprof` is + // absent when nothing asked, not `absent` (0082); a request may name an + // agent that states no version (0083); and an absent instant is told apart + // from an absent state (0087) — the raise that also moved the section. + {"backend-requirements.md", 829, 651}, } func TestNoCommentRunIsLongerThanAPointerToAnADR(t *testing.T) { diff --git a/docs/security.md b/docs/security.md index 9e8a9c5..8b823d0 100644 --- a/docs/security.md +++ b/docs/security.md @@ -272,8 +272,9 @@ and 20000 files, dropping the oldest first; and the `emptyDir` declares a 1 GiB - **The agent reads it once, at startup**, so a restart resumes its open windows rather than starting them over ([ADR 0072](adr/0072-a-restart-resumes-the-window-the-spool-holds.md), [ADR 0077](adr/0077-a-restart-resumes-every-open-window.md)). It is a pure read - of five kinds and no others — `usage_snapshot`, `container_restarts`, - `pod_disruptions`, `node_lifecycle`, `job_runs` — and it deletes nothing. + of six kinds and no others — `usage_snapshot`, `container_restarts`, + `pod_disruptions`, `node_lifecycle`, `job_runs`, `goroutine_counts` — and it + deletes nothing. - **Configuration is never cached on it.** Filters are read from the ConfigMap at every start and never reread while running; a stale filter set cannot resurrect from disk, and a filter change takes a restart. @@ -1092,7 +1093,9 @@ 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` +failing — and says since when, so an outage you have already fixed is not +reported as one you still have (ADR 0087). 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. **And named, to you, inside your cluster.** The counts are what *we* receive. You can read the same decisions resolved to objects at `/collection` on the controller's loopback address ([§5](#5-network-access)): one line per namespace, pod and Job, whether it is collected, whether it may be profiled, which of your five controls refused it, and whether its workload was one the agent cannot read. It is computed on request from the same caches and the same filter code the agent collects with, so it cannot disagree with what it does. Nothing of it is transmitted, and nothing of it is remembered. diff --git a/internal/collector/pods.go b/internal/collector/pods.go index 04aae06..44a394d 100644 --- a/internal/collector/pods.go +++ b/internal/collector/pods.go @@ -309,10 +309,12 @@ func (w *PodWatcher) SourceHealths() []model.SourceHealth { } out := make([]model.SourceHealth, 0, len(w.policySources)) for _, source := range w.policySources { + since := source.health.failingSince(now, window) out = append(out, model.SourceHealth{ - Name: source.name, - Synced: source.synced != nil && source.synced(), - Failing: source.health != nil && source.health.failedWithin(now, window), + Name: source.name, + Synced: source.synced != nil && source.synced(), + Failing: since != nil, + FailingSince: since, }) } sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name }) diff --git a/internal/collector/watchhealth.go b/internal/collector/watchhealth.go index 564b37b..9ed8b99 100644 --- a/internal/collector/watchhealth.go +++ b/internal/collector/watchhealth.go @@ -129,14 +129,30 @@ func (h *watchHealth) failingFor(now time.Time) (time.Duration, string) { return h.last.Sub(h.first), h.lastErr } -// failedWithin reports whether this cache failed at any point in the last d. -func (h *watchHealth) failedWithin(now time.Time, d time.Duration) bool { +// failingSince is when the run of failures that leaves this cache unfed began, +// and nil when it is not failing by that test. +// +// The instant is the run's first failure rather than its latest: what stopped +// being true is that the cache was fed, and that stopped when the run started. +// The latest failure is only when the agent last looked (ADR 0087). +func (h *watchHealth) failingSince(now time.Time, d time.Duration) *time.Time { if h == nil { - return false + return nil } h.mu.Lock() defer h.mu.Unlock() - return !h.last.IsZero() && now.Sub(h.last) <= d + if h.last.IsZero() || now.Sub(h.last) > d { + return nil + } + first := h.first + return &first +} + +// failedWithin reports whether this cache failed at any point in the last d. +// Answered from failingSince so that the boolean and the instant beside it in +// the payload cannot disagree. +func (h *watchHealth) failedWithin(now time.Time, d time.Duration) bool { + return h.failingSince(now, d) != nil } // watchdog blocks until ctx is canceled, or until one of the gating caches has diff --git a/internal/collector/watchhealth_test.go b/internal/collector/watchhealth_test.go index 76a5098..a1740e4 100644 --- a/internal/collector/watchhealth_test.go +++ b/internal/collector/watchhealth_test.go @@ -352,3 +352,50 @@ func waitFor(t *testing.T, within time.Duration, what string, cond func() bool) } t.Fatalf("timed out waiting for %s", what) } + +// The instant a source carries is where its run of failures began, not where it +// last failed. A reader asking when the cache stopped being fed is asking about +// the start; the latest failure only says when the agent last looked (ADR 0087). +func TestAFailingSourceIsDatedFromTheStartOfTheRun(t *testing.T) { + t0 := time.Date(2026, 9, 18, 12, 0, 0, 0, time.UTC) + h := newWatchHealth(2 * time.Minute) + h.record(t0, errors.New("forbidden")) + h.record(t0.Add(time.Minute), errors.New("forbidden")) + h.record(t0.Add(2*time.Minute), errors.New("forbidden")) + + since := h.failingSince(t0.Add(3*time.Minute), 10*time.Minute) + if since == nil { + t.Fatal("a cache failing three times running is not reported as failing") + } + if !since.Equal(t0) { + t.Errorf("failing since %s, want the first failure at %s", since, t0) + } +} + +// A run that ended leaves nothing behind: the field is absent, and the claim +// the payload makes is that the cache is fed again. What the outage was is the +// window's to answer, not this field's. +func TestASourceThatRecoveredCarriesNoInstant(t *testing.T) { + t0 := time.Date(2026, 9, 18, 12, 0, 0, 0, time.UTC) + h := newWatchHealth(2 * time.Minute) + h.record(t0, errors.New("connection reset")) + + if since := h.failingSince(t0.Add(time.Hour), 10*time.Minute); since != nil { + t.Errorf("a cache fed again for an hour is dated %s", since) + } +} + +// The boolean and the instant are one answer read twice, so no payload can +// carry a source that is failing with no start, or a start with no failure. +func TestTheBooleanAndTheInstantAgree(t *testing.T) { + t0 := time.Date(2026, 9, 18, 12, 0, 0, 0, time.UTC) + h := newWatchHealth(2 * time.Minute) + h.record(t0, errors.New("forbidden")) + + for _, at := range []time.Time{t0, t0.Add(time.Minute), t0.Add(time.Hour)} { + since, failing := h.failingSince(at, 10*time.Minute), h.failedWithin(at, 10*time.Minute) + if (since != nil) != failing { + t.Errorf("at %s: failing=%v, instant=%v", at, failing, since) + } + } +} diff --git a/internal/metrics/collect.go b/internal/metrics/collect.go index df21a69..6dadc96 100644 --- a/internal/metrics/collect.go +++ b/internal/metrics/collect.go @@ -243,6 +243,14 @@ func CollectController(c Controller) *Set { s.Gauge("shipping_halted", "Whether shipping has stopped on an identity failure the agent will not retry into.", b2f(sh.Halted)) + // And when, which is the half that matters here: nothing clears a halt + // but a restart, so the question a scrape answers is how long this + // agent has been silent (ADR 0087). + if sh.HaltedSince != nil { + s.Gauge("shipping_halted_since_seconds", + "When shipping stopped, in seconds since the epoch. Absent while shipping runs.", + float64(sh.HaltedSince.Unix())) + } } return s } diff --git a/internal/metrics/collect_test.go b/internal/metrics/collect_test.go index a8929e0..4cc7aaf 100644 --- a/internal/metrics/collect_test.go +++ b/internal/metrics/collect_test.go @@ -229,6 +229,7 @@ func TestAnUnconfiguredControllerOmitsWhatItDoesNotHave(t *testing.T) { // An agent with no backend has not failed to ship: it was told to ship // nowhere, and a delivered count of zero would claim otherwise. "shipments_total", "shipments_rejected_total", "shipping_halted", + "shipping_halted_since_seconds", } { if present[Prefix+absent] { t.Errorf("%s%s is exposed by an installation that has no such component", Prefix, absent) diff --git a/internal/model/coverage.go b/internal/model/coverage.go index ed94f7f..52a68eb 100644 --- a/internal/model/coverage.go +++ b/internal/model/coverage.go @@ -1,5 +1,7 @@ package model +import "time" + // Coverage is an aggregate snapshot of what the filter admitted and // excluded, counted once per pod appearance. It is the seed of the coverage // report (docs/security.md §11): full information about what is collected, @@ -46,11 +48,18 @@ type SourceHealth struct { // Name is the resource class, not a customer object: "services", // "endpoint_slices", and so on. Name string `json:"name"` - // Synced is whether the cache ever filled. + // Synced is whether the cache ever filled. It is a latch rather than a + // state that ends, and the instant it would carry is the payload's own + // `since` — so it stays a boolean where the two below do not (ADR 0087). Synced bool `json:"synced"` // Failing is whether its watch has been erroring recently enough to treat - // the cache as no longer fed (ADR 0035). + // the cache as no longer fed (ADR 0035). Derived from FailingSince so the + // two cannot disagree, and kept only for the compatibility window. Failing bool `json:"failing,omitempty"` + // FailingSince is when the run of failures that left the cache unfed began. + // Nil is the claim that it is still being fed: a source that recovers drops + // the field, and what it was doing before that is the window's to say. + FailingSince *time.Time `json:"failing_since,omitempty"` } // NodeDrops is what the node reductions refused to carry, cumulative since the @@ -129,11 +138,15 @@ type Shipping struct { Unreadable uint64 `json:"unreadable"` // Halted is whether shipping has stopped on an identity failure. Nothing // retries into a rejected credential, so the agent stops rather than turning - // a fleet's spools into a retry storm against it. + // a fleet's spools into a retry storm against it. Derived from HaltedSince, + // and kept only for the compatibility window. // // A halted agent ships nothing, this payload included, so its readers are // whoever holds the spool or scrapes the metrics endpoint. Halted bool `json:"halted"` + // HaltedSince is when that halt began. A halt has no end — it is cleared by + // a restart, which moves `since` and starts the counters over. + HaltedSince *time.Time `json:"halted_since,omitempty"` // Rejected is what the backend refused permanently, by reason. Rejected ShippingRejections `json:"rejected"` // PayloadBytes and TransmittedBytes are the same deliveries measured on diff --git a/internal/pb/kubernetesv1/collection_coverage.pb.go b/internal/pb/kubernetesv1/collection_coverage.pb.go index bcfea6c..7240bfd 100644 --- a/internal/pb/kubernetesv1/collection_coverage.pb.go +++ b/internal/pb/kubernetesv1/collection_coverage.pb.go @@ -458,9 +458,15 @@ type SourceHealth struct { // not stated; false is a claim that the cache never filled. Synced *bool `protobuf:"varint,2,opt,name=synced,proto3,oneof" json:"synced,omitempty"` // Whether the watch has been failing recently enough to treat the cache as no - // longer fed. Absent means it was not stated; false is a claim that the cache - // is still being fed. - Failing *bool `protobuf:"varint,3,opt,name=failing,proto3,oneof" json:"failing,omitempty"` + // longer fed. Superseded by `failing_since`, which states the same fact and + // says when it began; kept for the compatibility window of section 6. + // + // Deprecated: Marked as deprecated in rebuildstack/ingest/kubernetes/v1/collection_coverage.proto. + Failing *bool `protobuf:"varint,3,opt,name=failing,proto3,oneof" json:"failing,omitempty"` + // When the run of failures that left this cache unfed began. Absent is the + // claim that the cache is still being fed, so a source that recovers drops + // the field rather than restating it. + FailingSince *timestamppb.Timestamp `protobuf:"bytes,4,opt,name=failing_since,json=failingSince,proto3" json:"failing_since,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -509,6 +515,7 @@ func (x *SourceHealth) GetSynced() bool { return false } +// Deprecated: Marked as deprecated in rebuildstack/ingest/kubernetes/v1/collection_coverage.proto. func (x *SourceHealth) GetFailing() bool { if x != nil && x.Failing != nil { return *x.Failing @@ -516,6 +523,13 @@ func (x *SourceHealth) GetFailing() bool { return false } +func (x *SourceHealth) GetFailingSince() *timestamppb.Timestamp { + if x != nil { + return x.FailingSince + } + return nil +} + // How many pods and Jobs the filter looked at and excluded, counted once per // appearance, and under which control. // @@ -1404,9 +1418,10 @@ type ShippingCoverage struct { // Spool files the agent could not read, or whose kind is not one it ships. // They were left where they were. Unreadable *uint64 `protobuf:"varint,3,opt,name=unreadable,proto3,oneof" json:"unreadable,omitempty"` - // Whether shipping has stopped on an identity failure. A halted agent sends - // nothing at all, so this field arriving true describes a halt that came - // after this payload was written, never the one in force now. + // Whether shipping has stopped on an identity failure. Superseded by + // `halted_since`; kept for the compatibility window of section 6. + // + // Deprecated: Marked as deprecated in rebuildstack/ingest/kubernetes/v1/collection_coverage.proto. Halted *bool `protobuf:"varint,4,opt,name=halted,proto3,oneof" json:"halted,omitempty"` // What the backend refused permanently, by reason. Rejected *ShippingRejections `protobuf:"bytes,5,opt,name=rejected,proto3" json:"rejected,omitempty"` @@ -1415,8 +1430,12 @@ type ShippingCoverage struct { // every attempt, so a retried payload is counted each time it cost a request. PayloadBytes *uint64 `protobuf:"varint,6,opt,name=payload_bytes,json=payloadBytes,proto3,oneof" json:"payload_bytes,omitempty"` TransmittedBytes *uint64 `protobuf:"varint,7,opt,name=transmitted_bytes,json=transmittedBytes,proto3,oneof" json:"transmitted_bytes,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // When shipping stopped on an identity failure. A halted agent sends nothing + // at all, so this arriving describes a halt that came after this payload was + // written; the halt in force now is on the metrics endpoint alone. + HaltedSince *timestamppb.Timestamp `protobuf:"bytes,8,opt,name=halted_since,json=haltedSince,proto3" json:"halted_since,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *ShippingCoverage) Reset() { @@ -1470,6 +1489,7 @@ func (x *ShippingCoverage) GetUnreadable() uint64 { return 0 } +// Deprecated: Marked as deprecated in rebuildstack/ingest/kubernetes/v1/collection_coverage.proto. func (x *ShippingCoverage) GetHalted() bool { if x != nil && x.Halted != nil { return *x.Halted @@ -1498,6 +1518,13 @@ func (x *ShippingCoverage) GetTransmittedBytes() uint64 { return 0 } +func (x *ShippingCoverage) GetHaltedSince() *timestamppb.Timestamp { + if x != nil { + return x.HaltedSince + } + return nil +} + // What the backend refused to accept from the agent, by reason. A payload // counted here is not offered again by the process that counted it. // @@ -1768,11 +1795,12 @@ const file_rebuildstack_ingest_kubernetes_v1_collection_coverage_proto_rawDesc = "\x13_pprof_pull_enabledB\x1a\n" + "\x18_symbol_prefixes_allowedB\x16\n" + "\x14_node_intake_enabledB\x15\n" + - "\x13_backend_configured\"u\n" + + "\x13_backend_configured\"\xba\x01\n" + "\fSourceHealth\x12\x12\n" + "\x04name\x18\x01 \x01(\tR\x04name\x12\x1b\n" + - "\x06synced\x18\x02 \x01(\bH\x00R\x06synced\x88\x01\x01\x12\x1d\n" + - "\afailing\x18\x03 \x01(\bH\x01R\afailing\x88\x01\x01B\t\n" + + "\x06synced\x18\x02 \x01(\bH\x00R\x06synced\x88\x01\x01\x12!\n" + + "\afailing\x18\x03 \x01(\bB\x02\x18\x01H\x01R\afailing\x88\x01\x01\x12?\n" + + "\rfailing_since\x18\x04 \x01(\v2\x1a.google.protobuf.TimestampR\ffailingSinceB\t\n" + "\a_syncedB\n" + "\n" + "\b_failing\"\xd6\t\n" + @@ -1911,17 +1939,18 @@ const file_rebuildstack_ingest_kubernetes_v1_collection_coverage_proto_rawDesc = "\n" + "_too_largeB\f\n" + "\n" + - "_malformed\"\xa4\x03\n" + + "_malformed\"\xe7\x03\n" + "\x10ShippingCoverage\x12!\n" + "\tdelivered\x18\x01 \x01(\x04H\x00R\tdelivered\x88\x01\x01\x12\x1f\n" + "\bdeferred\x18\x02 \x01(\x04H\x01R\bdeferred\x88\x01\x01\x12#\n" + "\n" + "unreadable\x18\x03 \x01(\x04H\x02R\n" + - "unreadable\x88\x01\x01\x12\x1b\n" + - "\x06halted\x18\x04 \x01(\bH\x03R\x06halted\x88\x01\x01\x12Q\n" + + "unreadable\x88\x01\x01\x12\x1f\n" + + "\x06halted\x18\x04 \x01(\bB\x02\x18\x01H\x03R\x06halted\x88\x01\x01\x12Q\n" + "\brejected\x18\x05 \x01(\v25.rebuildstack.ingest.kubernetes.v1.ShippingRejectionsR\brejected\x12(\n" + "\rpayload_bytes\x18\x06 \x01(\x04H\x04R\fpayloadBytes\x88\x01\x01\x120\n" + - "\x11transmitted_bytes\x18\a \x01(\x04H\x05R\x10transmittedBytes\x88\x01\x01B\f\n" + + "\x11transmitted_bytes\x18\a \x01(\x04H\x05R\x10transmittedBytes\x88\x01\x01\x12=\n" + + "\fhalted_since\x18\b \x01(\v2\x1a.google.protobuf.TimestampR\vhaltedSinceB\f\n" + "\n" + "_deliveredB\v\n" + "\t_deferredB\r\n" + @@ -2012,14 +2041,16 @@ var file_rebuildstack_ingest_kubernetes_v1_collection_coverage_proto_depIdxs = [ 15, // 14: rebuildstack.ingest.kubernetes.v1.CollectionCoverage.goroutine_counts:type_name -> rebuildstack.ingest.kubernetes.v1.GoroutineCountCoverage 2, // 15: rebuildstack.ingest.kubernetes.v1.AgentInfo.config:type_name -> rebuildstack.ingest.kubernetes.v1.ConfigShape 17, // 16: rebuildstack.ingest.kubernetes.v1.ConfigShape.since:type_name -> google.protobuf.Timestamp - 17, // 17: rebuildstack.ingest.kubernetes.v1.ScanCoverage.oldest_asserted_at:type_name -> google.protobuf.Timestamp - 16, // 18: rebuildstack.ingest.kubernetes.v1.ProfileCoverage.states:type_name -> rebuildstack.ingest.kubernetes.v1.ProfileCoverage.StatesEntry - 13, // 19: rebuildstack.ingest.kubernetes.v1.ShippingCoverage.rejected:type_name -> rebuildstack.ingest.kubernetes.v1.ShippingRejections - 20, // [20:20] is the sub-list for method output_type - 20, // [20:20] is the sub-list for method input_type - 20, // [20:20] is the sub-list for extension type_name - 20, // [20:20] is the sub-list for extension extendee - 0, // [0:20] is the sub-list for field type_name + 17, // 17: rebuildstack.ingest.kubernetes.v1.SourceHealth.failing_since:type_name -> google.protobuf.Timestamp + 17, // 18: rebuildstack.ingest.kubernetes.v1.ScanCoverage.oldest_asserted_at:type_name -> google.protobuf.Timestamp + 16, // 19: rebuildstack.ingest.kubernetes.v1.ProfileCoverage.states:type_name -> rebuildstack.ingest.kubernetes.v1.ProfileCoverage.StatesEntry + 13, // 20: rebuildstack.ingest.kubernetes.v1.ShippingCoverage.rejected:type_name -> rebuildstack.ingest.kubernetes.v1.ShippingRejections + 17, // 21: rebuildstack.ingest.kubernetes.v1.ShippingCoverage.halted_since:type_name -> google.protobuf.Timestamp + 22, // [22:22] is the sub-list for method output_type + 22, // [22:22] is the sub-list for method input_type + 22, // [22:22] is the sub-list for extension type_name + 22, // [22:22] is the sub-list for extension extendee + 0, // [0:22] is the sub-list for field type_name } func init() { file_rebuildstack_ingest_kubernetes_v1_collection_coverage_proto_init() } diff --git a/internal/shipper/shipper.go b/internal/shipper/shipper.go index 722740c..d1e2c27 100644 --- a/internal/shipper/shipper.go +++ b/internal/shipper/shipper.go @@ -77,9 +77,13 @@ type Shipper struct { // because it cannot change while the process runs (ADR 0083). agent string - // halted latches on an identity failure and never clears: nothing retries - // into a rejected credential. - halted atomic.Bool + // haltedAt latches on an identity failure and never clears: nothing retries + // into a rejected credential. Zero is not halted; any other value is the + // instant it stopped, which is the whole of what a reader can act on when + // the agent has been silent for days (ADR 0087). + haltedAt atomic.Int64 + // now is the clock, replaced in tests. + now func() time.Time // mu guards everything below. The round runs on one goroutine; the lock is // for the coverage flush and the metrics handler reading across it. @@ -118,11 +122,15 @@ func New(base, dir, version string, logger *slog.Logger) *Shipper { // #nosec G404 -- jitter spreads a fleet's retries over a window; it is // not a secret and predicting it buys nothing jitter: rand.Int64N, + now: time.Now, interval: DefaultInterval, finalBudget: FinalRoundBudget, } } +// halted reports whether shipping has stopped on an identity failure. +func (s *Shipper) halted() bool { return s.haltedAt.Load() != 0 } + // Run ships until ctx is canceled. // // It returns only on cancellation, halted or not: in cmd/agent a lifecycle task @@ -140,7 +148,7 @@ func (s *Shipper) Run(ctx context.Context) error { case <-timer.C: } s.Round(ctx) - if s.halted.Load() { + if s.halted() { <-ctx.Done() return nil } @@ -156,7 +164,7 @@ func (s *Shipper) Run(ctx context.Context) error { // earlier of the two wins. Undelivered payloads stay in the spool as always; // what differs is that no next process may read them. func (s *Shipper) FinalRound(ctx context.Context) { - if s.halted.Load() { + if s.halted() { return // nothing retries into a rejected credential, this least of all } ctx, cancel := context.WithTimeout(ctx, s.finalBudget) @@ -183,7 +191,7 @@ func (s *Shipper) FinalRound(ctx context.Context) { // at the first payload the backend could not take, because the next one would // meet the same backend. func (s *Shipper) Round(ctx context.Context) { - if s.halted.Load() { + if s.halted() { return // nothing retries into a rejected credential } for _, f := range s.pending() { @@ -203,7 +211,10 @@ func (s *Shipper) Coverage() model.Shipping { s.mu.Lock() defer s.mu.Unlock() c := s.counts - c.Halted = s.halted.Load() + if at := s.haltedAt.Load(); at != 0 { + halted := time.Unix(0, at).UTC() + c.Halted, c.HaltedSince = true, &halted + } return c } @@ -304,7 +315,7 @@ func (s *Shipper) deliver(ctx context.Context, f spooled) bool { s.logger.Error("the backend refused this agent's identity; shipping stops until this agent is restarted", "file", f.name, "kind", kind, "status", status) s.count(func(c *model.Shipping) { c.Rejected.Unauthorized++ }) - s.halted.Store(true) + s.haltedAt.Store(s.now().UnixNano()) return false // Permanent for these bytes: held, never offered again, and left in the diff --git a/internal/shipper/shipper_test.go b/internal/shipper/shipper_test.go index a95e69e..6e4a0a7 100644 --- a/internal/shipper/shipper_test.go +++ b/internal/shipper/shipper_test.go @@ -21,6 +21,10 @@ import ( "github.com/RebuildStackCo/runtime-agent/internal/sink" ) +// haltClock is what the shipper's clock reads in these tests, so the instant a +// halt carries is a fixed number rather than whenever the suite ran. +var haltClock = time.Date(2026, 9, 18, 10, 12, 0, 0, time.UTC) + // The rule this package exists for: a spool file is deleted only after a 2xx. // Every other answer leaves the payload where it is, which is the difference // between an outage and data loss nobody detects (ADR 0075). @@ -64,6 +68,17 @@ func TestOnlyA2xxRemovesAPayload(t *testing.T) { t.Errorf("status %d: an attempt that reached the backend counted no bytes: %+v", c.status, got) } got.PayloadBytes, got.TransmittedBytes = 0, 0 + // The halt is the one answer that carries an instant, and it is + // asserted rather than zeroed away: a halt nobody can date is the + // defect this field exists to close (ADR 0087). + if c.want.Halted { + if got.HaltedSince == nil || !got.HaltedSince.Equal(haltClock) { + t.Errorf("status %d: halted since %v, want %v", c.status, got.HaltedSince, haltClock) + } + } else if got.HaltedSince != nil { + t.Errorf("status %d: not halted, yet dated %v", c.status, got.HaltedSince) + } + got.HaltedSince = nil if got != c.want { t.Errorf("status %d: coverage %+v, want %+v", c.status, got, c.want) } @@ -690,7 +705,7 @@ func TestAHaltedShipperRunsNoFinalRound(t *testing.T) { payload(t, dir, "collection-coverage.json", "collection_coverage") s.Round(t.Context()) // the 401 that halts it - if !s.halted.Load() { + if !s.halted() { t.Fatal("a 401 did not halt the shipper") } before := requests.Load() @@ -709,6 +724,7 @@ func newShipper(t *testing.T, base string) (*Shipper, string) { if s == nil { t.Fatalf("New returned nil for base %q", base) } + s.now = func() time.Time { return haltClock } return s, dir } diff --git a/internal/sink/spool_test.go b/internal/sink/spool_test.go index 3d77cc9..dc89490 100644 --- a/internal/sink/spool_test.go +++ b/internal/sink/spool_test.go @@ -134,8 +134,12 @@ func TestGoldenCollectionCoveragePayload(t *testing.T) { }, UsageSignals: []string{"cpu", "memory", "network", "throttling"}, } + // The failing cache is dated: a reader who finds no EndpointSlice-derived + // finding needs to know the cache stopped being fed at 09:47, not merely + // that it is unfed now (ADR 0087). + failingSince := capturedAt.Add(-25 * time.Minute) sources := []model.SourceHealth{ - {Name: "endpoint_slices", Synced: false, Failing: true}, + {Name: "endpoint_slices", Synced: false, Failing: true, FailingSince: &failingSince}, {Name: "services", Synced: true}, } filter := model.Coverage{ @@ -2102,3 +2106,43 @@ func TestCoverageNamesNothingItExcluded(t *testing.T) { t.Errorf("excluded count = %d, want 3", payload.Filter.ExcludedNamespaceFilter) } } + +// A halt is the one state whose payload cannot arrive: the agent stops shipping +// the moment it latches, so this file is written and never sent. Dating it is +// what makes the spool worth opening — "stopped" is a fact the file's own +// timestamp already implies, and "stopped at 04:31 on the 12th" is not +// (ADR 0087). +func TestAHaltedShipperDatesTheHaltInThePayload(t *testing.T) { + s, dir := newTestSpool(t) + haltedAt := capturedAt.Add(-9 * time.Hour) + shipping := model.Shipping{ + Delivered: 1204, Halted: true, HaltedSince: &haltedAt, + Rejected: model.ShippingRejections{Unauthorized: 1}, + } + if err := s.WriteCollectionCoverage(capturedAt, capturedAt.Add(-6*time.Hour), + AgentInfo{Version: "1.4.2"}, nil, model.Coverage{}, model.PlacementDrops{}, + model.NodeDrops{}, nil, nil, nil, nil, nil, nil, nil, &shipping); err != nil { + t.Fatal(err) + } + + var payload struct { + Shipping struct { + Halted bool `json:"halted"` + HaltedSince *time.Time `json:"halted_since"` + } `json:"shipping"` + } + // #nosec G304 -- dir is this test's own TempDir + body, err := os.ReadFile(filepath.Join(dir, "collection-coverage.json")) + if err != nil { + t.Fatal(err) + } + if err := json.Unmarshal(body, &payload); err != nil { + t.Fatalf("decoding the payload: %v", err) + } + if !payload.Shipping.Halted { + t.Error("a halted shipper wrote a payload that does not say so") + } + if payload.Shipping.HaltedSince == nil || !payload.Shipping.HaltedSince.Equal(haltedAt) { + t.Errorf("halted_since is %v, want %s", payload.Shipping.HaltedSince, haltedAt) + } +} diff --git a/internal/sink/testdata/collection-coverage.golden.json b/internal/sink/testdata/collection-coverage.golden.json index 8b3e6e7..f96fcb8 100644 --- a/internal/sink/testdata/collection-coverage.golden.json +++ b/internal/sink/testdata/collection-coverage.golden.json @@ -28,7 +28,8 @@ { "name": "endpoint_slices", "synced": false, - "failing": true + "failing": true, + "failing_since": "2026-08-06T09:47:00Z" }, { "name": "services", diff --git a/proto/rebuildstack/ingest/kubernetes/v1/collection_coverage.proto b/proto/rebuildstack/ingest/kubernetes/v1/collection_coverage.proto index a6b7bbc..1c81a54 100644 --- a/proto/rebuildstack/ingest/kubernetes/v1/collection_coverage.proto +++ b/proto/rebuildstack/ingest/kubernetes/v1/collection_coverage.proto @@ -151,9 +151,14 @@ message SourceHealth { optional bool synced = 2; // Whether the watch has been failing recently enough to treat the cache as no - // longer fed. Absent means it was not stated; false is a claim that the cache - // is still being fed. - optional bool failing = 3; + // longer fed. Superseded by `failing_since`, which states the same fact and + // says when it began; kept for the compatibility window of section 6. + optional bool failing = 3 [deprecated = true]; + + // When the run of failures that left this cache unfed began. Absent is the + // claim that the cache is still being fed, so a source that recovers drops + // the field rather than restating it. + google.protobuf.Timestamp failing_since = 4; } // How many pods and Jobs the filter looked at and excluded, counted once per @@ -414,10 +419,9 @@ message ShippingCoverage { // They were left where they were. optional uint64 unreadable = 3; - // Whether shipping has stopped on an identity failure. A halted agent sends - // nothing at all, so this field arriving true describes a halt that came - // after this payload was written, never the one in force now. - optional bool halted = 4; + // Whether shipping has stopped on an identity failure. Superseded by + // `halted_since`; kept for the compatibility window of section 6. + optional bool halted = 4 [deprecated = true]; // What the backend refused permanently, by reason. ShippingRejections rejected = 5; @@ -427,6 +431,11 @@ message ShippingCoverage { // every attempt, so a retried payload is counted each time it cost a request. optional uint64 payload_bytes = 6; optional uint64 transmitted_bytes = 7; + + // When shipping stopped on an identity failure. A halted agent sends nothing + // at all, so this arriving describes a halt that came after this payload was + // written; the halt in force now is on the metrics endpoint alone. + google.protobuf.Timestamp halted_since = 8; } // What the backend refused to accept from the agent, by reason. A payload