fix(scanner): reap finished scanner jobs independent of TTLAfterFinished controller (#492)

* fix(scanner): reap finished scanner jobs independent of TTLAfterFinished controller

Scanner Jobs were only cleaned up by the cluster's TTLAfterFinished
controller (via TTLSecondsAfterFinished=600). That controller can be
disabled in customer environments, leaving finished Jobs and their
Completed pods to accumulate forever.

Add a background reaper that periodically deletes the agent's own
finished Jobs (cascading to their pods) once past the same 600s grace
the TTL field encodes. It is strictly label- and namespace-scoped so it
can never touch customer Jobs, and is a no-op where the TTL controller
is present. Also add a delete_k8s_job primitive so the orchestrator can
delete a Job promptly after fetching its logs.

* chore: update image tags for main release

---------

Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

Author: mayankpande88

charts/nudgebee-agent/values.yaml

@@ -61,7 +61,7 @@ runnerServiceAccount:
 runner:
   image:
     repository: ghcr.io/nudgebee/nudgebee-agent
-    tag: 2026-06-20T09-33-11_75cc137fae195c9fc972e41d4537bda18bdc3776
+    tag: 2026-06-23T09-16-10_66bcf983a562d8776d7e8325884f25bc50864449
   # Image template the pod_profiler action launches debugger pods from.
   # The agent substitutes `{}` for the variant (bpf, jvm, python, perf, ruby).
   # Surfaces as PROFILER_IMAGE; leave empty to fall back to the binary default.

runner/cmd/agent/main.go

@@ -476,6 +476,9 @@ func run(ctx context.Context, logger *slog.Logger, cfg *config.Config) error {
 		for name := range sh {
 			lightActions[name] = struct{}{}
 		}
+		// Background cleanup of finished Jobs/pods, independent of the cluster's
+		// TTLAfterFinished controller (which may be disabled in customer envs).
+		runner.StartReaper(ctx, logger)
 		logger.Info("scanners enabled", "namespace", cfg.ScannerNamespace, "actions", len(sh))
 	}
 

runner/pkg/scanners/handlers.go

@@ -2,19 +2,20 @@ package scanners
 
 import "github.com/nudgebee/nudgebee-agent/pkg/dispatch"
 
-// Handlers wires the three generic Job primitives into the dispatch registry.
+// Handlers wires the generic Job primitives into the dispatch registry.
 //
 // The agent owns no scanner-specific behavior. The api-server's
-// scan_orchestrator builds JobSpecs (image, args, security context) and
-// drives them through schedule_k8s_job → wait_for_k8s_job → get_k8s_job_logs.
-// Adding a new scanner is a pure api-server change.
+// scan_orchestrator builds JobSpecs (image, args, security context) and drives
+// them through schedule_k8s_job → wait_for_k8s_job → get_k8s_job_logs →
+// delete_k8s_job. Adding a new scanner is a pure api-server change.
 //
-// All three actions require HMAC sig (or RSA partial-keys for callers that
-// pre-sign mutations). Not light-action — these create + read Jobs.
+// All actions require HMAC sig (or RSA partial-keys for callers that pre-sign
+// mutations). Not light-action — these create, read, and delete Jobs.
 func Handlers(r *Runner) map[string]dispatch.Handler {
 	return map[string]dispatch.Handler{
 		"schedule_k8s_job": r.handleScheduleJob,
 		"wait_for_k8s_job": r.handleWaitForJob,
 		"get_k8s_job_logs": r.handleGetJobLogs,
+		"delete_k8s_job":   r.handleDeleteJob,
 	}
 }

runner/pkg/scanners/primitives.go

@@ -190,6 +190,31 @@ func (r *Runner) handleGetJobLogs(ctx context.Context, params map[string]any) (a
 	}, nil
 }
 
