From 5c4cb367d07465cafb1e1580b96ac57224985094 Mon Sep 17 00:00:00 2001 From: Konstantins Feofantovs Date: Fri, 18 Sep 2026 17:33:04 +0300 Subject: [PATCH] feat: a window says how long a state held (ADR 0088) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ADR 0087 gave the two agent states a start instant and named what it did not do: a state that ended is absent from the next payload by design, because a state field describes the present and the kind supersedes. A watch that failed for nineteen minutes and recovered left the backend's row clean. `agent_states` is the other half — one record per (window, state, subject), hour-aligned, carrying how much of the hour the state held, how many episodes that was, and how much of the hour the agent was running to watch at all. The relationship to `collection_coverage` is `container_restarts` to `restart_counters`: one is a reading, the other a window, and the contract carries the sentence that keeps them apart. Occupancy is an advance rather than a reading, so resuming it is where this could go silently wrong. Accrual moves forward from `accrued_to` and is never recomputed from the episode's start; `open_since` is what tells a resumed process the episode it finds open is the one already counted. Both directions are covered by tests that fail on a number, not on an error. Sampling at the pass loses nothing, and that is measured rather than assumed: a pass runs every minute and a new run of watch failures needs two minutes of quiet to begin, so two episodes cannot both begin and end between two passes. The counter deltas are deliberately not here — subtracting a previous reading is a different mechanism with its own failure modes (ADR 0062 priced them). go test -race ./... all packages ok go vet -tags e2e ./... clean GOOS=linux golangci-lint run 0 issues Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01KeZECJXTDHL6Nts95eKMJA --- cmd/agent/main.go | 24 +- cmd/agent/recover.go | 6 +- cmd/agent/states.go | 34 ++ ...age-says-how-much-was-hidden-never-what.md | 2 +- docs/adr/0087-a-state-says-when-it-began.md | 2 + ...088-a-window-says-how-long-a-state-held.md | 131 ++++++++ docs/adr/README.md | 1 + docs/backend-requirements.md | 22 ++ docs/prose_test.go | 34 +- docs/security.md | 7 +- internal/journal/states.go | 306 ++++++++++++++++++ internal/journal/states_test.go | 222 +++++++++++++ internal/sink/recover.go | 12 +- internal/sink/recover_test.go | 59 ++++ internal/sink/registry.go | 12 + internal/sink/spool.go | 46 +++ internal/sink/spool_test.go | 46 +++ .../sink/testdata/agent-states.golden.json | 40 +++ 18 files changed, 978 insertions(+), 28 deletions(-) create mode 100644 cmd/agent/states.go create mode 100644 docs/adr/0088-a-window-says-how-long-a-state-held.md create mode 100644 internal/journal/states.go create mode 100644 internal/journal/states_test.go create mode 100644 internal/sink/testdata/agent-states.golden.json diff --git a/cmd/agent/main.go b/cmd/agent/main.go index f967032..00853bb 100644 --- a/cmd/agent/main.go +++ b/cmd/agent/main.go @@ -298,6 +298,11 @@ func run(ctx context.Context, logger *slog.Logger, clientset kubernetes.Interfac // supply for a workload that ran for part of a window (ADR 0029). jobJournal := journal.NewJobRuns(collector.UsageWindowLength) nodeJournal := journal.NewNodeEvents(collector.UsageWindowLength) + // The agent's own states over time. Its subject is this agent rather than + // the cluster, and it is a window for the reason the others are: a state + // that ended leaves no field behind in a payload that supersedes, so the + // only place it can be reported is a window (ADR 0088). + stateJournal := journal.NewStates(collector.UsageWindowLength) podWatcher.OnJobFinished(func(r model.JobRun) { logger.Info("job run finished", "namespace", r.Namespace, @@ -425,6 +430,7 @@ func run(ctx context.Context, logger *slog.Logger, clientset kubernetes.Interfac nodes: nodeJournal, jobs: jobJournal, goroutines: goroutineJournal, + states: stateJournal, }, time.Now()) } @@ -855,11 +861,25 @@ func run(ctx context.Context, logger *slog.Logger, clientset kubernetes.Interfac sc := ship.Coverage() shipping = &sc } - if err := spool.WriteCollectionCoverage(time.Now(), startedAt, agentInfo, - podWatcher.SourceHealths(), c, podWatcher.PlacementDrops(), nodeWatcher.Drops(), inv, scan, + // One reading of the clock stamps the snapshot and bounds the window's + // accrual, so the two cannot describe different instants (ADR 0078 §2). + now := time.Now() + sources := podWatcher.SourceHealths() + if err := spool.WriteCollectionCoverage(now, startedAt, agentInfo, + sources, c, podWatcher.PlacementDrops(), nodeWatcher.Drops(), inv, scan, ebpf, probeCoverage, pullCoverage, countCoverage, rejections, shipping); err != nil { logger.Error("spooling collection coverage", "error", err) } + // The same two states the snapshot just carried, folded into the window + // that will outlive them: what holds now is above, how long it held and + // how many episodes it was is here (ADR 0088). + stateJournal.Observe(now, agentStates(sources, shipping)) + records := append(stateJournal.CloseBefore(now), stateJournal.Snapshots()...) + if len(records) > 0 { + if err := spool.WriteAgentStates(now, records); err != nil { + logger.Error("spooling agent states", "error", err) + } + } } // Watchers run until ctx is canceled; a failing watcher must bring the diff --git a/cmd/agent/recover.go b/cmd/agent/recover.go index cea83d9..651ee0c 100644 --- a/cmd/agent/recover.go +++ b/cmd/agent/recover.go @@ -9,11 +9,12 @@ import ( "github.com/RebuildStackCo/runtime-agent/internal/sink" ) -// journals is the five accumulators a restart resumes beside the usage poller. +// journals is the six accumulators a restart resumes beside the usage poller. type journals struct { restarts *journal.Restarts disruptions *journal.Disruptions nodes *journal.NodeEvents + states *journal.States jobs *journal.JobRuns goroutines *journal.Goroutines } @@ -46,7 +47,8 @@ func seedOpenWindows(logger *slog.Logger, spool *sink.Spool, poller *collector.U j.disruptions.Resume(recovery.Disruptions) + j.nodes.Resume(recovery.NodeEvents) + j.jobs.Resume(recovery.JobRuns) + - j.goroutines.Resume(recovery.GoroutineCounts) + j.goroutines.Resume(recovery.GoroutineCounts) + + j.states.Resume(recovery.AgentStates) if recovery.JournalWindows > 0 { logger.Info("resumed open journal windows from the spool", "windows", recovery.JournalWindows, "records", seeded) diff --git a/cmd/agent/states.go b/cmd/agent/states.go new file mode 100644 index 0000000..66afba0 --- /dev/null +++ b/cmd/agent/states.go @@ -0,0 +1,34 @@ +package main + +import ( + "github.com/RebuildStackCo/runtime-agent/internal/journal" + "github.com/RebuildStackCo/runtime-agent/internal/model" +) + +// agentStates is the pass's own states in the shape the window accumulates, +// built from the values the coverage snapshot was just written from so the two +// cannot disagree (ADR 0088 §2). +// +// Every watched source contributes, failing or not: a source with no record is +// one nobody watched, and that is the distinction the window exists to keep. +func agentStates(sources []model.SourceHealth, shipping *model.Shipping) []journal.AgentState { + out := make([]journal.AgentState, 0, len(sources)+1) + for _, src := range sources { + out = append(out, journal.AgentState{ + State: journal.StateFailing, + Subject: src.Name, + Since: src.FailingSince, + }) + } + // Absent when the agent has no backend: an agent told to ship nowhere has + // not stopped shipping, and an hour of "not halted" would claim it had a + // backend to be halted from. + if shipping != nil { + out = append(out, journal.AgentState{ + State: journal.StateHalted, + Subject: journal.SubjectShip, + Since: shipping.HaltedSince, + }) + } + return out +} 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 1046f17..34a593d 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, 0087 +Amended by: 0057, 0058, 0060, 0087, 0088 Amends: 0012 §2, 0033 diff --git a/docs/adr/0087-a-state-says-when-it-began.md b/docs/adr/0087-a-state-says-when-it-began.md index caaf817..1bdf370 100644 --- a/docs/adr/0087-a-state-says-when-it-began.md +++ b/docs/adr/0087-a-state-says-when-it-began.md @@ -4,6 +4,8 @@ Date: 2026-09-18 Status: Accepted +Amended by: 0088 + Amends: 0054, 0067 `sources[].failing` and `shipping.halted` gain a start instant beside them, and diff --git a/docs/adr/0088-a-window-says-how-long-a-state-held.md b/docs/adr/0088-a-window-says-how-long-a-state-held.md new file mode 100644 index 0000000..bc862f6 --- /dev/null +++ b/docs/adr/0088-a-window-says-how-long-a-state-held.md @@ -0,0 +1,131 @@ +# 0088. A window says how long a state held + +Date: 2026-09-18 + +Status: Accepted + +Amends: 0054, 0087 + +Adds the payload kind `agent_states` and a sixth journal accumulator, resumed at +startup on ADR 0077's terms. No new collection, no new privilege, no new read: +its input is the two states `collection_coverage` already carries, taken from +the same call on the same pass. + +## Context + +ADR 0087 gave `sources[].failing` and `shipping.halted` the instant they began, +and said plainly what it did not do: a state that ended is absent from the next +payload by design. That is not an oversight in the field — it is what a state +field is. It describes the present, and this kind supersedes, so the backend is +entitled to keep one row. + +So the defect ADR 0087 closed was half of one. "Is it broken, and since when" is +answerable. "Was it broken" is not: + +``` +13:47 failing_since: 13:47 the payload says so +14:06 (no field) recovered — and the row is clean + nineteen minutes are now unrecoverable +``` + +Two different questions hide under one word. A state that **holds** is a fact +about now, and a boolean with a start is the right shape for it. A state that +**held** is a fact about an interval, and no field describing the present can +carry it. + +This repository has already answered a question of exactly this shape. +`container_restarts` is a window and `restart_counters` is a reading, and the +contract carries the sentence that keeps them apart: *"`restart_counters` is a +reading, and MUST NOT be added to the windows"* (ADR 0034). One fact in two +shapes, with the reader told which is which. The six window kinds beside it +share one key — `(window start, window length)`, hour-aligned — and ADR 0078 +made an open window declare itself open. The agent's own behaviour was the last +subject still reported only in the pre-window shape. + +What is deliberately **not** in scope is the counters. `filter.excluded_*`, +`shipping.deferred`, the puller's outcomes and the intake rejections are +cumulative from `since` and have the same blind spot for the same reason, but +subtracting a previous reading is a different mechanism with its own failure +modes (ADR 0062 priced them), and it is not decided here. + +## Decision + +**1. A window kind for the agent's own states.** `agent_states`, one record per +`(window, state, subject)`, hour-aligned, superseding within its window, written +every pass as a slice of the open window on ADR 0078's terms. Subjects are the +resource classes the agent watches and the agent's own shipping; no object in +the customer's cluster is nameable here, at any resolution. + +**2. Occupancy and episodes, not events.** Each record carries how many +nanoseconds of the window the subject spent in the state and how many episodes +began in it. Two numbers rather than a record per transition, because the cost +is then bounded by construction at whatever flap rate — and because one +nineteen-minute outage and three six-minute ones are different diagnoses that a +single duration would merge. + +**3. Accrual moves forward, and never re-measures.** A pass attributes the +interval since the previous pass, split at any window boundary it crosses. +`accrued_to` is how far a record is true, and is where a later process resumes +from; `open_since` is the marker that tells that process the episode it finds +open is the one already counted rather than a new one. Occupancy is never +recomputed as `now - open_since` — that expression is correct exactly once and +double-counts every span before a restart, which is the whole reason the marker +exists rather than a start instant alone. + +**4. Sampling at the pass loses nothing here, and that is measured.** A pass +runs every minute (`everyPass`), and a run of watch failures becomes a new +episode only after two minutes of quiet (`watchStreakGap`). Two episodes +therefore cannot both begin and end between two passes, and a single one that +does is still seen, because the state is defined with a three-minute tail. The +halt cannot flap at all: it latches until the process ends. + +**5. Observation travels beside occupancy.** Zero occupancy against a full +window is a healthy hour; zero against zero observation is an hour nobody +watched. Without the second number they are the same bytes, which is ADR 0054's +founding defect arriving in a new shape. A restart's gap is in neither: the +first pass of a process accrues nothing, so time the agent was not running is +never claimed as time it was watching. + +**6. Every watched subject writes a record, including a clear one.** A subject +with no record is one nobody watched. This is why a window with nothing wrong in +it is still written, where the journals beside this one write nothing. + +**7. An episode is counted where it was seen, not where it began.** Back-dating +an entry to the instant the episode started would reopen a window that had +already ended and been written final. The instant is not lost — it is in +`open_since` and in the occupancy split across the boundary. + +## What this still cannot say + +Stated here rather than discovered later: + +- **A restart across a window boundary adds one entry.** The earlier window has + ended and is not reopened (ADR 0077 §1), so the continuing episode is seen as + new. It is detectable: `since` in `collection_coverage` moves at the same + moment, so an entry adjacent to a moved `since` is suspect. +- **An episode that closed between two passes undercounts by up to one pass.** + The instant it closed is not knowable from two readings, and the alternative + is to invent it. +- **The marker survives a process restart, not a pod replacement.** The spool is + an `emptyDir` by decision (ADR 0007, ADR 0026); a rescheduled agent resumes + nothing, as ADR 0077 §4 already priced. + +## Consequences + +**Easier.** An outage that ended is reportable for the first time. "No finding +from EndpointSlices this week" acquires an answer that is neither a guess nor a +trip into the customer's logs, and `observed_nanos` makes every finding that +rests on a window quotable with the coverage it actually had. + +**Harder / given up.** A seventh kind in the startup read, so `security.md`'s +closed list moves again — the mechanism ADR 0077 declared and ADR 0080 then +failed to use. One file per hour of a handful of records: negligible in bytes, +and one more payload the backend implements. + +**Not changed.** What is collected about the cluster, what is filtered, what is +read, any other kind's cadence or key, or the one-way protocol. The two states +this measures are the two ADR 0087 named, read from the same call that writes +them to `collection_coverage`, so the window and the snapshot cannot disagree. + +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 54fd5fb..6227583 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -145,3 +145,4 @@ line to paste. - [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) +- [0088. A window says how long a state held](0088-a-window-says-how-long-a-state-held.md) diff --git a/docs/backend-requirements.md b/docs/backend-requirements.md index 3b4d3f7..8e4fc50 100644 --- a/docs/backend-requirements.md +++ b/docs/backend-requirements.md @@ -478,6 +478,28 @@ agent failed on. No field of this payload names an object the customer excluded, and the backend MUST NOT ask for one. +**`agent_states` is the same two states as time, and MUST NOT be added to them** +([ADR 0088](adr/0088-a-window-says-how-long-a-state-held.md)). One record per +(window, state, subject), superseding by (window start, window length): the +backend MUST upsert and MUST NOT append. `collection_coverage` says which states +hold now; this says how much of each hour they held and how many episodes that +was. The relationship is `container_restarts` to `restart_counters`, and the +same rule applies — one is a reading, the other a window. + +- `occupied_nanos` and `entries` are independent and neither derives the other: + one long episode and several short ones can share a duration. Summing + `entries` across windows counts episodes, because an episode continuing from + the previous window contributes occupancy here and no entry. +- **`observed_nanos` bounds both, and the backend MUST NOT read occupancy + without it.** Zero occupancy over a full window is a healthy hour; zero over + zero observation is an hour nobody watched. A restart's gap is in neither + number, which is what makes the second one the honest denominator. +- A restart that crosses a window boundary may add one entry to a continuing + episode. It is detectable: `collection_coverage.since` moves at the same time. +- `subject` is a resource class the agent watches or the agent's own + `shipping`. No object in the customer's cluster appears here at any window + length, and the backend MUST NOT ask for one. + **`network_window` says how much moved, never where it went** ([ADR 0053](adr/0053-network-counters-are-the-pods.md)). The counters are read at the pod's own interfaces, so the payload carries no destination, no peer and no diff --git a/docs/prose_test.go b/docs/prose_test.go index 2b0a49b..924fc58 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); 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 (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}, + // (0075); a window states when it was written (0078); an endpoint is read + // once a minute (0080); GODEBUG names come out of the build (0081); a + // request names its agent (0083); a listener names what the filters excluded + // (0084); the image states what it is made of (0085); a failing read says + // since when (0087); and a window says how long a state held (0088), whose + // row is the section's raise and whose diff moves the startup-read list to + // the seven kinds it reads. + {"security.md", 1139, 367}, + // Each raise names its obligation here too. A payload's size is no longer + // its 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 + // (0082); a request may name an agent with no version (0083); an absent + // instant is told apart from an absent state (0087); and a duration arrives + // with the observation bounding it (0088). The last two moved the section. + {"backend-requirements.md", 851, 673}, } func TestNoCommentRunIsLongerThanAPointerToAnADR(t *testing.T) { diff --git a/docs/security.md b/docs/security.md index 8b823d0..d9756cd 100644 --- a/docs/security.md +++ b/docs/security.md @@ -272,9 +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 six kinds and no others — `usage_snapshot`, `container_restarts`, - `pod_disruptions`, `node_lifecycle`, `job_runs`, `goroutine_counts` — and it - deletes nothing. + of seven kinds and no others — `usage_snapshot`, `container_restarts`, + `pod_disruptions`, `node_lifecycle`, `job_runs`, `goroutine_counts`, + `agent_states` — 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. @@ -556,6 +556,7 @@ nothing to what leaves your cluster. | `listening_ports` | Per collected workload container and build: the TCP ports its Go processes accept connections on, each with whether it is bound to loopback, and — for a port the agent asked — whether the Go pprof index answered there ([ADR 0082](adr/0082-a-confirmed-endpoint-is-named-on-its-port.md)). Read from the processes' own sockets, so a port nothing declared is here and a declared port nothing binds is not ([ADR 0056](adr/0056-a-pprof-endpoint-is-proved-not-probed.md)). No address, and no connection the workload has open | no | | `ebpf_profile` | One capture: allow-list-filtered symbolized pprof bytes, keyed by workload, image digest and capture window | no | | `pprof_profile` | One ten-second CPU profile fetched from the workload's own `/debug/pprof` endpoint, allow-list-filtered, under the same key ([ADR 0058](adr/0058-the-pull-starts-a-profiler-and-the-binary-says-whose-code-it-is.md)). The mappings — the executable's path on disk and its build ID — the sample labels, and the full source paths are removed with the redacted frames, so what remains is function names your allow-list admits, base file names and line numbers. Beside it travel the counts of what was redacted, never what | no | +| `agent_states` | Per window, for each of the agent's own states: how much of the hour it held, how many episodes that was, and how much of the hour the agent was running to watch at all ([ADR 0088](adr/0088-a-window-says-how-long-a-state-held.md)). Its subjects are this agent and the resource classes it watches — `services`, `endpoint_slices` — never an object in your cluster. Nothing here is about your workloads: it is how the agent answers for its own outages | no | | `goroutine_counts` | Per sampled process and window: how many goroutines it was running at each reading, and the pod, container and build it was read from ([ADR 0080](adr/0080-the-index-page-already-counts-the-goroutines.md)). A count and a timestamp. No stack, no function name and nothing about what those goroutines were doing | no | Each declares its provenance in a `source` field — `structural` (read from a diff --git a/internal/journal/states.go b/internal/journal/states.go new file mode 100644 index 0000000..d5b4003 --- /dev/null +++ b/internal/journal/states.go @@ -0,0 +1,306 @@ +package journal + +import ( + "sort" + "sync" + "time" +) + +// The agent's own states, as time rather than as a flag. `collection_coverage` +// says which states hold now and since when (ADR 0087); it cannot say that a +// state ended, because it supersedes and a state that ended leaves no field +// behind. This journal is the other half: how much of each window a state +// occupied, and how many episodes made it up (ADR 0088). + +// State names, and the one subject that is not a resource class. +const ( + StateFailing = "failing" + StateHalted = "halted" + SubjectShip = "shipping" + subjectKeySep = "\x00" +) + +// AgentState is one of the agent's own states as one pass found it. Since is +// when the current episode began, and nil is the claim that the subject is not +// in this state — the same two values `collection_coverage` carries, read from +// the same call, so the window and the snapshot cannot disagree. +type AgentState struct { + State string + Subject string + Since *time.Time +} + +// StateRecord is one subject's time in one state within one window. +// +// Occupied and Observed are both needed: zero occupancy against a full window +// is a healthy hour, and zero occupancy against zero observation is an hour +// nobody watched. Without the second, the two are the same bytes — which is the +// defect ADR 0054 exists to prevent, arriving in a new shape. +type StateRecord struct { + State string `json:"state"` + Subject string `json:"subject"` + + WindowStart time.Time `json:"window_start"` + WindowSeconds int64 `json:"window_seconds"` + + // OccupiedNanos is time in the state; ObservedNanos is time the agent was + // running and watching this subject at all. A restart's gap is in neither. + OccupiedNanos int64 `json:"occupied_nanos"` + ObservedNanos int64 `json:"observed_nanos"` + // Entries is episodes that began within this window. A state continuing + // from the previous window contributes occupancy here and no entry, so + // summing entries across windows counts episodes rather than windows. + Entries int64 `json:"entries"` + + // AccruedTo is the instant the two durations above are true up to, and what + // a later process resumes accrual from. It is never used to re-measure a + // span already counted (ADR 0088 §3). + AccruedTo time.Time `json:"accrued_to"` + // OpenSince is the episode's start when the state was still open at + // AccruedTo, and absent otherwise. It is what tells a resumed process that + // an episode it finds open is the one already counted, not a new one. + OpenSince *time.Time `json:"open_since,omitempty"` +} + +type stateKey struct { + state string + subject string + start int64 +} + +// lastSeen is what the previous pass found for one subject: how far accrual has +// reached, and which episode was open there. +type lastSeen struct { + accruedTo time.Time + since *time.Time +} + +// States accumulates the agent's own states into wall-clock-aligned windows, +// the same windows the usage rollups use. +type States struct { + mu sync.Mutex + windowLength time.Duration + open map[stateKey]*StateRecord + seen map[string]lastSeen +} + +// NewStates returns an accumulator over windows of the given length, which must +// be positive. +func NewStates(windowLength time.Duration) *States { + if windowLength <= 0 { + panic("journal: window length must be positive") + } + return &States{ + windowLength: windowLength, + open: map[stateKey]*StateRecord{}, + seen: map[string]lastSeen{}, + } +} + +// Observe folds the interval since the previous pass into the windows it spans, +// for every subject the pass looked at. +// +// A subject seen for the first time accrues nothing: there is no interval to +// attribute yet, and on a resumed process the span before the restart is +// already in the record on disk and must not be measured a second time. +func (s *States) Observe(at time.Time, states []AgentState) { + at = at.UTC() + s.mu.Lock() + defer s.mu.Unlock() + + for _, st := range states { + if st.State == "" || st.Subject == "" { + continue + } + key := st.State + subjectKeySep + st.Subject + prev, known := s.seen[key] + if known && at.After(prev.accruedTo) { + s.accrue(st, prev, at) + } + if st.Since != nil { + s.enter(st, prev, known, at) + } + s.stamp(st, at) + s.seen[key] = lastSeen{accruedTo: at, since: st.Since} + } +} + +// accrue attributes [prev.accruedTo, at] to the windows it spans: all of it as +// observed, and the part of it the state was open as occupied. +// +// Open at both ends means open throughout — an episode is separated from the +// next by longer than a pass (ADR 0088 §4). Open at the start and closed at the +// end attributes no occupancy at all: when it closed is not knowable from two +// readings, and guessing would be the payload's only invented number. +func (s *States) accrue(st AgentState, prev lastSeen, at time.Time) { + from := prev.accruedTo + var occupiedFrom time.Time + switch { + case st.Since == nil: + occupiedFrom = time.Time{} // closed at this end: nothing to attribute + case prev.since != nil: + occupiedFrom = from // open at both ends + default: + occupiedFrom = maxTime(*st.Since, from) // opened within the interval + } + for _, span := range s.split(from, at) { + rec := s.record(st, span.start) + rec.ObservedNanos += span.end.Sub(span.start).Nanoseconds() + // Each window this interval touched is now true up to its own end, not + // up to wherever the pass happened to read the clock. + rec.AccruedTo, rec.OpenSince = span.end, st.Since + if occupiedFrom.IsZero() { + continue + } + start := maxTime(occupiedFrom, span.start) + if start.Before(span.end) { + rec.OccupiedNanos += span.end.Sub(start).Nanoseconds() + } + } +} + +// enter counts an episode, in the window where this pass first saw it open. +// +// The instant the episode began is not where it is counted: an episode that +// began just before a window closed would otherwise reopen a window already +// flushed. What it began at is in OpenSince and in the occupancy; what this +// counts is episodes, and counting one where it was seen keeps every window +// final once it has ended. +func (s *States) enter(st AgentState, prev lastSeen, known bool, at time.Time) { + if known && prev.since != nil && prev.since.Equal(*st.Since) { + return // the same episode, still running + } + rec := s.record(st, at) + if !known && rec.OpenSince != nil && rec.OpenSince.Equal(*st.Since) { + return // resumed: this episode was counted by the process before this one + } + rec.Entries++ +} + +// stamp records how far this subject's accrual has reached and whether it is +// still open, on the window holding at — the record a resumed process reads. +func (s *States) stamp(st AgentState, at time.Time) { + rec := s.record(st, at) + rec.AccruedTo = at + rec.OpenSince = st.Since +} + +// record returns the open record for one subject in the window holding at, +// creating it if this is the first thing to land there. +func (s *States) record(st AgentState, at time.Time) *StateRecord { + start := at.UTC().Truncate(s.windowLength) + key := stateKey{state: st.State, subject: st.Subject, start: start.UnixNano()} + rec := s.open[key] + if rec == nil { + rec = &StateRecord{ + State: st.State, + Subject: st.Subject, + WindowStart: start, + WindowSeconds: int64(s.windowLength / time.Second), + } + s.open[key] = rec + } + return rec +} + +// span is one window's share of an interval. +type span struct{ start, end time.Time } + +// split cuts [from, to] at window boundaries, so an interval that crosses one +// is attributed to both windows rather than to whichever end it was read at. +func (s *States) split(from, to time.Time) []span { + var out []span + for from.Before(to) { + end := from.Truncate(s.windowLength).Add(s.windowLength) + if end.After(to) { + end = to + } + out = append(out, span{start: from, end: end}) + from = end + } + return out +} + +// Snapshots returns deep copies of every open record, sorted so the payload +// bytes are deterministic (the golden contract). +func (s *States) Snapshots() []StateRecord { + s.mu.Lock() + defer s.mu.Unlock() + out := make([]StateRecord, 0, len(s.open)) + for _, rec := range s.open { + out = append(out, rec.clone()) + } + sortStateRecords(out) + return out +} + +// Resume seeds open windows from records a previous process wrote, so a restart +// continues the window rather than writing a subset of it back over itself +// (ADR 0077). A key already present is left alone. +// +// The gap between the previous process's last accrual and this one's first pass +// is attributed to nothing: the agent was not watching, and ObservedNanos is +// the number that says so. +func (s *States) Resume(records []StateRecord) int { + s.mu.Lock() + defer s.mu.Unlock() + var seeded int + for _, rec := range records { + key := stateKey{state: rec.State, subject: rec.Subject, start: rec.WindowStart.UnixNano()} + if _, ok := s.open[key]; ok { + continue + } + resumed := rec.clone() + s.open[key] = &resumed + seeded++ + } + return seeded +} + +// CloseBefore removes and returns every record whose window ended at or before +// now. Removal is what bounds memory, exactly as it is for the journals beside +// this one. +func (s *States) CloseBefore(now time.Time) []StateRecord { + s.mu.Lock() + defer s.mu.Unlock() + var out []StateRecord + for key, rec := range s.open { + if rec.WindowStart.Add(s.windowLength).After(now) { + continue + } + out = append(out, rec.clone()) + delete(s.open, key) + } + sortStateRecords(out) + return out +} + +func (r *StateRecord) clone() StateRecord { + out := *r + if r.OpenSince != nil { + since := *r.OpenSince + out.OpenSince = &since + } + return out +} + +func sortStateRecords(records []StateRecord) { + sort.Slice(records, func(i, j int) bool { + a, b := records[i], records[j] + switch { + case !a.WindowStart.Equal(b.WindowStart): + return a.WindowStart.Before(b.WindowStart) + case a.State != b.State: + return a.State < b.State + default: + return a.Subject < b.Subject + } + }) +} + +func maxTime(a, b time.Time) time.Time { + if a.After(b) { + return a + } + return b +} diff --git a/internal/journal/states_test.go b/internal/journal/states_test.go new file mode 100644 index 0000000..74c74f3 --- /dev/null +++ b/internal/journal/states_test.go @@ -0,0 +1,222 @@ +package journal + +import ( + "testing" + "time" +) + +var stateT0 = time.Date(2026, 9, 18, 13, 0, 0, 0, time.UTC) + +// failing builds the one state shape these tests use. +func failing(subject string, since *time.Time) []AgentState { + return []AgentState{{State: StateFailing, Subject: subject, Since: since}} +} + +func at(d time.Duration) *time.Time { + t := stateT0.Add(d) + return &t +} + +// only returns the single record for a subject, failing the test if the +// accumulator holds any other number of them. +func only(t *testing.T, s *States) StateRecord { + t.Helper() + got := s.Snapshots() + if len(got) != 1 { + t.Fatalf("holding %d records, want exactly one: %+v", len(got), got) + } + return got[0] +} + +// The first pass has no interval behind it. Accruing from the window's start +// would credit the agent with watching a span it was not running for. +func TestTheFirstPassAccruesNothing(t *testing.T) { + s := NewStates(time.Hour) + s.Observe(stateT0.Add(10*time.Minute), failing("services", nil)) + + if rec := only(t, s); rec.ObservedNanos != 0 || rec.OccupiedNanos != 0 { + t.Errorf("first pass accrued observed=%d occupied=%d, want both zero", + rec.ObservedNanos, rec.OccupiedNanos) + } +} + +// A healthy subject is not an absent one. Observation grows while occupancy +// stays at zero, which is what separates "watched and fine" from "not watched". +func TestAHealthySubjectAccruesObservationAndNoOccupancy(t *testing.T) { + s := NewStates(time.Hour) + for i := range 4 { + s.Observe(stateT0.Add(time.Duration(i)*time.Minute), failing("services", nil)) + } + + rec := only(t, s) + if want := (3 * time.Minute).Nanoseconds(); rec.ObservedNanos != want { + t.Errorf("observed %d, want %d", rec.ObservedNanos, want) + } + if rec.OccupiedNanos != 0 || rec.Entries != 0 { + t.Errorf("a healthy subject shows occupancy %d and %d entries", rec.OccupiedNanos, rec.Entries) + } +} + +// The episode that started and ended between two payloads — the case a boolean +// in a superseding payload cannot report at all (ADR 0087, ADR 0088). +func TestAnEpisodeThatEndedIsStillCounted(t *testing.T) { + s := NewStates(time.Hour) + s.Observe(stateT0, failing("endpoint_slices", nil)) + s.Observe(stateT0.Add(time.Minute), failing("endpoint_slices", at(30*time.Second))) + s.Observe(stateT0.Add(2*time.Minute), failing("endpoint_slices", at(30*time.Second))) + s.Observe(stateT0.Add(3*time.Minute), failing("endpoint_slices", nil)) + + rec := only(t, s) + if rec.Entries != 1 { + t.Errorf("entries %d, want 1", rec.Entries) + } + // Open from 13:00:30 to the last pass that saw it open, at 13:02. The pass + // that found it closed attributes nothing: when it closed is not knowable + // from two readings. + if want := (90 * time.Second).Nanoseconds(); rec.OccupiedNanos != want { + t.Errorf("occupied %d, want %d", rec.OccupiedNanos, want) + } + if rec.OpenSince != nil { + t.Errorf("a closed episode left open_since at %v", rec.OpenSince) + } +} + +// Two episodes and one long one are different diagnoses, so the pair of numbers +// has to tell them apart. +func TestTwoEpisodesInOneWindowCountTwice(t *testing.T) { + s := NewStates(time.Hour) + s.Observe(stateT0, failing("services", nil)) + s.Observe(stateT0.Add(time.Minute), failing("services", at(time.Minute))) + s.Observe(stateT0.Add(2*time.Minute), failing("services", nil)) + s.Observe(stateT0.Add(10*time.Minute), failing("services", at(10*time.Minute))) + + if rec := only(t, s); rec.Entries != 2 { + t.Errorf("entries %d, want 2 — two episodes, not one long one", rec.Entries) + } +} + +// An episode running when the window turns over belongs to both windows: its +// occupancy is split at the boundary, and it is one episode, counted once. +func TestAnEpisodeCrossingAWindowIsSplitAndCountedOnce(t *testing.T) { + s := NewStates(time.Hour) + since := at(50 * time.Minute) + s.Observe(stateT0.Add(50*time.Minute), failing("services", since)) + s.Observe(stateT0.Add(70*time.Minute), failing("services", since)) + + got := s.Snapshots() + if len(got) != 2 { + t.Fatalf("holding %d records, want one per window: %+v", len(got), got) + } + first, second := got[0], got[1] + if want := (10 * time.Minute).Nanoseconds(); first.OccupiedNanos != want { + t.Errorf("first window occupied %d, want %d", first.OccupiedNanos, want) + } + if want := (10 * time.Minute).Nanoseconds(); second.OccupiedNanos != want { + t.Errorf("second window occupied %d, want %d", second.OccupiedNanos, want) + } + if first.Entries != 1 || second.Entries != 0 { + t.Errorf("entries %d and %d, want 1 and 0: one episode, not two", + first.Entries, second.Entries) + } +} + +// The invariant a resume turns on: occupancy moves forward from where it was +// left, and is never re-measured from the instant the episode began. Getting +// this wrong counts the whole pre-restart span a second time. +func TestAResumedWindowDoesNotMeasureItsOwnPastAgain(t *testing.T) { + since := at(0) + first := NewStates(time.Hour) + first.Observe(stateT0, failing("services", since)) + first.Observe(stateT0.Add(20*time.Minute), failing("services", since)) + saved := only(t, first) + + // The restart: nothing accrues for the gap, because nothing was watching. + second := NewStates(time.Hour) + if seeded := second.Resume([]StateRecord{saved}); seeded != 1 { + t.Fatalf("resumed %d records, want 1", seeded) + } + second.Observe(stateT0.Add(40*time.Minute), failing("services", since)) + second.Observe(stateT0.Add(45*time.Minute), failing("services", since)) + + rec := only(t, second) + if want := (25 * time.Minute).Nanoseconds(); rec.OccupiedNanos != want { + t.Errorf("occupied %d, want %d — 20 before the restart and 5 after, and "+ + "nothing for the 20 nobody watched", rec.OccupiedNanos, want) + } + if rec.Entries != 1 { + t.Errorf("entries %d, want 1: the resumed episode is the one already counted", rec.Entries) + } +} + +// A resumed process that finds a different episode open counts it, because it +// is one: the marker only silences the episode it already holds. +func TestAResumedProcessCountsANewEpisode(t *testing.T) { + since := at(0) + saved := StateRecord{ + State: StateFailing, Subject: "services", + WindowStart: stateT0, WindowSeconds: 3600, + AccruedTo: stateT0.Add(20 * time.Minute), OpenSince: since, + } + s := NewStates(time.Hour) + s.Resume([]StateRecord{saved}) + s.Observe(stateT0.Add(40*time.Minute), failing("services", at(35*time.Minute))) + + if rec := only(t, s); rec.Entries != 1 { + t.Errorf("entries %d, want 1 for the episode that began after the restart", rec.Entries) + } +} + +// Resume never overwrites an observation this process made (ADR 0077 §2). +func TestResumeLeavesAKeyThisProcessAlreadyHolds(t *testing.T) { + s := NewStates(time.Hour) + s.Observe(stateT0, failing("services", nil)) + s.Observe(stateT0.Add(time.Minute), failing("services", nil)) + + stale := StateRecord{ + State: StateFailing, Subject: "services", + WindowStart: stateT0, WindowSeconds: 3600, OccupiedNanos: 999, + } + if seeded := s.Resume([]StateRecord{stale}); seeded != 0 { + t.Errorf("resumed %d records over a live one, want 0", seeded) + } + if rec := only(t, s); rec.OccupiedNanos == 999 { + t.Error("a record on disk displaced one this process observed") + } +} + +// A window that has ended leaves the accumulator, which is what bounds memory. +func TestCloseBeforeReleasesEndedWindows(t *testing.T) { + s := NewStates(time.Hour) + s.Observe(stateT0, failing("services", nil)) + s.Observe(stateT0.Add(70*time.Minute), failing("services", nil)) + + closed := s.CloseBefore(stateT0.Add(80 * time.Minute)) + if len(closed) != 1 || !closed[0].WindowStart.Equal(stateT0) { + t.Fatalf("closed %+v, want the first window alone", closed) + } + if got := s.Snapshots(); len(got) != 1 || got[0].WindowStart.Equal(stateT0) { + t.Errorf("after closing, holding %+v, want the open window alone", got) + } +} + +// The halt is the other subject, and it is the one whose payload cannot arrive: +// a halted agent ships nothing, so the window is what a later reader has. +func TestTheHaltIsAccruedLikeAnyOtherState(t *testing.T) { + s := NewStates(time.Hour) + halted := []AgentState{{State: StateHalted, Subject: SubjectShip}} + s.Observe(stateT0, halted) + s.Observe(stateT0.Add(time.Minute), halted) + since := at(time.Minute) + s.Observe(stateT0.Add(5*time.Minute), []AgentState{ + {State: StateHalted, Subject: SubjectShip, Since: since}, + }) + + rec := only(t, s) + if rec.Entries != 1 || rec.OpenSince == nil || !rec.OpenSince.Equal(*since) { + t.Errorf("entries %d, open_since %v, want one episode open since %s", + rec.Entries, rec.OpenSince, since) + } + if want := (4 * time.Minute).Nanoseconds(); rec.OccupiedNanos != want { + t.Errorf("occupied %d, want %d", rec.OccupiedNanos, want) + } +} diff --git a/internal/sink/recover.go b/internal/sink/recover.go index 328ecfe..1039535 100644 --- a/internal/sink/recover.go +++ b/internal/sink/recover.go @@ -39,7 +39,10 @@ type Recovery struct { // GoroutineCounts resumes on the same terms, and is here rather than beside // Records because it is a window of readings rather than of usage (ADR 0080). GoroutineCounts []journal.GoroutineRecord - // JournalWindows is how many journal window files the five above came from. + // AgentStates resumes so that a restart within an hour continues that hour's + // occupancy instead of measuring its own past a second time (ADR 0088 §3). + AgentStates []journal.StateRecord + // JournalWindows is how many journal window files the six above came from. JournalWindows int } @@ -81,12 +84,12 @@ func (s *Spool) RecoverOpenWindows(now time.Time) (Recovery, error) { } s.recoverJournals(entries, now, &rec) journalRecords := len(rec.Restarts) + len(rec.Disruptions) + len(rec.NodeEvents) + - len(rec.JobRuns) + len(rec.GoroutineCounts) + len(rec.JobRuns) + len(rec.GoroutineCounts) + len(rec.AgentStates) s.countRecovery(len(rec.Records), rec.Windows, len(rec.Skipped), journalRecords, rec.JournalWindows) return rec, nil } -// recoverJournals reads the five journal windows still open at now. +// recoverJournals reads the six journal windows still open at now. // // A journal window's file is the same name whether the window is open or final, // so the only question is whether the window has ended: one that has is already @@ -116,6 +119,9 @@ func (s *Spool) recoverJournals(entries []os.DirEntry, now time.Time, rec *Recov case strings.HasPrefix(name, goroutineCountsNamePrefix): found, err = resumeInto(s, name, goroutineCountsNamePrefix, kindGoroutineCounts, now, &rec.GoroutineCounts, func(r journal.GoroutineRecord) (time.Time, int64) { return r.WindowStart, r.WindowSeconds }) + case strings.HasPrefix(name, agentStatesNamePrefix): + found, err = resumeInto(s, name, agentStatesNamePrefix, kindAgentStates, now, &rec.AgentStates, + func(r journal.StateRecord) (time.Time, int64) { return r.WindowStart, r.WindowSeconds }) default: continue } diff --git a/internal/sink/recover_test.go b/internal/sink/recover_test.go index bd52639..1e4f0ad 100644 --- a/internal/sink/recover_test.go +++ b/internal/sink/recover_test.go @@ -451,3 +451,62 @@ func TestARestartInsideAnOpenWindowKeepsTheSeriesItAlreadyHeld(t *testing.T) { "already had plus the new one: %+v", got, payload.Records[0].Samples) } } + +// agentStateRecords is one subject's open window, as the accumulator holds it +// when a process stops mid-hour. +func agentStateRecords(openSince time.Time) []journal.StateRecord { + return []journal.StateRecord{{ + State: journal.StateFailing, Subject: "endpoint_slices", + WindowStart: recoverWindow, WindowSeconds: 3600, + OccupiedNanos: (5 * time.Minute).Nanoseconds(), + ObservedNanos: (12 * time.Minute).Nanoseconds(), + Entries: 1, + AccruedTo: recoverWindow.Add(12 * time.Minute), + OpenSince: &openSince, + }} +} + +// The loss this kind is exposed to: occupancy is an advance, not a reading, so +// a restart that resumed by re-measuring from the episode's start would count +// the pre-restart span twice, and one that resumed from nothing would lose it. +// Neither shows up as an error — only as a number (ADR 0088 §3). +func TestARestartInsideAnOpenStateWindowNeitherLosesNorRepeatsIt(t *testing.T) { + s, _ := recoverSpool(t) + openSince := recoverWindow.Add(7 * time.Minute) + if err := s.WriteAgentStates(capturedAt, agentStateRecords(openSince)); err != nil { + t.Fatal(err) + } + + rec, err := s.RecoverOpenWindows(midWindow) + if err != nil { + t.Fatal(err) + } + if rec.JournalWindows != 1 || len(rec.AgentStates) != 1 { + t.Fatalf("recovered %d journal windows carrying %d records, want 1 and 1", + rec.JournalWindows, len(rec.AgentStates)) + } + + resumed := journal.NewStates(time.Hour) + if seeded := resumed.Resume(rec.AgentStates); seeded != 1 { + t.Fatalf("seeded %d records, want 1", seeded) + } + // Two passes of the new process, as cmd/agent runs them: the first accrues + // nothing, because the span before it belongs to the process that stopped. + for _, at := range []time.Time{midWindow, midWindow.Add(time.Minute)} { + resumed.Observe(at, []journal.AgentState{ + {State: journal.StateFailing, Subject: "endpoint_slices", Since: &openSince}, + }) + } + + got := resumed.Snapshots() + if len(got) != 1 { + t.Fatalf("holding %d records, want 1: %+v", len(got), got) + } + if want := (6 * time.Minute).Nanoseconds(); got[0].OccupiedNanos != want { + t.Errorf("occupied %d, want %d — five accrued before the restart and one after", + got[0].OccupiedNanos, want) + } + if got[0].Entries != 1 { + t.Errorf("entries %d, want 1: the resumed episode is the one already counted", got[0].Entries) + } +} diff --git a/internal/sink/registry.go b/internal/sink/registry.go index bbd87e8..f29a9e0 100644 --- a/internal/sink/registry.go +++ b/internal/sink/registry.go @@ -95,6 +95,18 @@ var registry = []PayloadKind{ Cadence: everyPass, ADR: "0054", }, + { + // The other half of the kind above: coverage says which of the agent's + // own states hold now, this says how much of each window they held and + // how many episodes that was. Written every pass, including one with + // every state clear — a clear window is the observation (ADR 0088). + Kind: "agent_states", + Source: SourceAgent, + NaturalKey: "(window start, window length)", + Delivery: DeliverySupersedes, + Cadence: everyPass, + ADR: "0088", + }, { Kind: "usage_snapshot", Source: SourceMeasured, diff --git a/internal/sink/spool.go b/internal/sink/spool.go index 75c29b9..50f6533 100644 --- a/internal/sink/spool.go +++ b/internal/sink/spool.go @@ -425,6 +425,22 @@ type nodeLifecyclePayload struct { Records []journal.NodeEventRecord `json:"records"` } +// agentStatesPayload is how much of one window each of the agent's own states +// occupied, and how many episodes made it up (ADR 0088). +// +// Its subject is this agent rather than the cluster, which is why it carries +// observation beside occupancy: an hour the agent did not run is an hour it +// cannot answer for, and a zero that means "fine" must not read like one that +// means "nobody was watching". +type agentStatesPayload struct { + Kind string `json:"kind"` + Source string `json:"source"` + CapturedAt time.Time `json:"captured_at"` + WindowStart time.Time `json:"window_start"` + WindowSeconds int64 `json:"window_seconds"` + Records []journal.StateRecord `json:"records"` +} + // jobRunsPayload is every finished Job run of one window: one file per window // holding many records, like the restart and disruption journals (ADR 0029). // @@ -594,12 +610,14 @@ const ( kindNodeLifecycle = "node_lifecycle" kindJobRuns = "job_runs" kindGoroutineCounts = "goroutine_counts" + kindAgentStates = "agent_states" restartsNamePrefix = "restarts-" disruptionsNamePrefix = "disruptions-" nodeLifecycleNamePrefix = "node-lifecycle-" jobRunsNamePrefix = "job-runs-" goroutineCountsNamePrefix = "goroutine-counts-" + agentStatesNamePrefix = "agent-states-" ) // journalWindowName is the filename of one journal window, and the inverse of @@ -784,6 +802,34 @@ func (s *Spool) WritePodDisruptions(capturedAt time.Time, records []journal.Disr return nil } +// WriteAgentStates writes the agent's own state windows, one file per window. +// +// Unlike the journals beside it, a window with nothing to report is still +// written: its records carry the observation that says the agent was running +// and the states were clear, which is the whole difference between a quiet hour +// and an absent one. +func (s *Spool) WriteAgentStates(capturedAt time.Time, records []journal.StateRecord) error { + grouped := make(map[windowKey][]journal.StateRecord) + for _, r := range records { + k := windowKey{start: r.WindowStart, seconds: r.WindowSeconds} + grouped[k] = append(grouped[k], r) + } + for k, group := range grouped { + payload := agentStatesPayload{ + Kind: kindAgentStates, + Source: SourceAgent, + CapturedAt: capturedAt, + WindowStart: k.start, + WindowSeconds: k.seconds, + Records: group, + } + if err := s.write(payload.Kind, journalWindowName(agentStatesNamePrefix, k), payload); err != nil { + return err + } + } + return nil +} + // WriteNodeLifecycle writes the node arrivals and departures of each window they // belong to, one file per window, atomically replacing that window's previous // file. records must already be in a deterministic order (the accumulator sorts diff --git a/internal/sink/spool_test.go b/internal/sink/spool_test.go index dc89490..f0fff65 100644 --- a/internal/sink/spool_test.go +++ b/internal/sink/spool_test.go @@ -2146,3 +2146,49 @@ func TestAHaltedShipperDatesTheHaltInThePayload(t *testing.T) { t.Errorf("halted_since is %v, want %s", payload.Shipping.HaltedSince, haltedAt) } } + +// fixedAgentStates is the hour the payload exists to make legible: one cache +// that was unfed for nineteen minutes across two episodes and is fed again, one +// that was clear throughout, and a shipper still halted when the hour ended. +// +// The clear cache is the reason the payload is written at all — without its +// record, "no finding from services" and "services was never watched" are the +// same absence (ADR 0088 §2). +func fixedAgentStates() []journal.StateRecord { + accruedTo := windowStart.Add(time.Hour) + haltedSince := windowStart.Add(41 * time.Minute) + return []journal.StateRecord{ + { + State: journal.StateFailing, Subject: "endpoint_slices", + WindowStart: windowStart, WindowSeconds: 3600, + OccupiedNanos: (19 * time.Minute).Nanoseconds(), + ObservedNanos: time.Hour.Nanoseconds(), + Entries: 2, + AccruedTo: accruedTo, + }, + { + State: journal.StateFailing, Subject: "services", + WindowStart: windowStart, WindowSeconds: 3600, + ObservedNanos: time.Hour.Nanoseconds(), + AccruedTo: accruedTo, + }, + { + State: journal.StateHalted, Subject: journal.SubjectShip, + WindowStart: windowStart, WindowSeconds: 3600, + OccupiedNanos: (19 * time.Minute).Nanoseconds(), + ObservedNanos: time.Hour.Nanoseconds(), + Entries: 1, + AccruedTo: accruedTo, + OpenSince: &haltedSince, + }, + } +} + +func TestGoldenAgentStatesPayload(t *testing.T) { + s, dir := newTestSpool(t) + if err := s.WriteAgentStates(capturedAt, fixedAgentStates()); err != nil { + t.Fatal(err) + } + name := fmt.Sprintf("agent-states-%d-3600.json", windowStart.Unix()) + checkGolden(t, filepath.Join(dir, name), "agent-states.golden.json") +} diff --git a/internal/sink/testdata/agent-states.golden.json b/internal/sink/testdata/agent-states.golden.json new file mode 100644 index 0000000..e2c1098 --- /dev/null +++ b/internal/sink/testdata/agent-states.golden.json @@ -0,0 +1,40 @@ +{ + "kind": "agent_states", + "source": "agent", + "captured_at": "2026-08-06T10:12:00Z", + "window_start": "2026-08-06T10:00:00Z", + "window_seconds": 3600, + "records": [ + { + "state": "failing", + "subject": "endpoint_slices", + "window_start": "2026-08-06T10:00:00Z", + "window_seconds": 3600, + "occupied_nanos": 1140000000000, + "observed_nanos": 3600000000000, + "entries": 2, + "accrued_to": "2026-08-06T11:00:00Z" + }, + { + "state": "failing", + "subject": "services", + "window_start": "2026-08-06T10:00:00Z", + "window_seconds": 3600, + "occupied_nanos": 0, + "observed_nanos": 3600000000000, + "entries": 0, + "accrued_to": "2026-08-06T11:00:00Z" + }, + { + "state": "halted", + "subject": "shipping", + "window_start": "2026-08-06T10:00:00Z", + "window_seconds": 3600, + "occupied_nanos": 1140000000000, + "observed_nanos": 3600000000000, + "entries": 1, + "accrued_to": "2026-08-06T11:00:00Z", + "open_since": "2026-08-06T10:41:00Z" + } + ] +}