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 + } + ] +}