From eafa100718a749dbadc5bf054c76c1c35a98bf4a Mon Sep 17 00:00:00 2001 From: Konstantins Feofantovs Date: Sun, 23 Aug 2026 20:29:57 +0300 Subject: [PATCH] feat: finished Job runs ship as a windowed journal (ADR 0029) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Usage rollups systematically misrepresent short-lived workloads. A Job that runs ninety seconds inside an hour-long window reports a tiny average, and covered_nanoseconds says the coverage was low without saying that the workload was a batch run that started, worked and finished. That gap lands where it costs most: CronJob requests are routinely set from the peak run and left, so a batch fleet is where overprovisioning accumulates quietly. `job_runs` is the denominator. Windowed and superseding like the other two journals, for ADR 0021's reason — a fleet of CronJobs can finish hundreds of runs an hour, and one file per run would put the spool's file count under the cluster's control. Each record carries the run's instants, its outcome and failure reason, its pod success/failure counts, and the declared parallelism, completions and backoffLimit: without the last two, "three succeeded" is unreadable, and without the pod counts a run that retried twice looks free. Runs are deliberately NOT baselined on first observation, which reverses what the plan said. ADR 0020 §5 baselines restart counters because a counter carries no time; a Job carries startTime, completionTime and its condition's transition time, so an already-finished run files itself in the window where it actually finished. That is ADR 0021's distinction, and it means a freshly started agent picks up the cluster's recent batch history correctly dated instead of discarding it. The burst is bounded by the cluster's own retention — three successful and one failed Job per CronJob by default. A run's Job is filtered on four objects rather than three: namespace, namespace annotation, the owning CronJob (ADR 0028's workload step), then the Job's own annotations or its pod template's. The last is what keeps this honest — before ADR 0028 the documented opt-out was a pod-template annotation, so a customer who used it has excluded pods, and reading only the Job object would ship facts about a workload whose pods the agent refuses to measure. Template metadata only; the env/args/command beneath it are never read. Job counters are kept apart from pod counters: one Job makes many pods, so a shared number would answer neither question. No new RBAC and no new informer — the Job handler rides the watch already running for owner resolution. Co-Authored-By: Claude Opus 5 (1M context) --- cmd/agent/main.go | 44 +++++ docs/adr/0029-job-runs.md | 121 ++++++++++++ docs/adr/README.md | 1 + docs/plans/job-and-rollout-facts.md | 18 +- docs/security.md | 19 +- internal/collector/filter.go | 76 ++++++++ internal/collector/jobs.go | 201 ++++++++++++++++++++ internal/collector/jobs_test.go | 190 ++++++++++++++++++ internal/collector/pods.go | 55 +++++- internal/journal/jobs.go | 155 +++++++++++++++ internal/journal/jobs_test.go | 95 +++++++++ internal/sink/registry.go | 10 + internal/sink/spool.go | 46 +++++ internal/sink/spool_test.go | 80 ++++++++ internal/sink/testdata/job-runs.golden.json | 60 ++++++ 15 files changed, 1156 insertions(+), 15 deletions(-) create mode 100644 docs/adr/0029-job-runs.md create mode 100644 internal/collector/jobs.go create mode 100644 internal/collector/jobs_test.go create mode 100644 internal/journal/jobs.go create mode 100644 internal/journal/jobs_test.go create mode 100644 internal/sink/testdata/job-runs.golden.json diff --git a/cmd/agent/main.go b/cmd/agent/main.go index 189ed25..2e1e771 100644 --- a/cmd/agent/main.go +++ b/cmd/agent/main.go @@ -220,6 +220,26 @@ func run(ctx context.Context, logger *slog.Logger, clientset kubernetes.Interfac ) disruptionJournal.Observe(d) }) + + // Finished Job runs. Same window shape as the journals above; the facts are + // the run's own instants and outcome, which is what a usage rollup cannot + // supply for a workload that ran for part of a window (ADR 0029). + jobJournal := journal.NewJobRuns(collector.UsageWindowLength) + podWatcher.OnJobFinished(func(r collector.JobRun) { + logger.Info("job run finished", + "namespace", r.Namespace, + "workload_kind", r.Workload.Kind, + "workload_name", r.Workload.Name, + "job", r.Name, + "result", r.Result, + "fail_reason", r.FailReason, + "started_at", r.StartedAt, + "finished_at", r.FinishedAt, + "succeeded", r.Succeeded, + "failed", r.Failed, + ) + jobJournal.Observe(r) + }) podWatcher.SetFilter(filter) nodeWatcher := collector.NewNodeWatcher(clientset, func(n collector.NodeInfo) { logger.Info("node observed", @@ -392,6 +412,21 @@ func run(ctx context.Context, logger *slog.Logger, clientset kubernetes.Interfac logger.Info("pod disruptions flushed", "records", len(records)) } + flushJobRuns := func() { + if spool == nil { + return + } + records := append(jobJournal.CloseBefore(time.Now()), jobJournal.Snapshots()...) + if len(records) == 0 { + return // a cluster with no batch workloads writes nothing + } + if err := spool.WriteJobRuns(records); err != nil { + logger.Error("spooling job runs", "error", err) + return + } + logger.Info("job runs flushed", "records", len(records)) + } + // Excluded pods are reported as aggregate counts only, never by name // (docs/security.md §8). Inventory counters are aggregate too: no identity // of an unjoined fact appears, only a count (CLAUDE.md invariant 6). @@ -403,6 +438,13 @@ func run(ctx context.Context, logger *slog.Logger, clientset kubernetes.Interfac "excluded_namespace_annotation", c.ExcludedNamespaceAnnotation, "excluded_workload_annotation", c.ExcludedWorkloadAnnotation, "excluded_pod_annotation", c.ExcludedPodAnnotation, + // Jobs are counted apart from pods: one Job makes many pods, so a + // shared number would answer neither question (ADR 0029). + "jobs_observed", c.JobsObserved, + "jobs_excluded_namespace_filter", c.JobsExcludedNamespaceFilter, + "jobs_excluded_namespace_annotation", c.JobsExcludedNamespaceAnnotation, + "jobs_excluded_workload_annotation", c.JobsExcludedWorkloadAnnotation, + "jobs_excluded_annotation", c.JobsExcludedAnnotation, // The blind spot, not an exclusion: pods collected without their // workload-level opt-out being checked (ADR 0028). The first is a // standing property of a cluster running operators the agent does @@ -446,6 +488,7 @@ func run(ctx context.Context, logger *slog.Logger, clientset kubernetes.Interfac flushInventory() flushRestarts() flushDisruptions() + flushJobRuns() } } }() @@ -556,6 +599,7 @@ func run(ctx context.Context, logger *slog.Logger, clientset kubernetes.Interfac flushInventory() flushRestarts() flushDisruptions() + flushJobRuns() return errors.Join(errs...) } diff --git a/docs/adr/0029-job-runs.md b/docs/adr/0029-job-runs.md new file mode 100644 index 0000000..341052c --- /dev/null +++ b/docs/adr/0029-job-runs.md @@ -0,0 +1,121 @@ +# 0029. Finished Job runs ship as a windowed journal, and are not baselined + +Date: 2026-08-23 +Status: Accepted + +Adds the payload kind `job_runs`. Nothing existing changes shape; the filter +gains a Job path built on ADR 0028's workload step. + +## Context + +Usage rollups systematically misrepresent short-lived workloads. A Job that runs +for ninety seconds inside an hour-long window reports a tiny average, because the +average is over the window and the workload existed for a fortieth of it. ADR +0013's `covered_nanoseconds` says the coverage was low; it does not say the +workload was a batch run that started, did its work, and finished. + +That gap lands where it costs most. Requests for CronJobs are routinely set from +the peak run and then left, so a batch fleet is where overprovisioning +accumulates quietly — and it is the one workload class the agent measures but +cannot explain. Every fact needed to explain it is already on the Job object, +already watched by an informer the controller runs for owner resolution, and +already covered by RBAC the product has. + +## Decision + +**1. `job_runs` is a windowed journal, one record per finished run.** Kind +`job_runs`, source `journal`, natural key (window start, window length), +superseding — the same shape as `container_restarts` and `pod_disruptions`. + +The window is a delivery boundary, not an aggregation: a run finishes once and +carries its own instants. It exists for ADR 0021's reason, which applies harder +here. A fleet of CronJobs can finish hundreds of runs an hour, and one file per +run would put the spool's file count under the cluster's control. + +The record carries the run's namespace, its workload (the scheduling CronJob, or +the Job itself when it has none), the run's own object name, `started_at` and +`finished_at`, the result, the failure reason, the run's pod success and failure +counts, and the declared `parallelism`, `completions` and `backoffLimit`. + +The pod counts and the declared shape are there because without them the outcome +is unreadable. A run that succeeded with `failed: 2` retried twice, and those +retries consumed resources nothing else in the protocol reports. "Three +succeeded" is a complete run against `completions: 3` and a third of one against +`completions: 9`. + +**2. Failure reasons yes, messages no.** The terminal condition's `reason` is +Kubernetes' own vocabulary, and `BackoffLimitExceeded` and `DeadlineExceeded` are +different facts for a capacity analysis: one is a workload that cannot succeed, +the other one that cannot succeed *in time*, which is often a resource +statement. The adjacent free-text `message` is never read (ADR 0020 §6). + +**3. Runs are not baselined on first observation, and this is the opposite of +ADR 0020 §5.** A Job already finished when the informer syncs is reported, and it +lands in the window where it actually finished. + +ADR 0020 baselines restart counters because a restart counter carries no time: +reporting a pre-existing count would file history in whichever window happens to +be open at startup, dating events by when the agent started. A Job carries +`status.startTime`, `status.completionTime` and its condition's transition time, +so nothing has to be invented — this is ADR 0021's distinction, made there for +disruptions, and it applies here for the same reason. + +The startup burst is bounded by the cluster's own retention: a CronJob keeps +three successful and one failed Job by default, and `ttlSecondsAfterFinished` +removes them sooner where it is set. What the agent gains is the recent history +of a cluster it just started watching, for free and correctly dated. + +**4. A run's Job is filtered on four objects, one more than a pod.** Namespace +allow/deny, the namespace's annotation, the owning CronJob's annotation (ADR +0028's workload step), then the Job's own annotations **or its pod template's**. + +The last of those is the one that matters. Before ADR 0028 the documented way to +opt a workload out was an annotation in its pod template, so a customer who did +that has excluded pods — and if `job_runs` read only the Job object, the agent +would ship facts about a workload whose pods it refuses to measure. Collected and +sent is worse than collected and dropped, and worst of all when the customer +explicitly asked for neither. + +Reading `spec.template.metadata.annotations` is metadata. The `env`, `args` and +`command` beneath it are never read, on a Job no less than on a pod (CLAUDE.md +invariant 4). + +**5. Job counters are kept apart from pod counters.** One Job produces many +pods, so a shared `observed` number would answer neither question. The coverage +report gains `jobs_observed` and four `jobs_excluded_*` counters. Aggregates +only; no Job is named (CLAUDE.md invariant 6). + +**6. No resource envelope in this payload.** Requests and limits belong to +`workload_metadata`, and repeating them here would give one question two answers +that can disagree. + +## Consequences + +**Easier.** A batch workload's cost becomes readable: the run took this long, +succeeded on the third try, and was declared with this parallelism, joined to the +usage window it sat in. `covered_nanoseconds` stops being the only hint that +something short-lived happened. And a cluster that starts the agent gets its +recent batch history immediately rather than after an hour. + +**Harder / given up.** A run whose Job is reaped before the agent sees it is +lost — `ttlSecondsAfterFinished` set aggressively, and an agent restart across +it. This is the property `oom_kill`, `container_restarts` and `pod_disruptions` +already have, and it is why none of them is a billing record. + +A run reported with no resource envelope is possible: if the Job's pods were +reaped before the agent observed them, `workload_metadata` has nothing to join +to. The run's timings and outcome are still true; only the "how much did it ask +for" half is missing, and it is missing because the cluster deleted it. + +Reporting a Job that finished before startup means the first flush after a +restart can write a window that has already closed. The spool's age sweep bounds +how far back that can reach, and the payload is idempotent under its key, so a +rewrite is a rewrite rather than a duplicate. + +**Not changed.** No existing payload's shape, no RBAC, and no new informer — the +Job handler rides the watch already running for owner resolution. A cluster with +no batch workloads writes nothing at all: a window in which no Job finished +produces no file. + +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 7e88083..20e7bda 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -86,3 +86,4 @@ line to paste. - [0026. The controller needs no persistent volume, and therefore no StatefulSet](0026-no-persistent-volume.md) - [0027. No payload carries an ordering field: the spool holds one version of a key](0027-no-payload-ordering-field.md) - [0028. The opt-out annotation works on the workload, and the workload step fails open](0028-workload-level-opt-out.md) +- [0029. Finished Job runs ship as a windowed journal, and are not baselined](0029-job-runs.md) diff --git a/docs/plans/job-and-rollout-facts.md b/docs/plans/job-and-rollout-facts.md index 22f003f..3bb41cb 100644 --- a/docs/plans/job-and-rollout-facts.md +++ b/docs/plans/job-and-rollout-facts.md @@ -65,7 +65,7 @@ everything; C1b and C2 are independent of each other. --- -## Slice C1a — the opt-out annotation works on the workload +## Slice C1a — the opt-out annotation works on the workload ✅ MERGED (#56) Not originally planned. It surfaced while designing C1b's filter: a Job's pods inherit `spec.template.metadata.annotations`, so the documented way to opt a @@ -84,7 +84,7 @@ and `workload_not_cached`. `PodFilter` becomes `Filter`. --- -## Slice C1b — `job_runs` +## Slice C1b — `job_runs` ✅ **Kind:** `job_runs` · **source:** `journal` · **key:** (window start, window length) · **delivery:** supersedes @@ -114,11 +114,15 @@ duplicating them here would give two answers to one question. Consequence to state in the ADR: when a Job's pods are reaped before the agent sees them, the run is reported with no resource envelope to join to. -**Reporting rule — baseline on first observation.** Jobs already in a terminal -state when the informer syncs are recorded as seen and never reported. ADR 0020 -§5 set this: reporting pre-existing history would place it in whichever window -happens to be open at startup, dating events by when the agent started rather -than when they happened. +**Reporting rule — corrected while building: no baseline.** The plan said to +baseline already-terminal Jobs on the ADR 0020 §5 precedent. That precedent does +not transfer. Restarts are baselined because a restart counter carries no time; +a Job carries `startTime`, `completionTime` and its condition's transition time, +so an already-finished run files itself in the window where it actually +finished. This is ADR 0021's distinction, and the agent gains the cluster's +recent batch history at startup instead of discarding it. The startup burst is +bounded by the cluster's own retention — three successful and one failed Job per +CronJob by default. **Honest limitation:** with `ttlSecondsAfterFinished` set, a Job disappears shortly after finishing, so a run completed while the agent was down is lost. diff --git a/docs/security.md b/docs/security.md index eb794b4..c28ab85 100644 --- a/docs/security.md +++ b/docs/security.md @@ -115,7 +115,7 @@ bound to nothing and its token is not mounted — so it appears in no row here. | Resource | Verbs | Why | |---|---|---| | `pods` | get/list/watch | Map containers to workloads; read `requests`/`limits` from spec; detect OOM kills and count restarts from `status.containerStatuses[]` (`restartCount`, and the reason and exit code of `lastState.terminated`); read the `PodScheduled` and `DisruptionTarget` conditions to report why a replica is not running and when the cluster took one away; read `containerPorts` to locate pprof endpoints | -| `replicasets`, `deployments`, `statefulsets`, `daemonsets`, `jobs`, `cronjobs` | get/list/watch | Resolve the `ownerReferences` chain (Pod → ReplicaSet → Deployment) so findings are aggregated per workload, not per pod | +| `replicasets`, `deployments`, `statefulsets`, `daemonsets`, `jobs`, `cronjobs` | get/list/watch | Resolve the `ownerReferences` chain (Pod → ReplicaSet → Deployment) so findings are aggregated per workload, not per pod; read each workload's own annotations to honor an opt-out written there ([ADR 0028](adr/0028-workload-level-opt-out.md)); and read finished Jobs' timings and outcomes for `job_runs` ([ADR 0029](adr/0029-job-runs.md)) | | `nodes` | get/list/watch | `allocatable`/`capacity` for node idle computation; labels `node.kubernetes.io/instance-type` and capacity-type (spot/on-demand) for the cost model; `status.nodeInfo.kernelVersion` to report whether nodes meet the kernel floor for the eBPF profile (`CAP_BPF` requires kernel 5.8+, see [§7](#7-node-privileges)) | | `namespaces` | get/list/watch | Evaluate namespace allow/deny filters and the opt-out annotation | | `nodes/proxy` | get | Poll each kubelet's `/stats/summary` and `/metrics/cadvisor` through the API server for usage counters: CPU, memory working set, CFS throttling, PSI where exposed (ADR 0006). **Honest disclosure:** this verb technically permits any kubelet GET endpoint through the API server, including node logs. The agent calls exactly the two stats paths above — auditable, since all kubelet access lives in a single poll loop | @@ -379,6 +379,7 @@ appearing here (ADR 0022). | `oom_kill` | One out-of-memory kill: namespace, pod, container, workload, when it finished, exit code, restart count, and the memory limit in force | **yes** | | `container_restarts` | Per (namespace, pod, container) and hour: restart count, breakdown by termination reason, how many restarts had no reason visible, last exit code | **yes** | | `pod_disruptions` | Per hour: the pods the *cluster* removed — preempted, evicted under node pressure, drained, or removed via the eviction API — with the node and the instant | **yes** | +| `job_runs` | Per hour: each Job that finished — when it started and finished, whether it succeeded, the failure reason, its pod success/failure counts, and its declared `parallelism`/`completions`/`backoffLimit` | **yes** | | `workload_metadata` | Declared shape per (namespace, workload, container, image digest): image, requests, limits, ports, QoS, replica counts by phase, by node, and by unscheduled reason | no | | `node_metadata` | Per node: name, size, instance and capacity type, zone, region, kernel version, CPU architecture | no | | `go_inventory` | Per (namespace, workload, container): Go version, main module, image digest, PGO flag, plus a fleet-coverage block | no | @@ -395,6 +396,15 @@ section describes each in detail. ### Field minimization - The agent **never reads Secrets or ConfigMaps** (no RBAC for them exists). + Its own configuration is a different thing and arrives as a mounted file, not + through the API. +- From Jobs, the agent keeps `metadata`, the terminal condition's *type*, + *reason* and *transition time*, `status.startTime`, `status.completionTime`, + `status.succeeded`, `status.failed`, and three declared fields: + `spec.parallelism`, `spec.completions`, `spec.backoffLimit`. From + `spec.template` it reads **annotations only**, to honor an opt-out written + there — never the `env`, `args` or `command` beneath it, on a Job no less than + on a pod. - From pod specs, the agent keeps only `metadata`, `spec.containers[].resources`, `spec.containers[].ports`, `ownerReferences`, and `status`. It explicitly discards `env`, `args`, and `command` before anything is stored or @@ -448,6 +458,13 @@ cluster, so that a stale snapshot can be recognized rather than assumed current ### Object history, and the one place pods are named (ADR 0020) +`job_runs` names the Job object, which for a scheduled job is a generated +per-run name such as `rollup-29123456`, and the CronJob that scheduled it. The +name is what tells two runs of one schedule apart, and it is a name your cluster +generated, never workload content — the same line `container_restarts` draws +([ADR 0029](adr/0029-job-runs.md)). A run whose Job, CronJob, namespace or pod +template carries the opt-out annotation produces no record at all. + `oom_kill`, `container_restarts` and `pod_disruptions` report what the cluster's own object status records about a container's history — see the table above for what each carries. All three are `journal` provenance, and all three name the diff --git a/internal/collector/filter.go b/internal/collector/filter.go index 6ccb450..df74e9d 100644 --- a/internal/collector/filter.go +++ b/internal/collector/filter.go @@ -4,6 +4,7 @@ import ( "strings" "sync/atomic" + batchv1 "k8s.io/api/batch/v1" corev1 "k8s.io/api/core/v1" ) @@ -32,6 +33,12 @@ const ( ExcludedByNamespaceAnnotation ExclusionReason = "namespace_annotation" ExcludedByWorkloadAnnotation ExclusionReason = "workload_annotation" ExcludedByPodAnnotation ExclusionReason = "pod_annotation" + // ExcludedByObjectAnnotation is the object's own annotation where that + // object is not a pod — today a Job, whose own metadata is where an + // annotation on a CronJob's `jobTemplate.metadata` lands, and whose pod + // template is where an opt-out written the pre-ADR-0028 way lives. The two + // are one reason because they mean one thing: this run was opted out. + ExcludedByObjectAnnotation ExclusionReason = "object_annotation" ) // UnresolvedWorkload says why a pod's workload annotations could not be read. @@ -75,6 +82,15 @@ type Filter struct { workloadUnknownKind atomic.Int64 workloadNotCached atomic.Int64 + + // Jobs are counted apart from pods. Folding them together would make + // `pods_observed` state a falsehood about pods, and the coverage report is + // the one place the customer is told what was not collected. + jobsObserved atomic.Int64 + jobsExcludedNamespaceFilter atomic.Int64 + jobsExcludedNamespaceAnnotation atomic.Int64 + jobsExcludedWorkloadAnnotation atomic.Int64 + jobsExcludedAnnotation atomic.Int64 } // NewFilter builds a filter from namespace name lists; entries may use @@ -111,6 +127,35 @@ func (f *Filter) AdmitPod(pod *corev1.Pod, nsAnnotations map[string]string, work return true, "" } +// AdmitJob reports whether a finished Job's facts may be collected. +// +// It is the pod decision with one step added. The workload step is the Job's +// owning CronJob; the object step is the Job's own annotations; and the last +// step reads the Job's **pod template**, which is what excluded this run's pods +// if the customer opted out the way security.md documented before ADR 0028. +// +// Without that last step the agent would ship `job_runs` for a workload whose +// pods it refuses to measure — collected and sent, which is worse than +// collected and dropped. Reading `spec.template.metadata.annotations` is +// metadata, never the `env`, `args` or `command` beneath it (CLAUDE.md +// invariant 4). +func (f *Filter) AdmitJob(job *batchv1.Job, nsAnnotations map[string]string, workload WorkloadLookup) (bool, ExclusionReason) { + if !f.namespaceAllowed(job.Namespace) { + return false, ExcludedByNamespaceFilter + } + if nsAnnotations[CollectAnnotation] == "false" { + return false, ExcludedByNamespaceAnnotation + } + if workload.Annotations[CollectAnnotation] == "false" { + return false, ExcludedByWorkloadAnnotation + } + if job.Annotations[CollectAnnotation] == "false" || + job.Spec.Template.Annotations[CollectAnnotation] == "false" { + return false, ExcludedByObjectAnnotation + } + return true, "" +} + func (f *Filter) namespaceAllowed(namespace string) bool { if len(f.allow) > 0 && !matchAny(f.allow, namespace) { return false @@ -135,6 +180,23 @@ func (f *Filter) countExcluded(reason ExclusionReason) { } } +func (f *Filter) countJobObserved() { + f.jobsObserved.Add(1) +} + +func (f *Filter) countJobExcluded(reason ExclusionReason) { + switch reason { + case ExcludedByNamespaceFilter: + f.jobsExcludedNamespaceFilter.Add(1) + case ExcludedByNamespaceAnnotation: + f.jobsExcludedNamespaceAnnotation.Add(1) + case ExcludedByWorkloadAnnotation: + f.jobsExcludedWorkloadAnnotation.Add(1) + case ExcludedByObjectAnnotation: + f.jobsExcludedAnnotation.Add(1) + } +} + // countUnresolvedWorkload records that one pod was admitted without its // workload-level opt-out being checked. It is counted on every admission // decision, not only on exclusions: the number is the size of the blind spot, @@ -165,6 +227,14 @@ type Coverage struct { // second should sit at zero. WorkloadUnknownKind int64 WorkloadNotCached int64 + + // Job counters, kept apart from the pod ones on purpose: one Job produces + // many pods, so a shared counter would answer neither question. + JobsObserved int64 + JobsExcludedNamespaceFilter int64 + JobsExcludedNamespaceAnnotation int64 + JobsExcludedWorkloadAnnotation int64 + JobsExcludedAnnotation int64 } // Snapshot returns the current coverage counters. @@ -177,6 +247,12 @@ func (f *Filter) Snapshot() Coverage { ExcludedPodAnnotation: f.excludedPodAnnotation.Load(), WorkloadUnknownKind: f.workloadUnknownKind.Load(), WorkloadNotCached: f.workloadNotCached.Load(), + + JobsObserved: f.jobsObserved.Load(), + JobsExcludedNamespaceFilter: f.jobsExcludedNamespaceFilter.Load(), + JobsExcludedNamespaceAnnotation: f.jobsExcludedNamespaceAnnotation.Load(), + JobsExcludedWorkloadAnnotation: f.jobsExcludedWorkloadAnnotation.Load(), + JobsExcludedAnnotation: f.jobsExcludedAnnotation.Load(), } } diff --git a/internal/collector/jobs.go b/internal/collector/jobs.go new file mode 100644 index 0000000..2554c78 --- /dev/null +++ b/internal/collector/jobs.go @@ -0,0 +1,201 @@ +package collector + +import ( + "time" + + batchv1 "k8s.io/api/batch/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +// JobRun is one finished execution of a Job: when it ran, how it ended, and the +// shape it was declared with. It is a journal fact — the object's status records +// that it happened (ADR 0029). +// +// It exists because usage rollups systematically misrepresent short-lived +// workloads. A Job that runs ninety seconds inside an hour-long window reports a +// tiny average, and `covered_nanoseconds` says the coverage was low without +// saying why. This is the denominator: the run happened, it took this long, and +// it finished. +type JobRun struct { + Namespace string + // Workload is the CronJob that scheduled this run, or the Job itself when + // it has no controller. Many runs of one schedule therefore aggregate under + // one workload, exactly as replicas of a Deployment do. + Workload WorkloadRef + // Name is this run's own object name. For a CronJob it is the generated + // per-run name, which is what distinguishes two runs of one schedule. + Name string + StartedAt time.Time + FinishedAt time.Time + Result string + FailReason string + Succeeded int32 + Failed int32 + Parallelism *int32 + Completions *int32 + BackoffLimit *int32 +} + +// Job outcomes. The vocabulary is Kubernetes' own condition types, not ours. +const ( + JobSucceeded = "succeeded" + JobFailed = "failed" +) + +// OnJobFinished registers fn to be called once per finished Job. Must be called +// before Run. fn is called from the informer goroutine and must not block. +func (w *PodWatcher) OnJobFinished(fn func(JobRun)) { + w.onJobFinished = fn +} + +// reportJobRun reports a Job that has reached a terminal state, exactly once. +// +// Unlike a container restart, a Job is timestamped by the cluster, so a run +// already finished when the agent starts lands in the window where it actually +// happened. That is why there is no baseline pass here and ADR 0020 §5 does not +// transfer: the reason restarts are baselined is that a restart counter carries +// no time, and this object does (ADR 0021 made the same distinction). +func (w *PodWatcher) reportJobRun(job *batchv1.Job) { + if w.onJobFinished == nil { + return + } + result, reason, finishedAt, ok := jobOutcomeOf(job) + if !ok { + return // still running, or never started + } + if !w.admitJob(job, true) { + return + } + w.mu.Lock() + _, duplicate := w.reportedJobs[job.UID] + if !duplicate { + w.reportedJobs[job.UID] = struct{}{} + } + w.mu.Unlock() + if duplicate { + return + } + + run := JobRun{ + Namespace: job.Namespace, + Workload: w.resolveJobWorkload(job), + Name: job.Name, + FinishedAt: finishedAt, + Result: result, + FailReason: reason, + Succeeded: job.Status.Succeeded, + Failed: job.Status.Failed, + Parallelism: job.Spec.Parallelism, + Completions: job.Spec.Completions, + BackoffLimit: job.Spec.BackoffLimit, + } + if job.Status.StartTime != nil { + run.StartedAt = job.Status.StartTime.UTC() + } + w.onJobFinished(run) +} + +// forgetJobRun drops the dedup entry of a deleted Job. Losing the map on +// restart is harmless while the object still exists: the run is reported once +// more, with the same instants, into the same window — a byte-identical rewrite +// of that window's payload. +func (w *PodWatcher) forgetJobRun(job *batchv1.Job) { + w.mu.Lock() + defer w.mu.Unlock() + delete(w.reportedJobs, job.UID) +} + +// resolveJobWorkload is the Job's controller when it has one, and the Job +// itself when it does not. It stops at one hop: a CronJob is the top of this +// chain by construction. +func (w *PodWatcher) resolveJobWorkload(job *batchv1.Job) WorkloadRef { + if owner := metav1.GetControllerOf(job); owner != nil { + return WorkloadRef{Kind: owner.Kind, Name: owner.Name} + } + return WorkloadRef{Kind: "Job", Name: job.Name} +} + +// admitJob runs a Job through the filter. The workload step is its owning +// CronJob; the object step is the Job's own annotations, which is where an +// annotation on a CronJob's `jobTemplate.metadata` lands; and the pod-template +// step is what excluded the run's pods. +// +// That last step is what keeps this payload honest. A customer who opted out +// the way security.md documented before ADR 0028 wrote the annotation into the +// pod template. Their pods are excluded, and if `job_runs` ignored the template +// the agent would ship facts about a workload the customer refused — collected +// and sent, which is worse than collected and dropped. +func (w *PodWatcher) admitJob(job *batchv1.Job, count bool) bool { + var nsAnnotations map[string]string + if ns, err := w.nsLister.Get(job.Namespace); err == nil { + nsAnnotations = ns.Annotations + } + workload := w.jobWorkloadAnnotations(job) + allowed, reason := w.filter.AdmitJob(job, nsAnnotations, workload) + if count { + if allowed { + w.filter.countJobObserved() + } else { + w.filter.countJobExcluded(reason) + } + if workload.Unresolved != "" { + w.filter.countUnresolvedWorkload(workload.Unresolved) + } + } + return allowed +} + +// jobWorkloadAnnotations reads the annotations of the CronJob that scheduled +// this Job. A Job with no controller is its own workload and has nothing above +// it to consult, which is resolved rather than unresolved; a Job owned by +// something that is not a CronJob is a kind this agent does not read (ADR 0028). +func (w *PodWatcher) jobWorkloadAnnotations(job *batchv1.Job) WorkloadLookup { + owner := metav1.GetControllerOf(job) + if owner == nil { + return WorkloadLookup{} + } + if owner.Kind != "CronJob" { + return WorkloadLookup{Unresolved: WorkloadKindUnknown} + } + cron, err := w.cronLister.CronJobs(job.Namespace).Get(owner.Name) + if err != nil { + return WorkloadLookup{Unresolved: WorkloadNotCached} + } + return WorkloadLookup{Annotations: cron.Annotations} +} + +// jobOutcomeOf reports whether the Job has finished, how, and when. +// +// Success is timestamped by `status.completionTime`, which Kubernetes sets only +// on success. A failure has no such field, so the terminal condition's +// transition time is the instant — the same fallback ADR 0021 uses for a +// disruption whose condition carries the only timestamp there is. +func jobOutcomeOf(job *batchv1.Job) (result, reason string, finishedAt time.Time, ok bool) { + for _, cond := range job.Status.Conditions { + if cond.Status != corev1.ConditionTrue { + continue + } + switch cond.Type { + case batchv1.JobComplete: + at := job.Status.CompletionTime + if at != nil && !at.IsZero() { + return JobSucceeded, "", at.UTC(), true + } + return JobSucceeded, "", conditionTime(cond), true + case batchv1.JobFailed: + return JobFailed, cond.Reason, conditionTime(cond), true + } + } + return "", "", time.Time{}, false +} + +// conditionTime is the condition's transition time, falling back to now when +// the API server recorded none. A zero instant would file the run in the epoch +// window, which is worse than filing it slightly late. +func conditionTime(cond batchv1.JobCondition) time.Time { + if cond.LastTransitionTime.IsZero() { + return time.Now().UTC() + } + return cond.LastTransitionTime.UTC() +} diff --git a/internal/collector/jobs_test.go b/internal/collector/jobs_test.go new file mode 100644 index 0000000..e2fb916 --- /dev/null +++ b/internal/collector/jobs_test.go @@ -0,0 +1,190 @@ +package collector + +import ( + "context" + "sync" + "testing" + "time" + + batchv1 "k8s.io/api/batch/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/kubernetes/fake" + "k8s.io/utils/ptr" +) + +var jobFinished = time.Date(2026, 8, 6, 10, 7, 0, 0, time.UTC) + +func finishedJob(name string, condType batchv1.JobConditionType, reason string, owner *metav1.OwnerReference) *batchv1.Job { + job := &batchv1.Job{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "analytics", Name: name, UID: types.UID("uid-" + name), + }, + Spec: batchv1.JobSpec{ + Parallelism: ptr.To(int32(1)), + Completions: ptr.To(int32(1)), + BackoffLimit: ptr.To(int32(6)), + }, + Status: batchv1.JobStatus{ + StartTime: &metav1.Time{Time: jobFinished.Add(-2 * time.Minute)}, + Succeeded: 1, + Failed: 2, + Conditions: []batchv1.JobCondition{{ + Type: condType, + Status: corev1.ConditionTrue, + Reason: reason, + LastTransitionTime: metav1.Time{Time: jobFinished}, + }}, + }, + } + if condType == batchv1.JobComplete { + job.Status.CompletionTime = &metav1.Time{Time: jobFinished} + } + if owner != nil { + job.OwnerReferences = []metav1.OwnerReference{*owner} + } + return job +} + +func collectRuns(t *testing.T, filter *Filter, objects ...runtime.Object) map[string]JobRun { + t.Helper() + clientset := fake.NewClientset(objects...) + + var mu sync.Mutex + runs := map[string]JobRun{} + watcher := NewPodWatcher(clientset, func(PodInfo) {}) + watcher.SetFilter(filter) + watcher.OnJobFinished(func(r JobRun) { + mu.Lock() + runs[r.Name] = r + mu.Unlock() + }) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { done <- watcher.Run(ctx) }() + + deadline := time.Now().Add(5 * time.Second) + for { + c := filter.Snapshot() + if c.JobsObserved+c.JobsExcludedNamespaceFilter+c.JobsExcludedNamespaceAnnotation+ + c.JobsExcludedWorkloadAnnotation+c.JobsExcludedAnnotation > 0 || time.Now().After(deadline) { + break + } + time.Sleep(20 * time.Millisecond) + } + // Give the remaining Adds a moment to drain before reading. + time.Sleep(200 * time.Millisecond) + cancel() + if err := <-done; err != nil { + t.Fatalf("watcher returned error: %v", err) + } + + mu.Lock() + defer mu.Unlock() + out := make(map[string]JobRun, len(runs)) + for k, v := range runs { + out[k] = v + } + return out +} + +// A finished Job becomes a run with the cluster's own instants and outcome, and +// it is reported even though it was already terminal when the informer synced: +// unlike a restart counter, a Job carries its own time (ADR 0029 §3). +func TestFinishedJobIsReportedWithItsOwnInstants(t *testing.T) { + cron := &batchv1.CronJob{ObjectMeta: metav1.ObjectMeta{Namespace: "analytics", Name: "rollup"}} + owner := controllerRef("CronJob", "rollup") + runs := collectRuns(t, NewFilter(nil, nil), + cron, + finishedJob("rollup-1", batchv1.JobComplete, "", &owner), + finishedJob("rollup-2", batchv1.JobFailed, "BackoffLimitExceeded", &owner), + ) + + ok, found := runs["rollup-1"] + if !found { + t.Fatalf("the succeeded run was not reported; got %v", keysOf(runs)) + } + if ok.Result != JobSucceeded || ok.FailReason != "" { + t.Errorf("run = (%q, %q), want (%q, \"\")", ok.Result, ok.FailReason, JobSucceeded) + } + if !ok.FinishedAt.Equal(jobFinished) { + t.Errorf("finished_at = %s, want %s (the cluster's instant, not the observation's)", + ok.FinishedAt, jobFinished) + } + if ok.Workload.Kind != "CronJob" || ok.Workload.Name != "rollup" { + t.Errorf("workload = %+v, want the scheduling CronJob", ok.Workload) + } + if ok.Succeeded != 1 || ok.Failed != 2 { + t.Errorf("pod counts = (%d, %d), want (1, 2); the retries are what cost resources", + ok.Succeeded, ok.Failed) + } + + failed, found := runs["rollup-2"] + if !found { + t.Fatalf("the failed run was not reported; got %v", keysOf(runs)) + } + if failed.Result != JobFailed || failed.FailReason != "BackoffLimitExceeded" { + t.Errorf("failed run = (%q, %q), want (%q, %q)", + failed.Result, failed.FailReason, JobFailed, "BackoffLimitExceeded") + } +} + +// A running Job is not a run. +func TestUnfinishedJobIsNotReported(t *testing.T) { + running := &batchv1.Job{ObjectMeta: metav1.ObjectMeta{ + Namespace: "analytics", Name: "rollup-live", UID: "uid-live", + }} + filter := NewFilter(nil, nil) + runs := collectRuns(t, filter, running) + if len(runs) != 0 { + t.Errorf("reported %v for a Job with no terminal condition", keysOf(runs)) + } +} + +// The opt-out written the pre-ADR-0028 way — in the Job's pod template — is what +// excluded this run's pods. If job_runs ignored it the agent would ship facts +// about a workload whose pods it refuses to measure (ADR 0029 §4). +func TestPodTemplateOptOutExcludesTheRun(t *testing.T) { + job := finishedJob("legacy-optout", batchv1.JobComplete, "", nil) + job.Spec.Template.Annotations = map[string]string{CollectAnnotation: "false"} + + filter := NewFilter(nil, nil) + runs := collectRuns(t, filter, job) + if len(runs) != 0 { + t.Errorf("reported %v for a run whose pod template opted out", keysOf(runs)) + } + if got := filter.Snapshot(); got.JobsExcludedAnnotation != 1 { + t.Errorf("jobs_excluded_annotation = %d, want 1", got.JobsExcludedAnnotation) + } +} + +// And the opt-out on the scheduling CronJob, which is ADR 0028's step applied to +// the Job path. +func TestCronJobOptOutExcludesItsRuns(t *testing.T) { + cron := &batchv1.CronJob{ObjectMeta: metav1.ObjectMeta{ + Namespace: "analytics", Name: "rollup", + Annotations: map[string]string{CollectAnnotation: "false"}, + }} + owner := controllerRef("CronJob", "rollup") + + filter := NewFilter(nil, nil) + runs := collectRuns(t, filter, cron, finishedJob("rollup-1", batchv1.JobComplete, "", &owner)) + if len(runs) != 0 { + t.Errorf("reported %v for a run of an opted-out CronJob", keysOf(runs)) + } + if got := filter.Snapshot(); got.JobsExcludedWorkloadAnnotation != 1 { + t.Errorf("jobs_excluded_workload_annotation = %d, want 1", got.JobsExcludedWorkloadAnnotation) + } +} + +func keysOf(runs map[string]JobRun) []string { + out := make([]string, 0, len(runs)) + for k := range runs { + out = append(out, k) + } + return out +} diff --git a/internal/collector/pods.go b/internal/collector/pods.go index c9ed847..e180698 100644 --- a/internal/collector/pods.go +++ b/internal/collector/pods.go @@ -9,6 +9,7 @@ import ( "strings" "sync" + batchv1 "k8s.io/api/batch/v1" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" @@ -85,12 +86,13 @@ type PodInfo struct { // new ones. Every observed pod is reported through the OnPod callback, // resolved to the workload that manages it. type PodWatcher struct { - clientset kubernetes.Interface - onPod func(PodInfo) - onOOM func(OOMKill) - onRestart func(ContainerRestart) - onDisruption func(PodDisruption) - filter *Filter + clientset kubernetes.Interface + onPod func(PodInfo) + onOOM func(OOMKill) + onRestart func(ContainerRestart) + onDisruption func(PodDisruption) + onJobFinished func(JobRun) + filter *Filter // Listers for the owner chain. The ReplicaSet and Job listers resolve a // pod to its top-level workload; the four below read that workload's own @@ -112,6 +114,10 @@ type PodWatcher struct { // reportedDisruptions deduplicates disrupted pods across the many status // updates one receives between being condemned and disappearing. reportedDisruptions map[string]struct{} + // reportedJobs deduplicates finished Jobs across the status updates a Job + // receives after it terminates. Keyed by UID rather than name because a + // CronJob's run names are generated but a bare Job's name is reusable. + reportedJobs map[types.UID]struct{} // reportedSig deduplicates pod reports across the many status updates a // pod receives: the collected view is re-sent only when its image-digest // signature changes (digests appear on the update that follows container @@ -163,6 +169,7 @@ func NewPodWatcher(clientset kubernetes.Interface, onPod func(PodInfo)) *PodWatc reportedOOMs: make(map[string]struct{}), restartCounts: make(map[string]int32), reportedDisruptions: make(map[string]struct{}), + reportedJobs: make(map[types.UID]struct{}), reportedSig: make(map[types.UID]string), index: make(map[types.UID]podIndexEntry), nameIndex: make(map[string]types.UID), @@ -202,9 +209,10 @@ func (w *PodWatcher) Run(ctx context.Context) error { // Informers must be instantiated before Start, or the factory won't // run them. podsInformer := pods.Informer() + jobsInformer := jobs.Informer() ownerSynced := []cache.InformerSynced{ rs.Informer().HasSynced, - jobs.Informer().HasSynced, + jobsInformer.HasSynced, deployments.Informer().HasSynced, statefulSets.Informer().HasSynced, daemonSets.Informer().HasSynced, @@ -274,6 +282,39 @@ func (w *PodWatcher) Run(ctx context.Context) error { } defer func() { _ = podsInformer.RemoveEventHandler(reg) }() + // The Job handler rides the informer that already resolves owner chains, + // so `job_runs` costs no watch the controller was not already running + // (ADR 0029). Like the pod handler it is registered after sync and replays + // the cache as Adds — which is wanted here: a Job carries its own + // timestamps, so a run that finished before the agent started files itself + // in the window where it actually happened. + jobReg, err := jobsInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{ + AddFunc: func(obj any) { + if job, ok := obj.(*batchv1.Job); ok { + w.reportJobRun(job) + } + }, + // A Job reaches its terminal condition through an update, which is the + // path that reports almost every run of a running cluster. + UpdateFunc: func(_, obj any) { + if job, ok := obj.(*batchv1.Job); ok { + w.reportJobRun(job) + } + }, + DeleteFunc: func(obj any) { + if unknown, ok := obj.(cache.DeletedFinalStateUnknown); ok { + obj = unknown.Obj + } + if job, ok := obj.(*batchv1.Job); ok { + w.forgetJobRun(job) + } + }, + }) + if err != nil { + return fmt.Errorf("register job handler: %w", err) + } + defer func() { _ = jobsInformer.RemoveEventHandler(jobReg) }() + <-ctx.Done() return nil } diff --git a/internal/journal/jobs.go b/internal/journal/jobs.go new file mode 100644 index 0000000..e4897f0 --- /dev/null +++ b/internal/journal/jobs.go @@ -0,0 +1,155 @@ +package journal + +import ( + "sort" + "sync" + "time" + + "github.com/RebuildStackCo/runtime-agent/internal/collector" +) + +// JobRunRecord is one finished Job run, in the window where it finished. +// +// Like a disruption and unlike a restart, a run is timestamped by the cluster, +// so the record carries the instants themselves and the window is only a +// delivery boundary (ADR 0021, ADR 0029). The pair of instants is the point of +// the payload: it is the denominator a usage rollup cannot supply for a +// workload that ran for ninety seconds of an hour-long window. +type JobRunRecord struct { + Namespace string `json:"namespace"` + Workload collector.WorkloadRef `json:"workload"` + // Name is the run's own object name, which for a CronJob is generated per + // run and is what tells two runs of one schedule apart. + Name string `json:"name"` + // StartedAt is zero for a Job that failed before the controller recorded a + // start. A zero instant is a real state — declared and never started — and + // is omitted rather than rendered as the epoch. + StartedAt time.Time `json:"started_at,omitzero"` + FinishedAt time.Time `json:"finished_at"` + // Result is "succeeded" or "failed", from the Job's terminal condition. + Result string `json:"result"` + // FailReason is the terminal condition's reason — `BackoffLimitExceeded` + // and `DeadlineExceeded` are different capacity facts. Kubernetes' + // vocabulary, never the free-text message beside it (ADR 0020 §6). + FailReason string `json:"fail_reason,omitempty"` + // Succeeded and Failed count the run's own pods. A run that succeeded with + // Failed > 0 retried, and the retries consumed resources nothing else + // reports. + Succeeded int32 `json:"succeeded"` + Failed int32 `json:"failed"` + // The declared shape. Without it "3 succeeded" cannot be read as a complete + // run or a partial one. Nil means the field was unset and Kubernetes' + // default applies — a fact, distinct from an explicit value. + Parallelism *int32 `json:"parallelism,omitempty"` + Completions *int32 `json:"completions,omitempty"` + BackoffLimit *int32 `json:"backoff_limit,omitempty"` + WindowStart time.Time `json:"window_start"` + WindowSeconds int64 `json:"window_seconds"` +} + +type jobRunKey struct { + namespace string + name string + start int64 +} + +// JobRuns collects finished Job runs into wall-clock-aligned windows. It is +// safe for concurrent use: observations arrive on the informer goroutine while +// the flush goroutine reads snapshots. +// +// Like Disruptions it accumulates records rather than counters — a run finishes +// once and there is nothing to add up. The window bounds the payload's file +// count, which matters here because a fleet of CronJobs can finish hundreds of +// runs an hour and one file per run would put the spool's file count under the +// cluster's control (ADR 0021's reasoning, same shape). +type JobRuns struct { + mu sync.Mutex + windowLength time.Duration + open map[jobRunKey]JobRunRecord +} + +// NewJobRuns returns an accumulator over windows of the given length, which +// must be positive. +func NewJobRuns(windowLength time.Duration) *JobRuns { + if windowLength <= 0 { + panic("journal: window length must be positive") + } + return &JobRuns{windowLength: windowLength, open: map[jobRunKey]JobRunRecord{}} +} + +// Observe files one run in the window holding the instant it finished. +// +// Keying by (namespace, name, window) makes this idempotent: the same run +// reported twice — after an agent restart, while the object still exists — +// rewrites its own record rather than appearing twice. That is what keeps the +// collector's in-memory dedup loss-harmless. +func (j *JobRuns) Observe(run collector.JobRun) { + if run.FinishedAt.IsZero() { + return + } + j.mu.Lock() + defer j.mu.Unlock() + + start := run.FinishedAt.UTC().Truncate(j.windowLength) + j.open[jobRunKey{namespace: run.Namespace, name: run.Name, start: start.UnixNano()}] = JobRunRecord{ + Namespace: run.Namespace, + Workload: run.Workload, + Name: run.Name, + StartedAt: run.StartedAt.UTC(), + FinishedAt: run.FinishedAt.UTC(), + Result: run.Result, + FailReason: run.FailReason, + Succeeded: run.Succeeded, + Failed: run.Failed, + Parallelism: run.Parallelism, + Completions: run.Completions, + BackoffLimit: run.BackoffLimit, + WindowStart: start, + WindowSeconds: int64(j.windowLength / time.Second), + } +} + +// Snapshots returns every open record, sorted by window then namespace then +// name so the payload bytes are deterministic (the golden contract). +func (j *JobRuns) Snapshots() []JobRunRecord { + j.mu.Lock() + defer j.mu.Unlock() + out := make([]JobRunRecord, 0, len(j.open)) + for _, rec := range j.open { + out = append(out, rec) + } + sortJobRuns(out) + return out +} + +// CloseBefore removes and returns every record whose window ended at or before +// now. Removal bounds memory; the records are returned so the caller writes them +// one last time, with the same loss bound the usage accumulator accepts +// (ADR 0007). +func (j *JobRuns) CloseBefore(now time.Time) []JobRunRecord { + j.mu.Lock() + defer j.mu.Unlock() + var out []JobRunRecord + for key, rec := range j.open { + if rec.WindowStart.Add(j.windowLength).After(now) { + continue + } + out = append(out, rec) + delete(j.open, key) + } + sortJobRuns(out) + return out +} + +func sortJobRuns(records []JobRunRecord) { + sort.Slice(records, func(i, k int) bool { + a, b := records[i], records[k] + if !a.WindowStart.Equal(b.WindowStart) { + return a.WindowStart.Before(b.WindowStart) + } + if a.Namespace != b.Namespace { + return a.Namespace < b.Namespace + } + return a.Name < b.Name + }) +} diff --git a/internal/journal/jobs_test.go b/internal/journal/jobs_test.go new file mode 100644 index 0000000..11f69d8 --- /dev/null +++ b/internal/journal/jobs_test.go @@ -0,0 +1,95 @@ +package journal + +import ( + "testing" + "time" + + "github.com/RebuildStackCo/runtime-agent/internal/collector" +) + +var jobWindow = time.Date(2026, 8, 6, 10, 0, 0, 0, time.UTC) + +func run(name string, finishedAt time.Time) collector.JobRun { + return collector.JobRun{ + Namespace: "analytics", + Workload: collector.WorkloadRef{Kind: "CronJob", Name: "rollup"}, + Name: name, + StartedAt: finishedAt.Add(-time.Minute), + FinishedAt: finishedAt, + Result: collector.JobSucceeded, + Succeeded: 1, + } +} + +// A run files itself in the window holding the instant it finished, not the +// window that happens to be open when the agent observes it. That is what lets +// a run that finished before startup be reported at all (ADR 0029 §3). +func TestRunFilesItselfByWhenItFinished(t *testing.T) { + j := NewJobRuns(time.Hour) + j.Observe(run("rollup-1", jobWindow.Add(20*time.Minute))) + j.Observe(run("rollup-2", jobWindow.Add(-3*time.Hour))) + + got := j.Snapshots() + if len(got) != 2 { + t.Fatalf("got %d records, want 2", len(got)) + } + if !got[0].WindowStart.Equal(jobWindow.Add(-3 * time.Hour)) { + t.Errorf("the older run landed in window %s, want its own", got[0].WindowStart) + } + if !got[1].WindowStart.Equal(jobWindow) { + t.Errorf("the recent run landed in window %s, want %s", got[1].WindowStart, jobWindow) + } +} + +// Re-observing a run rewrites its record rather than adding one. This is what +// makes the collector's in-memory dedup loss-harmless: after a restart the same +// Job object is reported again, with the same instants, into the same window. +func TestReobservingARunRewritesIt(t *testing.T) { + j := NewJobRuns(time.Hour) + first := run("rollup-1", jobWindow.Add(20*time.Minute)) + j.Observe(first) + + grown := first + grown.Failed = 3 + j.Observe(grown) + + got := j.Snapshots() + if len(got) != 1 { + t.Fatalf("got %d records after re-observing one run, want 1", len(got)) + } + if got[0].Failed != 3 { + t.Errorf("surviving record has failed=%d, want 3 (newest rewrites)", got[0].Failed) + } +} + +// A run with no finish instant is not a run. Filing it would put it in the epoch +// window, which is a fact about nothing. +func TestUnfinishedRunIsNotFiled(t *testing.T) { + j := NewJobRuns(time.Hour) + unfinished := run("rollup-1", time.Time{}) + j.Observe(unfinished) + if got := j.Snapshots(); len(got) != 0 { + t.Fatalf("filed %d records for a run that never finished, want 0", len(got)) + } +} + +// CloseBefore hands back ended windows and forgets them, so memory is bounded by +// the open window rather than by how many runs the cluster has ever finished. +func TestJobRunsCloseBeforeReturnsAndForgetsEndedWindows(t *testing.T) { + j := NewJobRuns(time.Hour) + j.Observe(run("old", jobWindow.Add(10*time.Minute))) + j.Observe(run("current", jobWindow.Add(90*time.Minute))) + + closed := j.CloseBefore(jobWindow.Add(2 * time.Hour)) + if len(closed) != 2 { + t.Fatalf("closed %d records, want 2 (both windows ended)", len(closed)) + } + if got := j.Snapshots(); len(got) != 0 { + t.Errorf("accumulator still holds %d records after closing every window", len(got)) + } + + j.Observe(run("current", jobWindow.Add(90*time.Minute))) + if closed := j.CloseBefore(jobWindow.Add(90 * time.Minute)); len(closed) != 0 { + t.Errorf("closed %d records of a window that has not ended", len(closed)) + } +} diff --git a/internal/sink/registry.go b/internal/sink/registry.go index 4457c0d..98c20d7 100644 --- a/internal/sink/registry.go +++ b/internal/sink/registry.go @@ -97,6 +97,16 @@ var registry = []PayloadKind{ Delivery: DeliverySupersedes, ADR: "0021", }, + { + // Windowed like the other two journals, and for the same reason: the + // window bounds the spool's file count, which a fleet of CronJobs would + // otherwise control (ADR 0029). + Kind: "job_runs", + Source: SourceJournal, + NaturalKey: "(window start, window length)", + Delivery: DeliverySupersedes, + ADR: "0029", + }, { Kind: "go_inventory", Source: SourceStructural, diff --git a/internal/sink/spool.go b/internal/sink/spool.go index e090c48..7acac5e 100644 --- a/internal/sink/spool.go +++ b/internal/sink/spool.go @@ -179,6 +179,23 @@ type podDisruptionsPayload struct { Records []journal.DisruptionRecord `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). +// +// The window is a delivery boundary rather than an aggregation — a run finishes +// once and carries its own instants — and it exists to keep the spool's file +// count bounded by time rather than by how many CronJobs the cluster schedules. +// +// It supersedes within its window: a run reported again while its object still +// exists rewrites its own record, so the file's newest write is its value. +type jobRunsPayload struct { + Kind string `json:"kind"` + Source string `json:"source"` + WindowStart time.Time `json:"window_start"` + WindowSeconds int64 `json:"window_seconds"` + Records []journal.JobRunRecord `json:"records"` +} + // workloadMetadataPayload is the declared shape of every collected workload // container — requests, limits, QoS, ports — and where its replicas run. Like // go-inventory it is a single superseding batch under a fixed natural key (the @@ -368,6 +385,35 @@ func (s *Spool) WriteOOMKill(e collector.OOMKill) error { return s.write(name, oomPayload{Kind: "oom_kill", Source: SourceJournal, Event: e}) } +// WriteJobRuns writes the finished Job runs 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 them) so the +// payload bytes are stable — the golden contract. +// +// A window in which no Job finished writes nothing. A cluster with no batch +// workloads has nothing to say, and an empty payload would claim otherwise for +// every quiet hour of every cluster. +func (s *Spool) WriteJobRuns(records []journal.JobRunRecord) error { + grouped := make(map[windowKey][]journal.JobRunRecord) + for _, r := range records { + k := windowKey{start: r.WindowStart, seconds: r.WindowSeconds} + grouped[k] = append(grouped[k], r) + } + for k, group := range grouped { + payload := jobRunsPayload{ + Kind: "job_runs", + Source: SourceJournal, + WindowStart: k.start, + WindowSeconds: k.seconds, + Records: group, + } + if err := s.write(fmt.Sprintf("job-runs-%d-%d.json", k.start.Unix(), k.seconds), payload); err != nil { + return err + } + } + return nil +} + // WriteGoInventory writes the current Go inventory as one superseding batch // (ADR 0010). One file per cluster, atomically replaced each flush: the newest // inventory supersedes its predecessor, exactly as an open-window usage diff --git a/internal/sink/spool_test.go b/internal/sink/spool_test.go index 68a4b4d..fc568cb 100644 --- a/internal/sink/spool_test.go +++ b/internal/sink/spool_test.go @@ -285,6 +285,86 @@ func fixedDisruptions() []journal.DisruptionRecord { } } +// fixedJobRuns is a deterministic window of finished runs: a CronJob run that +// succeeded after a retry, a CronJob run that exhausted its backoff limit, and +// a bare Job that never started before failing. The three cover the fields a +// consumer has to tell apart — a retry that cost resources, a failure reason +// that is a capacity fact, and a zero start instant that is a real state. +func fixedJobRuns() []journal.JobRunRecord { + return []journal.JobRunRecord{ + { + Namespace: "analytics", + Workload: collector.WorkloadRef{Kind: "CronJob", Name: "rollup"}, + Name: "rollup-29123456", + StartedAt: windowStart.Add(5 * time.Minute), + FinishedAt: windowStart.Add(7 * time.Minute), + Result: "succeeded", + Succeeded: 1, + Failed: 2, + Parallelism: ptrTo(int32(1)), + Completions: ptrTo(int32(1)), + BackoffLimit: ptrTo(int32(6)), + WindowStart: windowStart, + WindowSeconds: 3600, + }, + { + Namespace: "analytics", + Workload: collector.WorkloadRef{Kind: "CronJob", Name: "rollup"}, + Name: "rollup-29123457", + StartedAt: windowStart.Add(35 * time.Minute), + FinishedAt: windowStart.Add(41 * time.Minute), + Result: "failed", + FailReason: "BackoffLimitExceeded", + Failed: 7, + Parallelism: ptrTo(int32(1)), + Completions: ptrTo(int32(1)), + BackoffLimit: ptrTo(int32(6)), + WindowStart: windowStart, + WindowSeconds: 3600, + }, + { + Namespace: "shop", + Workload: collector.WorkloadRef{Kind: "Job", Name: "migrate-v42"}, + Name: "migrate-v42", + // No StartedAt: the run failed before the controller recorded a + // start, which the payload omits rather than rendering as the epoch. + FinishedAt: windowStart.Add(12 * time.Minute), + Result: "failed", + FailReason: "DeadlineExceeded", + WindowStart: windowStart, + WindowSeconds: 3600, + }, + } +} + +func ptrTo[T any](v T) *T { return &v } + +func TestGoldenJobRunsPayload(t *testing.T) { + s, dir := newTestSpool(t) + if err := s.WriteJobRuns(fixedJobRuns()); err != nil { + t.Fatal(err) + } + name := fmt.Sprintf("job-runs-%d-3600.json", windowStart.Unix()) + checkGolden(t, filepath.Join(dir, name), "job-runs.golden.json") +} + +// One file per window, not per run. A cluster whose CronJobs finish hundreds of +// runs an hour must not put the spool's file count under its own control +// (ADR 0029, the reasoning ADR 0021 established for disruptions). +func TestJobRunsOfOneWindowShareAFile(t *testing.T) { + s, dir := newTestSpool(t) + if err := s.WriteJobRuns(fixedJobRuns()); err != nil { + t.Fatal(err) + } + entries, err := os.ReadDir(dir) + if err != nil { + t.Fatal(err) + } + if len(entries) != 1 { + t.Fatalf("spool holds %d files for three runs of one window, want 1", len(entries)) + } +} + func TestGoldenPodDisruptionsPayload(t *testing.T) { s, dir := newTestSpool(t) if err := s.WritePodDisruptions(fixedDisruptions()); err != nil { diff --git a/internal/sink/testdata/job-runs.golden.json b/internal/sink/testdata/job-runs.golden.json new file mode 100644 index 0000000..86c7fc4 --- /dev/null +++ b/internal/sink/testdata/job-runs.golden.json @@ -0,0 +1,60 @@ +{ + "kind": "job_runs", + "source": "journal", + "window_start": "2026-08-06T10:00:00Z", + "window_seconds": 3600, + "records": [ + { + "namespace": "analytics", + "workload": { + "kind": "CronJob", + "name": "rollup" + }, + "name": "rollup-29123456", + "started_at": "2026-08-06T10:05:00Z", + "finished_at": "2026-08-06T10:07:00Z", + "result": "succeeded", + "succeeded": 1, + "failed": 2, + "parallelism": 1, + "completions": 1, + "backoff_limit": 6, + "window_start": "2026-08-06T10:00:00Z", + "window_seconds": 3600 + }, + { + "namespace": "analytics", + "workload": { + "kind": "CronJob", + "name": "rollup" + }, + "name": "rollup-29123457", + "started_at": "2026-08-06T10:35:00Z", + "finished_at": "2026-08-06T10:41:00Z", + "result": "failed", + "fail_reason": "BackoffLimitExceeded", + "succeeded": 0, + "failed": 7, + "parallelism": 1, + "completions": 1, + "backoff_limit": 6, + "window_start": "2026-08-06T10:00:00Z", + "window_seconds": 3600 + }, + { + "namespace": "shop", + "workload": { + "kind": "Job", + "name": "migrate-v42" + }, + "name": "migrate-v42", + "finished_at": "2026-08-06T10:12:00Z", + "result": "failed", + "fail_reason": "DeadlineExceeded", + "succeeded": 0, + "failed": 0, + "window_start": "2026-08-06T10:00:00Z", + "window_seconds": 3600 + } + ] +}