Emit JobRunTerminated event when executor observes pod terminal phase

[dejanzele]

When the scheduler preempts a run, the pod takes a variable grace period (5s to 5min depending on TerminationGracePeriodSeconds) to actually exit. preempted_timestamp records when the scheduler made the decision, but nothing records when the kubelet finished stopping the containers, so any future retry-policy consumer has no per-run signal to wait on.

This PR adds that signal. It does not consume it; the retry-gate is a separate follow-up.

What this PR does

Adds a new JobRunTerminated event emitted by the executor when it observes a pod reach PodSucceeded or PodFailed. The scheduleringester writes runs.terminated_timestamp from this event when the column is NULL, so the value reflects the kubelet-observed stop time rather than the scheduler decision time.

To preserve the existing Failed-state histogram, this PR also adds a failed_timestamp column (migration 039). MarkRunsFailed writes to failed_timestamp instead of terminated_timestamp, and state_metrics.go reads FailedTime() for the failed branch. The semantic split:

• failed_timestamp: when armada decided the run failed (scheduler event-time).
• terminated_timestamp: when the kubelet observed the pod stop (executor-observed).

For organic Succeeded or Cancelled runs the existing writers fill terminated_timestamp first and the JobRunTerminated write is a no-op. For preempted or organic-failed runs the column is NULL after the initial write, so JobRunTerminated fills it with the real kubelet-observed time.

[greptile-apps]

Greptile Summary

This PR introduces a new JobRunTerminated proto event and the infrastructure to emit, ingest, and store it, filling terminated_timestamp with the kubelet-observed pod stop time rather than the scheduler decision time. A companion failed_timestamp column (migration 039) separates "when Armada decided the run failed" from "when the kubelet stopped the pod", preserving existing metrics semantics.

• Executor side: attemptToReportJobRunTerminatedEvent is called inline from the informer Add/Update handlers and from a new reconciliation loop in ReportMissingJobEvents; the kubelet FinishedAt time flows through LatestContainerFinishedAt → event.Created → MarkRunsTerminated.
• Scheduler-ingester side: handleJobRunTerminated produces MarkRunsTerminated, which uses an IS NULL guard (runsSetTimestampIfNullStmt) so succeeded/cancelled runs keep their existing timestamp, while preempted and failed runs get the kubelet-observed time filled in.
• Metrics side: the Failed histogram branch now reads FailedTime() (decision time) and falls back to TerminatedTime() for pre-migration rows where failed_timestamp is NULL.

Confidence Score: 5/5

This PR is safe to merge. The IS NULL guard correctly arbitrates between all writers of terminated_timestamp, the FailedTime/TerminatedTime semantic split is sound, and the pre-migration fallback preserves existing histogram behaviour.

The changes are well-scoped and internally consistent. The IS NULL guard in runsSetTimestampIfNullStmt correctly defers to earlier writers (MarkRunsSucceeded, MarkJobsCancelled) while letting the kubelet-observed time fill in for preempted and organically-failed runs. The FailedTime() → TerminatedTime() fallback in state_metrics.go handles pre-migration rows cleanly. LatestContainerFinishedAt documents its node-loss edge case and falls back safely. Tests cover the main guard paths and reconciliation logic.

No files require special attention.

Important Files Changed

Filename	Overview
internal/executor/reporter/event.go	Adds CreateJobRunTerminatedEvent using LatestContainerFinishedAt for the event timestamp — correctly reads kubelet-reported FinishedAt rather than wall-clock time.
internal/executor/util/pod_util.go	Adds LatestContainerFinishedAt (scans container and init-container statuses, falls back to time.Now() for node-loss pods) and HasJobRunTerminatedBeenReported; well-documented and tested.
internal/executor/service/job_state_reporter.go	Wires attemptToReportJobRunTerminatedEvent into Add/UpdateFunc handlers and adds a reconciliation loop in ReportMissingJobEvents; idempotent via reported_terminated annotation.
internal/scheduleringester/schedulerdb.go	Changes MarkRunsFailed to write failed_timestamp instead of terminated_timestamp; adds MarkRunsTerminated case using runsSetTimestampIfNullStmt with a clean IS NULL guard.
internal/scheduler/metrics/state_metrics.go	Failed-histogram branch now uses FailedTime() with TerminatedTime() fallback for pre-migration rows; correctly preserves semantics across the migration boundary.
internal/scheduler/database/migrations/039_add_runs_failed_timestamp.sql	Adds failed_timestamp nullable timestamptz column with IF NOT EXISTS guard; safe to apply.
internal/scheduleringester/dbops.go	Adds MarkRunsTerminated map type with Merge, CanBeAppliedBefore, and GetOperation implementations consistent with sibling operations.
internal/scheduleringester/instructions.go	Routes JobRunTerminated events to handleJobRunTerminated which emits MarkRunsTerminated using event.Created (the kubelet FinishedAt time) as the stored timestamp.
internal/scheduler/metrics.go	Fixes inflated queued-time window: falls back to FailedTime() when TerminatedTime() is NULL (between MarkRunsFailed and MarkRunsTerminated arrival).
internal/scheduleringester/schedulerdb_test.go	New TestMarkRunsTerminated covers preempted, succeeded, and failed seed states; MarkRunsFailed assertion updated to check failed_timestamp instead of terminated_timestamp.

Sequence Diagram

sequenceDiagram
    participant K8s as Kubernetes/Kubelet
    participant Exec as Executor
    participant Kafka as Kafka/Pulsar
    participant Ingest as SchedulerIngester
    participant DB as Scheduler DB

    K8s->>Exec: Pod → PodSucceeded/PodFailed (informer event)
    Exec->>Exec: reportCurrentStatus/reportStatusUpdate
    Exec->>Exec: attemptToReportJobRunTerminatedEvent
    Exec->>Kafka: "JobRunTerminated {Created=LatestContainerFinishedAt}"

    Note over Exec: Separately, scheduler also emits:
    Exec->>Kafka: JobRunErrors (for failed/preempted runs)

    Kafka->>Ingest: JobRunErrors
    Ingest->>DB: "MarkRunsFailed → failed=true, failed_timestamp=event_time"

    Kafka->>Ingest: JobRunTerminated
    Ingest->>DB: "MarkRunsTerminated → terminated_timestamp=kubelet_time (WHERE IS NULL)"

    Note over DB: Succeeded runs: MarkRunsSucceeded already set terminated_timestamp → IS NULL blocks
    Note over DB: Failed/preempted runs: terminated_timestamp was NULL → IS NULL allows kubelet time in

Loading

_{Reviews (24): Last reviewed commit: "Emit JobRunTerminated event when executo..." | Re-trigger Greptile}

[dejanzele]

@greptileai

[dejanzele]

@greptileai

[dejanzele]

@greptileai