+// delete_k8s_job
+// --------------
+// Inbound params:  { "job_name": "<name>" }
+//
+// Outbound:        { "deleted": true }   (also returned when the Job is already gone)
+//
+// The orchestrator calls this right after it has fetched a Job's logs so the
+// Job (and its pods) are cleaned up promptly on the happy path, instead of
+// waiting out the TTL/reaper grace window. The background reaper is the backstop
+// for crash/timeout paths where this call never happens.
+//
+// Only the runner's own namespace is touched, and only by Job name handed back
+// from schedule_k8s_job — the agent never deletes arbitrary cluster objects.
+func (r *Runner) handleDeleteJob(ctx context.Context, params map[string]any) (any, error) {
+	jobName, _ := params["job_name"].(string)
+	if jobName == "" {
+		return nil, errors.New("delete_k8s_job: job_name is required")
+	}
+	if err := r.deleteJobCascade(ctx, jobName); err != nil && !apierrors.IsNotFound(err) {
+		return nil, fmt.Errorf("delete_k8s_job: %w", err)
+	}
+	slog.Info("delete_k8s_job: deleted", "job_name", jobName, "namespace", r.Namespace)
+	return map[string]any{"deleted": true}, nil
+}
+
 // decodeJobSpec re-marshals the params["spec"] map through encoding/json so
 // JobSpec's struct tags drive the decode (handles nested corev1.Volume /
 // VolumeMount cleanly without bespoke parsing).

runner/pkg/scanners/reaper.go

