diff --git a/cmd/lookout/main.go b/cmd/lookout/main.go index 85f85143a8c..e325fb7fc81 100644 --- a/cmd/lookout/main.go +++ b/cmd/lookout/main.go @@ -117,8 +117,12 @@ func prune(ctx *armadacontext.Context, config configuration.LookoutConfig) { if config.PrunerConfig.ZombieRepairThreshold != nil { zombieRepairThreshold = *config.PrunerConfig.ZombieRepairThreshold } - log.Infof("expireAfter: %v, batchSize: %v, timeout: %v, zombieRepairThreshold: %v", - config.PrunerConfig.ExpireAfter, config.PrunerConfig.BatchSize, config.PrunerConfig.Timeout, zombieRepairThreshold) + leaseReturnedZombieRepairThreshold := 3 * time.Hour + if config.PrunerConfig.LeaseReturnedZombieRepairThreshold != nil { + leaseReturnedZombieRepairThreshold = *config.PrunerConfig.LeaseReturnedZombieRepairThreshold + } + log.Infof("expireAfter: %v, batchSize: %v, timeout: %v, zombieRepairThreshold: %v, leaseReturnedZombieRepairThreshold: %v", + config.PrunerConfig.ExpireAfter, config.PrunerConfig.BatchSize, config.PrunerConfig.Timeout, zombieRepairThreshold, leaseReturnedZombieRepairThreshold) ctxTimeout, cancel := armadacontext.WithTimeout(ctx, config.PrunerConfig.Timeout) defer cancel() @@ -128,6 +132,7 @@ func prune(ctx *armadacontext.Context, config configuration.LookoutConfig) { config.PrunerConfig.ExpireAfter, config.PrunerConfig.DeduplicationExpireAfter, zombieRepairThreshold, + leaseReturnedZombieRepairThreshold, config.PrunerConfig.BatchSize, clock.RealClock{}, config.ExperimentalHotColdSplit, diff --git a/config/lookout/config.yaml b/config/lookout/config.yaml index bcdce897c2b..9e97a2e0cf0 100644 --- a/config/lookout/config.yaml +++ b/config/lookout/config.yaml @@ -23,6 +23,7 @@ prunerConfig: expireAfter: 1008h # 42 days / 6 weeks deduplicationExpireAfter: 168h # 7 days zombieRepairThreshold: 15m # grace period before repairing jobs whose latest run is terminal but state is non-terminal + leaseReturnedZombieRepairThreshold: 3h # longer grace period before repairing (to failed) jobs whose latest run is lease-returned/lease-expired but state is non-terminal timeout: 1h batchSize: 1000 pushgatewayUrl: "" diff --git a/internal/lookout/configuration/types.go b/internal/lookout/configuration/types.go index 16e6ef08e66..589da2cfdd2 100644 --- a/internal/lookout/configuration/types.go +++ b/internal/lookout/configuration/types.go @@ -51,9 +51,21 @@ type PrunerConfig struct { // If nil, defaults to 15 minutes. Set to 0 explicitly to disable zombie // reconciliation entirely. ZombieRepairThreshold *time.Duration - Timeout time.Duration - BatchSize int - Postgres configuration.PostgresConfig + // LeaseReturnedZombieRepairThreshold is the minimum age of a lease-returned + // or lease-expired latest run before a zombie job in that state will be + // repaired by the pruner (to FAILED). Unlike the other zombie-causing run + // states, lease-returned/lease-expired are not unconditionally terminal for + // the job: the scheduler is normally expected to follow up with either a + // requeue or a failure, and that decision can legitimately take much + // longer than ordinary ingester lag. This threshold should therefore + // typically be set substantially longer than ZombieRepairThreshold. + // + // If nil, defaults to 3 hours. Set to 0 explicitly to disable repair of + // lease-returned/lease-expired zombies entirely. + LeaseReturnedZombieRepairThreshold *time.Duration + Timeout time.Duration + BatchSize int + Postgres configuration.PostgresConfig // PushgatewayUrl is the URL of a Prometheus Pushgateway (or compatible // endpoint) to push pruner metrics to after each run. If empty, no metrics // are pushed. diff --git a/internal/lookout/pruner/doc.go b/internal/lookout/pruner/doc.go index ec0d23e8db3..63a26988e1c 100644 --- a/internal/lookout/pruner/doc.go +++ b/internal/lookout/pruner/doc.go @@ -3,9 +3,15 @@ // The pruner runs periodically and performs three tasks: // // 1. Reconciles "zombie" jobs whose state column is non-terminal but whose -// latest run is in a terminal state. This addresses the residue of a now- -// fixed ingester bug. Reconciliation is gated by a configurable grace -// period to avoid racing in-flight state transitions and ingester lag. +// latest run is in a terminal state, or in a lease-returned/lease-expired +// state whose expected follow-up event (requeue or failure) never +// arrived. This addresses the residue of a now-fixed ingester bug, and of +// job-level events lost by the ingester more generally. Reconciliation is +// gated by two independently configurable grace periods to avoid racing +// in-flight state transitions and ingester lag: +// - a short one for the unconditionally terminal run states +// - a longer one for lease-returned/lease-expired, which may still be +// legitimately retried // // 2. Deletes terminal jobs (and their associated run, spec, and error rows) // that are older than a configurable lifetime, in batches. diff --git a/internal/lookout/pruner/pruner.go b/internal/lookout/pruner/pruner.go index f4485854db6..a60c4891ca9 100644 --- a/internal/lookout/pruner/pruner.go +++ b/internal/lookout/pruner/pruner.go @@ -19,14 +19,15 @@ func PruneDb( jobLifetime time.Duration, deduplicationLifetime time.Duration, zombieRepairThreshold time.Duration, + leaseReturnedZombieRepairThreshold time.Duration, batchLimit int, clock clock.Clock, hotColdSplit bool, ) error { var result *multierror.Error - if zombieRepairThreshold > 0 { - if _, err := ReconcileZombieJobs(ctx, db, zombieRepairThreshold, batchLimit, clock); err != nil { + if zombieRepairThreshold > 0 || leaseReturnedZombieRepairThreshold > 0 { + if _, err := ReconcileZombieJobs(ctx, db, zombieRepairThreshold, leaseReturnedZombieRepairThreshold, batchLimit, clock); err != nil { result = multierror.Append(result, err) } } @@ -43,7 +44,10 @@ func PruneDb( } func deleteDeduplications(ctx *armadacontext.Context, db *pgx.Conn, deduplicationLifetime time.Duration, clock clock.Clock) error { - cutOffTime := clock.Now().Add(-deduplicationLifetime) + // job_deduplication.inserted is written in UTC; clock.Now() is not + // guaranteed to be, so normalize before using it as a cutoff bind + // parameter (see the identical reasoning in reconcile_zombies.go). + cutOffTime := clock.Now().UTC().Add(-deduplicationLifetime) log.Infof("Deleting all rows from job_deduplication older than %s", cutOffTime) cmdTag, err := db.Exec(ctx, "DELETE FROM job_deduplication WHERE inserted <= $1", cutOffTime) if err != nil { @@ -55,7 +59,11 @@ func deleteDeduplications(ctx *armadacontext.Context, db *pgx.Conn, deduplicatio func deleteJobs(ctx *armadacontext.Context, db *pgx.Conn, jobLifetime time.Duration, batchLimit int, clock clock.Clock, hotColdSplit bool) error { now := clock.Now() - cutOffTime := now.Add(-jobLifetime) + // job.last_transition_time is written in UTC; normalize before using it + // as a cutoff bind parameter (see the identical reasoning in + // reconcile_zombies.go). now itself is left as-is since it is only used + // below for elapsed-time logging, where the zone does not matter. + cutOffTime := now.UTC().Add(-jobLifetime) totalJobsToDelete, err := createJobIdsToDeleteTempTable(ctx, db, cutOffTime, hotColdSplit) if err != nil { return errors.WithStack(err) diff --git a/internal/lookout/pruner/pruner_test.go b/internal/lookout/pruner/pruner_test.go index 0ee6e29cf47..68bd74f0492 100644 --- a/internal/lookout/pruner/pruner_test.go +++ b/internal/lookout/pruner/pruner_test.go @@ -222,7 +222,7 @@ func TestPruneDb(t *testing.T) { dbConn, err := db.Acquire(ctx) assert.NoError(t, err) isHC := isHotColdSchema(ctx, db) - err = PruneDb(ctx, dbConn.Conn(), tc.expireAfter, 0, 0, 10, clock.NewFakeClock(baseTime), isHC) + err = PruneDb(ctx, dbConn.Conn(), tc.expireAfter, 0, 0, 0, 10, clock.NewFakeClock(baseTime), isHC) assert.NoError(t, err) queriedJobIdsPerTable := []map[string]bool{ diff --git a/internal/lookout/pruner/reconcile_zombies.go b/internal/lookout/pruner/reconcile_zombies.go index 3a5eb8f4f80..9bc411089a6 100644 --- a/internal/lookout/pruner/reconcile_zombies.go +++ b/internal/lookout/pruner/reconcile_zombies.go @@ -20,6 +20,34 @@ import ( // parameters) is safe because they are compile-time constants from // internal/common/database/lookout, not user input. It also lets PostgreSQL // infer mapping.new_state as smallint without explicit casts. +// +// LEASE_RETURNED and LEASE_EXPIRED are mapped to job FAILED rather than left +// unhandled. Ordinarily a lease-returned/expired run is followed by either a +// JobRequeued or a terminal JobErrors event, so the run itself is not a +// reliable terminal signal for the job. But if that follow-up event is lost +// (e.g. dropped by the ingester), the job is stuck showing a non-terminal +// state forever with no other event to correct it. FAILED is a conservative +// choice: it may mislabel a job that was actually still being retried, but it +// stops a permanently stuck job from being reported as active indefinitely. +// +// LEASE_RETURNED/LEASE_EXPIRED use a separate, longer cutoff ($2) than the +// other four run states ($1): unlike the other four, which are unconditionally +// terminal for the job the moment the run reaches them, a lease-returned or +// lease-expired run is normally expected to be followed by a legitimate +// scheduler decision (retry or fail) that can take substantially longer than +// ordinary ingester lag to arrive. +// +// The LEASE_RETURNED/LEASE_EXPIRED branch additionally requires +// job.last_transition_time < run.finished (strictly). A successful requeue +// moves job.state to QUEUED (advancing last_transition_time) without +// changing latest_run_id, which still keeps pointing at the old +// lease-returned/expired run until the next lease is granted. Without this +// check, a job that was legitimately requeued and is simply waiting for its +// next lease -- rather than one whose follow-up event was lost -- would be +// misdetected as a zombie and incorrectly marked FAILED. The comparison must +// be strict rather than <=: a requeue timestamped identically to the +// lease-returned/expired event (e.g. both derived from the same ingested +// batch) still counts as having touched the job. var reconcileZombiesQuery = fmt.Sprintf(` UPDATE job SET state = mapping.new_state, @@ -32,10 +60,12 @@ var reconcileZombiesQuery = fmt.Sprintf(` j.state AS old_state, j.latest_run_id AS run_id, CASE r.job_run_state - WHEN %[1]d THEN %[2]d -- run succeeded -> job succeeded - WHEN %[3]d THEN %[4]d -- run failed -> job failed - WHEN %[5]d THEN %[6]d -- run cancelled -> job cancelled - WHEN %[7]d THEN %[8]d -- run preempted -> job preempted + WHEN %[1]d THEN %[2]d -- run succeeded -> job succeeded + WHEN %[3]d THEN %[4]d -- run failed -> job failed + WHEN %[5]d THEN %[6]d -- run cancelled -> job cancelled + WHEN %[7]d THEN %[8]d -- run preempted -> job preempted + WHEN %[13]d THEN %[4]d -- run lease returned -> job failed + WHEN %[14]d THEN %[4]d -- run lease expired -> job failed ELSE j.state -- defensive: a job_run_state that -- passes the IN filter but is not -- listed above (e.g. after a future @@ -47,10 +77,28 @@ var reconcileZombiesQuery = fmt.Sprintf(` FROM job j JOIN job_run r ON r.run_id = j.latest_run_id WHERE j.state IN (%[9]d, %[10]d, %[11]d, %[12]d) - AND r.job_run_state IN (%[1]d, %[3]d, %[5]d, %[7]d) AND r.finished IS NOT NULL - AND r.finished < $1 - LIMIT $2 + AND ( + (r.job_run_state IN (%[1]d, %[3]d, %[5]d, %[7]d) AND r.finished < $1) + OR + ( + r.job_run_state IN (%[13]d, %[14]d) + AND r.finished < $2 + -- A JobRequeued event moves job.state to QUEUED and advances + -- last_transition_time, but leaves latest_run_id pointing at the + -- now-stale lease-returned/expired run until the next lease is + -- granted. Without this check, a job that was legitimately + -- requeued and is simply waiting for its next lease would be + -- misdetected as a zombie. The comparison is strict (<, not <=) + -- because a requeue recorded with the same timestamp as the + -- lease-returned/expired event (e.g. both derived from the same + -- ingested batch) must still count as "touched": only a + -- last_transition_time strictly before finished proves no + -- later event has updated the job since the run finished. + AND j.last_transition_time < r.finished + ) + ) + LIMIT $3 ) AS mapping WHERE job.job_id = mapping.job_id -- Filter out ELSE-branch rows so they don't (a) get last_transition_time @@ -66,6 +114,8 @@ var reconcileZombiesQuery = fmt.Sprintf(` lookout.JobLeasedOrdinal, lookout.JobPendingOrdinal, lookout.JobRunningOrdinal, + lookout.JobRunLeaseReturnedOrdinal, + lookout.JobRunLeaseExpiredOrdinal, ) // countZombiesWithNullFinishedQuery counts jobs that match the zombie shape @@ -76,13 +126,15 @@ var countZombiesWithNullFinishedQuery = fmt.Sprintf(` SELECT COUNT(*) FROM job j JOIN job_run r ON r.run_id = j.latest_run_id - WHERE j.state IN (%[5]d, %[6]d, %[7]d, %[8]d) - AND r.job_run_state IN (%[1]d, %[2]d, %[3]d, %[4]d) + WHERE j.state IN (%[7]d, %[8]d, %[9]d, %[10]d) + AND r.job_run_state IN (%[1]d, %[2]d, %[3]d, %[4]d, %[5]d, %[6]d) AND r.finished IS NULL`, lookout.JobRunSucceededOrdinal, lookout.JobRunFailedOrdinal, lookout.JobRunCancelledOrdinal, lookout.JobRunPreemptedOrdinal, + lookout.JobRunLeaseReturnedOrdinal, + lookout.JobRunLeaseExpiredOrdinal, lookout.JobQueuedOrdinal, lookout.JobLeasedOrdinal, lookout.JobPendingOrdinal, @@ -90,20 +142,47 @@ var countZombiesWithNullFinishedQuery = fmt.Sprintf(` ) // ReconcileZombieJobs finds jobs whose state column is non-terminal but whose -// latest run is in a terminal state and finished more than zombieRepairThreshold -// ago, and updates job.state (and last_transition_time) to match the run. -// Returns the number of jobs repaired. +// latest run is in a terminal state, or in a lease-returned/lease-expired +// state that never received its expected follow-up event, and updates +// job.state (and last_transition_time) to match the run. The two cases use +// separate grace periods: zombieRepairThreshold for the unconditionally +// terminal run states, and leaseReturnedZombieRepairThreshold (normally much +// longer) for lease-returned/lease-expired, since those are legitimately +// followed by a scheduler retry-or-fail decision that can take a while to +// arrive. A zero threshold disables reconciliation for its run-state group +// independently of the other. Returns the number of jobs repaired. func ReconcileZombieJobs( ctx *armadacontext.Context, db *pgx.Conn, zombieRepairThreshold time.Duration, + leaseReturnedZombieRepairThreshold time.Duration, batchLimit int, clock clock.Clock, ) (int, error) { - cutOffTime := clock.Now().Add(-zombieRepairThreshold) + // A zero threshold disables reconciliation for that run-state group. Using + // the zero time.Time (year 1) as the cutoff, rather than clock.Now(), + // ensures "r.finished < cutoff" can never match instead of matching + // everything already finished. + // + // clock.Now() is normalized to UTC before use: job_run.finished and + // job.last_transition_time are always written in UTC (see + // protoutil.ToStdTime(...).UTC() in the ingester), but clock.Now() is not + // guaranteed to return a UTC-zoned time.Time -- Postgres's naive + // "timestamp" columns encode the wall-clock digits of whatever zone the + // bind parameter is in, not a zone-normalized instant, so comparing a + // non-UTC cutoff against a UTC-stored value would silently skew every + // comparison in this file by the process's UTC offset. + cutOffTime := time.Time{} + if zombieRepairThreshold > 0 { + cutOffTime = clock.Now().UTC().Add(-zombieRepairThreshold) + } + leaseReturnedCutOffTime := time.Time{} + if leaseReturnedZombieRepairThreshold > 0 { + leaseReturnedCutOffTime = clock.Now().UTC().Add(-leaseReturnedZombieRepairThreshold) + } totalRepaired := 0 for { - batchRepaired, err := reconcileZombieBatch(ctx, db, cutOffTime, batchLimit) + batchRepaired, err := reconcileZombieBatch(ctx, db, cutOffTime, leaseReturnedCutOffTime, batchLimit) if err != nil { return totalRepaired, err } @@ -151,6 +230,7 @@ func reconcileZombieBatch( ctx *armadacontext.Context, db *pgx.Conn, cutOffTime time.Time, + leaseReturnedCutOffTime time.Time, batchLimit int, ) (int, error) { type repair struct { @@ -168,7 +248,7 @@ func reconcileZombieBatch( IsoLevel: pgx.ReadCommitted, AccessMode: pgx.ReadWrite, }, func(tx pgx.Tx) error { - rows, err := tx.Query(ctx, reconcileZombiesQuery, cutOffTime, batchLimit) + rows, err := tx.Query(ctx, reconcileZombiesQuery, cutOffTime, leaseReturnedCutOffTime, batchLimit) if err != nil { return errors.WithStack(err) } diff --git a/internal/lookout/pruner/reconcile_zombies_test.go b/internal/lookout/pruner/reconcile_zombies_test.go index 71d97d73dec..ea935c69b6c 100644 --- a/internal/lookout/pruner/reconcile_zombies_test.go +++ b/internal/lookout/pruner/reconcile_zombies_test.go @@ -156,7 +156,7 @@ func TestReconcileZombieJobs(t *testing.T) { dbConn, err := db.Acquire(ctx) require.NoError(t, err) - repaired, err := ReconcileZombieJobs(ctx, dbConn.Conn(), tc.zombieRepairThreshold, tc.batchSize, clock.NewFakeClock(baseTime)) + repaired, err := ReconcileZombieJobs(ctx, dbConn.Conn(), tc.zombieRepairThreshold, tc.zombieRepairThreshold, tc.batchSize, clock.NewFakeClock(baseTime)) require.NoError(t, err) expectedRepairs := 0 @@ -214,7 +214,7 @@ func TestReconcileZombieJobsLeavesNonZombiesAlone(t *testing.T) { dbConn, err := db.Acquire(ctx) require.NoError(t, err) - repaired, err := ReconcileZombieJobs(ctx, dbConn.Conn(), 1*time.Hour, 10, clock.NewFakeClock(baseTime)) + repaired, err := ReconcileZombieJobs(ctx, dbConn.Conn(), 1*time.Hour, 1*time.Hour, 10, clock.NewFakeClock(baseTime)) require.NoError(t, err) assert.Equal(t, 0, repaired) @@ -226,6 +226,222 @@ func TestReconcileZombieJobsLeavesNonZombiesAlone(t *testing.T) { assert.NoError(t, err) } +// TestReconcileZombieJobsRepairsLeaseReturnedAndExpired verifies that jobs +// left stuck in a non-terminal state whose latest run is lease-returned or +// lease-expired -- because the job-level JobRequeued/JobErrors event that +// should have followed was lost -- are conservatively repaired to FAILED. +func TestReconcileZombieJobsRepairsLeaseReturnedAndExpired(t *testing.T) { + err := lookout.WithLookoutDb(func(db *pgxpool.Pool) error { + converter := instructions.NewInstructionConverter(metrics.Get().Metrics, "armadaproject.io/", []string{}, &compress.NoOpCompressor{}) + store := lookoutdb.NewLookoutDb(db, nil, metrics.Get(), 10, 10) + + ctx, cancel := armadacontext.WithTimeout(armadacontext.Background(), 5*time.Minute) + defer cancel() + + leaseReturnedJobId := util.NewULID() + leaseReturnedRunId := uuid.NewString() + repository.NewJobSimulator(converter, store). + Submit("queue", "jobSet", "owner", "namespace", baseTime.Add(-3*time.Hour), &repository.JobOptions{JobId: leaseReturnedJobId}). + Lease(leaseReturnedRunId, "cluster", "node", "pool", baseTime.Add(-3*time.Hour)). + Pending(leaseReturnedRunId, "cluster", baseTime.Add(-3*time.Hour)). + LeaseReturned(leaseReturnedRunId, "lease returned", baseTime.Add(-2*time.Hour)). + Build() + + leaseExpiredJobId := util.NewULID() + leaseExpiredRunId := uuid.NewString() + repository.NewJobSimulator(converter, store). + Submit("queue", "jobSet", "owner", "namespace", baseTime.Add(-3*time.Hour), &repository.JobOptions{JobId: leaseExpiredJobId}). + Lease(leaseExpiredRunId, "cluster", "node", "pool", baseTime.Add(-3*time.Hour)). + Pending(leaseExpiredRunId, "cluster", baseTime.Add(-3*time.Hour)). + LeaseExpired(leaseExpiredRunId, baseTime.Add(-2*time.Hour), clock.NewFakeClock(baseTime)). + Build() + + dbConn, err := db.Acquire(ctx) + require.NoError(t, err) + + // zombieRepairThreshold is set to a value too long to have repaired + // these on its own (were it wrongly applied to this run-state group), + // proving leaseReturnedZombieRepairThreshold is what governs here. + repaired, err := ReconcileZombieJobs(ctx, dbConn.Conn(), 24*time.Hour, 1*time.Hour, 10, clock.NewFakeClock(baseTime)) + require.NoError(t, err) + assert.Equal(t, 2, repaired) + + assert.Equal(t, lookout.JobFailed, readJobState(t, ctx, db, leaseReturnedJobId)) + assert.Equal(t, lookout.JobFailed, readJobState(t, ctx, db, leaseExpiredJobId)) + + return nil + }) + assert.NoError(t, err) +} + +// TestReconcileZombieJobsLeasesReturnedWithinGracePeriodAreLeftAlone verifies +// that a lease-returned/expired run does not get repaired while still within +// the grace period, since the job may simply be about to be re-leased. +func TestReconcileZombieJobsLeasesReturnedWithinGracePeriodAreLeftAlone(t *testing.T) { + err := lookout.WithLookoutDb(func(db *pgxpool.Pool) error { + converter := instructions.NewInstructionConverter(metrics.Get().Metrics, "armadaproject.io/", []string{}, &compress.NoOpCompressor{}) + store := lookoutdb.NewLookoutDb(db, nil, metrics.Get(), 10, 10) + + ctx, cancel := armadacontext.WithTimeout(armadacontext.Background(), 5*time.Minute) + defer cancel() + + jobId := util.NewULID() + runId := uuid.NewString() + repository.NewJobSimulator(converter, store). + Submit("queue", "jobSet", "owner", "namespace", baseTime.Add(-1*time.Hour), &repository.JobOptions{JobId: jobId}). + Lease(runId, "cluster", "node", "pool", baseTime.Add(-1*time.Hour)). + Pending(runId, "cluster", baseTime.Add(-1*time.Hour)). + LeaseReturned(runId, "lease returned", baseTime.Add(-30*time.Minute)). + Build() + + dbConn, err := db.Acquire(ctx) + require.NoError(t, err) + + // zombieRepairThreshold is set far shorter than the run's 30-minute + // age -- if it were wrongly applied to this run-state group, the job + // would get repaired. leaseReturnedZombieRepairThreshold (1 hour) is + // what should govern here, and correctly leaves the job alone. + repaired, err := ReconcileZombieJobs(ctx, dbConn.Conn(), 1*time.Minute, 1*time.Hour, 10, clock.NewFakeClock(baseTime)) + require.NoError(t, err) + assert.Equal(t, 0, repaired) + + assert.Equal(t, lookout.JobPending, readJobState(t, ctx, db, jobId)) + + return nil + }) + assert.NoError(t, err) +} + +// TestReconcileZombieJobsLeavesLegitimatelyRequeuedJobsAlone verifies that a +// job whose lease was returned/expired and was then legitimately requeued -- +// moving job.state to QUEUED via a real JobRequeued event, well past the +// leaseReturnedZombieRepairThreshold cutoff -- is NOT repaired to FAILED just +// because it is still waiting for its next lease. latest_run_id still points +// at the old lease-returned run (it is only updated on the next JobRunLeased), +// so without last_transition_time being taken into account, this job would be +// indistinguishable from a true zombie whose JobRequeued event was lost. +func TestReconcileZombieJobsLeavesLegitimatelyRequeuedJobsAlone(t *testing.T) { + err := lookout.WithLookoutDb(func(db *pgxpool.Pool) error { + converter := instructions.NewInstructionConverter(metrics.Get().Metrics, "armadaproject.io/", []string{}, &compress.NoOpCompressor{}) + store := lookoutdb.NewLookoutDb(db, nil, metrics.Get(), 10, 10) + + ctx, cancel := armadacontext.WithTimeout(armadacontext.Background(), 5*time.Minute) + defer cancel() + + jobId := util.NewULID() + runId := uuid.NewString() + repository.NewJobSimulator(converter, store). + Submit("queue", "jobSet", "owner", "namespace", baseTime.Add(-3*time.Hour), &repository.JobOptions{JobId: jobId}). + Lease(runId, "cluster", "node", "pool", baseTime.Add(-3*time.Hour)). + Pending(runId, "cluster", baseTime.Add(-3*time.Hour)). + LeaseReturned(runId, "lease returned", baseTime.Add(-2*time.Hour)). + Requeued(baseTime.Add(-90 * time.Minute)). + Build() + + dbConn, err := db.Acquire(ctx) + require.NoError(t, err) + + // leaseReturnedZombieRepairThreshold (1 hour) is comfortably shorter + // than the run's 2-hour age, so this would be repaired if the + // requeue were not taken into account. + repaired, err := ReconcileZombieJobs(ctx, dbConn.Conn(), 1*time.Hour, 1*time.Hour, 10, clock.NewFakeClock(baseTime)) + require.NoError(t, err) + assert.Equal(t, 0, repaired) + + assert.Equal(t, lookout.JobQueued, readJobState(t, ctx, db, jobId)) + + return nil + }) + assert.NoError(t, err) +} + +// TestReconcileZombieJobsLeavesRequeuedJobsAloneEvenWithEqualTimestamp +// verifies that the requeue guard uses a strict "<" comparison: a JobRequeued +// event timestamped identically to the preceding lease-returned/expired event +// (e.g. both derived from the same ingested batch) must still count as having +// touched the job, since a non-strict "<=" would incorrectly let this +// legitimately-requeued job be repaired to FAILED. +func TestReconcileZombieJobsLeavesRequeuedJobsAloneEvenWithEqualTimestamp(t *testing.T) { + err := lookout.WithLookoutDb(func(db *pgxpool.Pool) error { + converter := instructions.NewInstructionConverter(metrics.Get().Metrics, "armadaproject.io/", []string{}, &compress.NoOpCompressor{}) + store := lookoutdb.NewLookoutDb(db, nil, metrics.Get(), 10, 10) + + ctx, cancel := armadacontext.WithTimeout(armadacontext.Background(), 5*time.Minute) + defer cancel() + + jobId := util.NewULID() + runId := uuid.NewString() + leaseReturnedAt := baseTime.Add(-2 * time.Hour) + repository.NewJobSimulator(converter, store). + Submit("queue", "jobSet", "owner", "namespace", baseTime.Add(-3*time.Hour), &repository.JobOptions{JobId: jobId}). + Lease(runId, "cluster", "node", "pool", baseTime.Add(-3*time.Hour)). + Pending(runId, "cluster", baseTime.Add(-3*time.Hour)). + LeaseReturned(runId, "lease returned", leaseReturnedAt). + Requeued(leaseReturnedAt). + Build() + + dbConn, err := db.Acquire(ctx) + require.NoError(t, err) + + repaired, err := ReconcileZombieJobs(ctx, dbConn.Conn(), 1*time.Hour, 1*time.Hour, 10, clock.NewFakeClock(baseTime)) + require.NoError(t, err) + assert.Equal(t, 0, repaired) + + assert.Equal(t, lookout.JobQueued, readJobState(t, ctx, db, jobId)) + + return nil + }) + assert.NoError(t, err) +} + +// TestReconcileZombieJobsUsesUTCClock verifies that the reconciler normalizes +// clock.Now() to UTC before using it as a cutoff, so a clock returning a +// non-UTC-zoned time.Time for the same instant still produces the correct +// repair decision. job_run.finished is always written in UTC by the ingester +// (see protoutil.ToStdTime), so a cutoff derived from the wall-clock digits of +// a non-UTC time.Time -- rather than the UTC-normalized instant -- would be +// skewed by the zone's offset and silently mis-repair jobs. +func TestReconcileZombieJobsUsesUTCClock(t *testing.T) { + err := lookout.WithLookoutDb(func(db *pgxpool.Pool) error { + converter := instructions.NewInstructionConverter(metrics.Get().Metrics, "armadaproject.io/", []string{}, &compress.NoOpCompressor{}) + store := lookoutdb.NewLookoutDb(db, nil, metrics.Get(), 10, 10) + + ctx, cancel := armadacontext.WithTimeout(armadacontext.Background(), 5*time.Minute) + defer cancel() + + zombie := zombieScenario{ + jobId: util.NewULID(), + runFinishedAt: baseTime.Add(-2 * time.Hour), + terminalJobState: lookout.JobSucceeded, + rewindToJobState: lookout.JobRunning, + } + seedTerminalJob(t, ctx, db, store, converter, zombie) + rewindJobState(t, ctx, db, zombie.jobId, zombie.rewindToJobState) + + dbConn, err := db.Acquire(ctx) + require.NoError(t, err) + + // The fake clock returns the same instant as baseTime, but represented + // in a non-UTC fixed zone. With a 1-hour threshold, the correct + // UTC-normalized cutoff (baseTime - 1h) is after the run's finished + // time (baseTime - 2h), so the zombie should be repaired. Without the + // .UTC() normalization, the cutoff would be computed from the -5h + // zone's wall-clock digits instead (effectively baseTime - 6h), which + // falls before the run's finished time and would wrongly leave the + // zombie unrepaired. + nonUTCClock := clock.NewFakeClock(baseTime.In(time.FixedZone("UTC-5", -5*60*60))) + + repaired, err := ReconcileZombieJobs(ctx, dbConn.Conn(), 1*time.Hour, 1*time.Hour, 10, nonUTCClock) + require.NoError(t, err) + assert.Equal(t, 1, repaired) + + assert.Equal(t, lookout.JobSucceeded, readJobState(t, ctx, db, zombie.jobId)) + + return nil + }) + assert.NoError(t, err) +} + // TestReconcileZombieJobsCountsNullFinishedZombies verifies that zombie jobs // whose latest run has no finished timestamp are observed via the // zombiesSkippedNullFinished metric (and are NOT silently repaired with bogus @@ -255,7 +471,7 @@ func TestReconcileZombieJobsCountsNullFinishedZombies(t *testing.T) { dbConn, err := db.Acquire(ctx) require.NoError(t, err) - repaired, err := ReconcileZombieJobs(ctx, dbConn.Conn(), 1*time.Hour, 10, clock.NewFakeClock(baseTime)) + repaired, err := ReconcileZombieJobs(ctx, dbConn.Conn(), 1*time.Hour, 1*time.Hour, 10, clock.NewFakeClock(baseTime)) require.NoError(t, err) // Not repaired: the reconciler refuses to touch a row with NULL finished. @@ -269,6 +485,57 @@ func TestReconcileZombieJobsCountsNullFinishedZombies(t *testing.T) { assert.NoError(t, err) } +// TestReconcileZombieJobsCountsLegitimatelyRequeuedNullFinishedJob documents a +// known, currently-unfixable limitation of countZombiesWithNullFinishedQuery +// (see the comment above that query): a job whose LEASE_RETURNED/LEASE_EXPIRED +// run has a lost finished write, and which is then legitimately requeued +// while waiting for its next lease, is indistinguishable from a true zombie +// by this diagnostic query, since it has no run.finished to compare +// job.last_transition_time against. This test locks in that documented +// behaviour so a future change does not silently alter it. It only affects +// the zombiesSkippedNullFinished metric -- the job's state is never mutated, +// since the repair query still refuses to touch a row with NULL finished. +func TestReconcileZombieJobsCountsLegitimatelyRequeuedNullFinishedJob(t *testing.T) { + err := lookout.WithLookoutDb(func(db *pgxpool.Pool) error { + converter := instructions.NewInstructionConverter(metrics.Get().Metrics, "armadaproject.io/", []string{}, &compress.NoOpCompressor{}) + store := lookoutdb.NewLookoutDb(db, nil, metrics.Get(), 10, 10) + + ctx, cancel := armadacontext.WithTimeout(armadacontext.Background(), 5*time.Minute) + defer cancel() + + jobId := util.NewULID() + runId := uuid.NewString() + repository.NewJobSimulator(converter, store). + Submit("queue", "jobSet", "owner", "namespace", baseTime.Add(-3*time.Hour), &repository.JobOptions{JobId: jobId}). + Lease(runId, "cluster", "node", "pool", baseTime.Add(-3*time.Hour)). + Pending(runId, "cluster", baseTime.Add(-3*time.Hour)). + LeaseReturned(runId, "lease returned", baseTime.Add(-2*time.Hour)). + Requeued(baseTime.Add(-90 * time.Minute)). + Build() + + _, err := db.Exec(ctx, `UPDATE job_run SET finished = NULL WHERE job_id = $1`, jobId) + require.NoError(t, err) + + dbConn, err := db.Acquire(ctx) + require.NoError(t, err) + + repaired, err := ReconcileZombieJobs(ctx, dbConn.Conn(), 1*time.Hour, 1*time.Hour, 10, clock.NewFakeClock(baseTime)) + require.NoError(t, err) + + // Not repaired: the repair query still refuses to touch a row with + // NULL finished, regardless of the requeue guard. + assert.Equal(t, 0, repaired) + assert.Equal(t, lookout.JobQueued, readJobState(t, ctx, db, jobId)) + + // Documented limitation: the legitimately-requeued job is still + // counted here, because countZombiesWithNullFinishedQuery has no + // run.finished to check the requeue guard against. + assert.Equal(t, 1.0, testutil.ToFloat64(zombiesSkippedNullFinished)) + return nil + }) + assert.NoError(t, err) +} + // TestPruneDbRunsZombieReconciliation is a thin smoke test that // PruneDb invokes ReconcileZombieJobs when the threshold is non-zero. func TestPruneDbRunsZombieReconciliation(t *testing.T) { @@ -294,7 +561,7 @@ func TestPruneDbRunsZombieReconciliation(t *testing.T) { require.NoError(t, err) isHC := isHotColdSchema(ctx, db) - err = PruneDb(ctx, dbConn.Conn(), 100*time.Hour, 100*time.Hour, 1*time.Hour, 10, clock.NewFakeClock(baseTime), isHC) + err = PruneDb(ctx, dbConn.Conn(), 100*time.Hour, 100*time.Hour, 1*time.Hour, 1*time.Hour, 10, clock.NewFakeClock(baseTime), isHC) require.NoError(t, err) assert.Equal(t, lookout.JobSucceeded, readJobState(t, ctx, db, zombie.jobId)) diff --git a/internal/lookout/repository/util.go b/internal/lookout/repository/util.go index fcbf2c43502..7f2011c5632 100644 --- a/internal/lookout/repository/util.go +++ b/internal/lookout/repository/util.go @@ -420,6 +420,24 @@ func (js *JobSimulator) Cancelled(timestamp time.Time, cancelUser string) *JobSi return js } +func (js *JobSimulator) Requeued(timestamp time.Time) *JobSimulator { + ts := timestampOrNow(timestamp) + requeuedTime := protoutil.ToStdTime(ts) + requeued := &armadaevents.EventSequence_Event{ + Created: ts, + Event: &armadaevents.EventSequence_Event_JobRequeued{ + JobRequeued: &armadaevents.JobRequeued{ + JobId: js.jobId, + }, + }, + } + js.events = append(js.events, requeued) + + js.job.State = string(lookout.JobQueued) + js.job.LastTransitionTime = requeuedTime + return js +} + func (js *JobSimulator) Reprioritized(newPriority uint32, timestamp time.Time) *JobSimulator { ts := timestampOrNow(timestamp) reprioritized := &armadaevents.EventSequence_Event{