@@ -0,0 +1,147 @@
+package scanners
+
+import (
+	"context"
+	"log/slog"
+	"time"
+
+	batchv1 "k8s.io/api/batch/v1"
+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+// Job cleanup is normally handled by Kubernetes' TTLAfterFinished controller
+// (we stamp every Job with TTLSecondsAfterFinished=jobTTLSeconds). That
+// controller is GA and on by default since k8s 1.21, but a customer cluster
+// can have it disabled (kube-controller-manager --controllers=-ttl-after-finished,
+// hardened/managed distros, or pre-1.21). When it's off, finished scanner Jobs
+// and their Completed pods accumulate forever.
+//
+// The reaper makes cleanup independent of that controller: the agent itself
+// periodically deletes its own finished Jobs once they're past the same grace
+// window the TTL field encodes. It only ever touches Jobs carrying BOTH the
+// managed-by and orchestrator labels, scoped to the runner's namespace, so it
+// can never delete a customer's Jobs. Where the TTL controller IS present this
+// is a harmless no-op (the controller usually wins the race first).
+const (
+	// reaperInterval is how often the sweep runs.
+	reaperInterval = 60 * time.Second
+	// reaperSweepTimeout bounds a single sweep's API calls, safely below
+	// reaperInterval so a hung API server can't wedge the loop.
+	reaperSweepTimeout = 45 * time.Second
+	// reaperListLimit caps how many Jobs a single sweep lists/deletes, so a large
+	// accumulated backlog (TTL controller off for a long time) is drained in
+	// bounded batches across successive sweeps instead of one giant List that
+	// could spike agent memory or overload the API server.
+	reaperListLimit = 100
+	// reapGraceSeconds is how long after a Job finishes before it's eligible for
+	// deletion. Matched to jobTTLSeconds so behavior mirrors the TTL field: the
+	// orchestrator's completion→log-fetch gap is seconds, so a 10-min grace
+	// preserves the existing window for retries/manual debugging of logs.
+	reapGraceSeconds = jobTTLSeconds
+)
+
+// StartReaper launches the background cleanup loop and returns immediately. The
+// loop runs until ctx is cancelled (agent shutdown). Safe to call once per
+// Runner from the agent's main wiring.
+func (r *Runner) StartReaper(ctx context.Context, logger *slog.Logger) {
+	if logger == nil {
+		logger = slog.Default()
+	}
+	logger = logger.With("component", "scanner_job_reaper", "namespace", r.Namespace)
+	go func() {
+		ticker := time.NewTicker(reaperInterval)
+		defer ticker.Stop()
+		logger.Info("scanner job reaper started",
+			"interval", reaperInterval.String(), "grace_seconds", reapGraceSeconds)
+		for {
+			select {
+			case <-ctx.Done():
+				logger.Info("scanner job reaper stopped")
+				return
+			case <-ticker.C:
+				// Bound each sweep below the tick interval so an unresponsive API
+				// server can't wedge the loop indefinitely.
+				sweepCtx, cancel := context.WithTimeout(ctx, reaperSweepTimeout)
+				if n := r.reapFinishedJobs(sweepCtx, logger, time.Now()); n > 0 {
+					logger.Info("reaped finished scanner jobs", "deleted", n)
+				}
+				cancel()
+			}
+		}
+	}()
+}
+
+// reapFinishedJobs deletes managed Jobs that finished more than reapGraceSeconds
+// before now. Returns the number deleted. Errors are logged and skipped — a
+// single bad Job must not stall the sweep. `now` is a parameter so tests can
+// drive it deterministically.
+func (r *Runner) reapFinishedJobs(ctx context.Context, logger *slog.Logger, now time.Time) int {
+	jobs, err := r.Client.BatchV1().Jobs(r.Namespace).List(ctx, metav1.ListOptions{
+		LabelSelector: managedByLabel + "=" + managedByValue + "," + orchestratorLabel + "=" + orchestratorValue,
+		Limit:         reaperListLimit,
+	})
+	if err != nil {
+		// Suppress expected cancellation noise during agent shutdown.
+		if ctx.Err() == nil {
+			logger.Warn("reaper: list managed jobs failed", "error", err)
+		}
+		return 0
+	}
+
+	grace := time.Duration(reapGraceSeconds) * time.Second
+	deleted := 0
+	for i := range jobs.Items {
+		job := &jobs.Items[i]
+		finishedAt, ok := jobFinishedAt(job)
+		if !ok {
+			continue // still running / no terminal condition yet
+		}
+		if now.Sub(finishedAt) < grace {
+			continue // within the grace window — leave it for log fetch / debugging
+		}
+		if err := r.deleteJobCascade(ctx, job.Name); err != nil && !apierrors.IsNotFound(err) {
+			if ctx.Err() == nil {
+				logger.Warn("reaper: delete job failed", "job_name", job.Name, "error", err)
+			}
+			continue
+		}
+		deleted++
+		logger.Info("reaper: deleted finished job",
+			"job_name", job.Name, "finished_at", finishedAt.UTC().Format(time.RFC3339))
+	}
+	return deleted
+}
+
+// deleteJobCascade deletes a Job with background propagation so the Job's pods
+// are garbage-collected along with it (the default foreground/orphan behavior
+// would otherwise leave the Completed pods behind).
+func (r *Runner) deleteJobCascade(ctx context.Context, jobName string) error {
+	policy := metav1.DeletePropagationBackground
+	return r.Client.BatchV1().Jobs(r.Namespace).Delete(ctx, jobName, metav1.DeleteOptions{
+		PropagationPolicy: &policy,
+	})
+}
+
+// jobFinishedAt returns when a Job reached a terminal (Complete/Failed)
+// condition. Complete Jobs carry Status.CompletionTime; Failed Jobs don't, so
+// we fall back to the terminal condition's LastTransitionTime. Returns false
+// when the Job has not finished.
+func jobFinishedAt(j *batchv1.Job) (time.Time, bool) {
+	if !jobIsTerminal(j) {
+		return time.Time{}, false
+	}
+	if j.Status.CompletionTime != nil {
+		return j.Status.CompletionTime.Time, true
+	}
+	for _, c := range j.Status.Conditions {
+		if c.Status != corev1.ConditionTrue {
+			continue
+		}
+		if c.Type == batchv1.JobComplete || c.Type == batchv1.JobFailed {
+			return c.LastTransitionTime.Time, true
+		}
+	}
+	return time.Time{}, false
+}

runner/pkg/scanners/reaper_test.go

@@ -0,0 +1,119 @@
+package scanners
+
+import (
+	"context"
+	"io"
+	"log/slog"
+	"testing"
+	"time"
+
+	batchv1 "k8s.io/api/batch/v1"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes/fake"
+)
+
+func managedJob(name string) *batchv1.Job {
+	return &batchv1.Job{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: "ns",
+			Labels: map[string]string{
+				managedByLabel:    managedByValue,
+				orchestratorLabel: orchestratorValue,
+			},
+		},
+	}
+}
+
+func completedAt(j *batchv1.Job, t time.Time) *batchv1.Job {
+	ct := metav1.NewTime(t)
+	j.Status.CompletionTime = &ct
+	j.Status.Conditions = []batchv1.JobCondition{{
+		Type:               batchv1.JobComplete,
+		Status:             corev1.ConditionTrue,
+		LastTransitionTime: ct,
+	}}
+	return j
+}
+
+func failedAt(j *batchv1.Job, t time.Time) *batchv1.Job {
+	// Failed Jobs carry no CompletionTime — only the terminal condition's time.
+	j.Status.Conditions = []batchv1.JobCondition{{
+		Type:               batchv1.JobFailed,
+		Status:             corev1.ConditionTrue,
+		LastTransitionTime: metav1.NewTime(t),
+	}}
+	return j
+}
+
+func discardLogger() *slog.Logger {
+	return slog.New(slog.NewTextHandler(io.Discard, nil))
+}
+
+func TestReap_DeletesFinishedJobsPastGrace(t *testing.T) {
+	now := time.Now()
+	grace := time.Duration(reapGraceSeconds) * time.Second
+
+	staleComplete := completedAt(managedJob("trivy-image-scan-old"), now.Add(-grace-time.Minute))
+	staleFailed := failedAt(managedJob("trivy-image-scan-failed"), now.Add(-grace-time.Minute))
+	freshComplete := completedAt(managedJob("trivy-image-scan-fresh"), now.Add(-time.Second))
+	running := managedJob("trivy-image-scan-running") // no terminal condition
+
+	r := NewRunner(fake.NewClientset(staleComplete, staleFailed, freshComplete, running), "ns", "sa")
+
+	if n := r.reapFinishedJobs(context.Background(), discardLogger(), now); n != 2 {
+		t.Fatalf("reaped = %d; want 2 (both stale jobs)", n)
+	}
+
+	remaining, err := r.Client.BatchV1().Jobs("ns").List(context.Background(), metav1.ListOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	got := map[string]bool{}
+	for _, j := range remaining.Items {
+		got[j.Name] = true
+	}
+	if got["trivy-image-scan-old"] || got["trivy-image-scan-failed"] {
+		t.Errorf("stale jobs not deleted: %v", got)
+	}
+	if !got["trivy-image-scan-fresh"] || !got["trivy-image-scan-running"] {
+		t.Errorf("fresh/running jobs must survive: %v", got)
+	}
+}
+
+func TestReap_IgnoresUnmanagedJobs(t *testing.T) {
+	now := time.Now()
+	old := metav1.NewTime(now.Add(-24 * time.Hour))
+
+	// A finished Job that is NOT ours (e.g. a customer CronJob) — must be left alone.
+	customer := &batchv1.Job{
+		ObjectMeta: metav1.ObjectMeta{Name: "customer-backup", Namespace: "ns"},
+		Status: batchv1.JobStatus{
+			CompletionTime: &old,
+			Conditions: []batchv1.JobCondition{{
+				Type: batchv1.JobComplete, Status: corev1.ConditionTrue, LastTransitionTime: old,
+			}},
+		},
+	}
+	r := NewRunner(fake.NewClientset(customer), "ns", "sa")
+
+	if n := r.reapFinishedJobs(context.Background(), discardLogger(), now); n != 0 {
+		t.Fatalf("reaped = %d; want 0 (unmanaged job must be untouched)", n)
+	}
+	if _, err := r.Client.BatchV1().Jobs("ns").Get(context.Background(), "customer-backup", metav1.GetOptions{}); err != nil {
+		t.Errorf("customer job was deleted: %v", err)
+	}
+}
+
+func TestDeleteJobCascade_BackgroundPropagation(t *testing.T) {
+	job := completedAt(managedJob("trivy-image-scan-x"), time.Now())
+	r := NewRunner(fake.NewClientset(job), "ns", "sa")
+
+	if err := r.deleteJobCascade(context.Background(), "trivy-image-scan-x"); err != nil {
+		t.Fatalf("deleteJobCascade: %v", err)
+	}
+	if _, err := r.Client.BatchV1().Jobs("ns").Get(context.Background(), "trivy-image-scan-x", metav1.GetOptions{}); err == nil {
+		t.Error("job still present after cascade delete")
+	}
+}

runner/pkg/scanners/scanners_test.go

@@ -511,7 +511,7 @@ func TestGetJobLogs_RejectsMissingName(t *testing.T) {
 func TestHandlers_RegistersOnlyPrimitives(t *testing.T) {
 	r := NewRunner(fake.NewClientset(), "ns", "")
 	hs := Handlers(r)
-	want := []string{"schedule_k8s_job", "wait_for_k8s_job", "get_k8s_job_logs"}
+	want := []string{"schedule_k8s_job", "wait_for_k8s_job", "get_k8s_job_logs", "delete_k8s_job"}
 	if len(hs) != len(want) {
 		t.Fatalf("Handlers count = %d; want %d (primitives only)", len(hs), len(want))
 	}