Filter owned-resource events in the workqueue (#4647)

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

controllers/actions.github.com/autoscalinglistener_controller.go

@@ -28,6 +28,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
@@ -873,8 +874,8 @@ func (r *AutoscalingListenerReconciler) SetupWithManager(mgr ctrl.Manager, opts
 	return builderWithOptions(
 		ctrl.NewControllerManagedBy(mgr).
 			For(&v1alpha1.AutoscalingListener{}).
-			Owns(&corev1.Pod{}).
-			Owns(&corev1.ServiceAccount{}).
+			Owns(&corev1.Pod{}, builder.WithPredicates(autoscalingListenerOwnedPodPredicate())).
+			Owns(&corev1.ServiceAccount{}, builder.WithPredicates(autoscalingListenerOwnedServiceAccountPredicate())).
 			Watches(&rbacv1.Role{}, handler.EnqueueRequestsFromMapFunc(labelBasedWatchFunc)).
 			Watches(&rbacv1.RoleBinding{}, handler.EnqueueRequestsFromMapFunc(labelBasedWatchFunc)).
 			WithEventFilter(predicate.ResourceVersionChangedPredicate{}),

controllers/actions.github.com/autoscalingrunnerset_controller.go

@@ -36,6 +36,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
@@ -888,7 +889,7 @@ func (r *AutoscalingRunnerSetReconciler) SetupWithManager(mgr ctrl.Manager, opts
 	return builderWithOptions(
 		ctrl.NewControllerManagedBy(mgr).
 			For(&v1alpha1.AutoscalingRunnerSet{}).
-			Owns(&v1alpha1.EphemeralRunnerSet{}).
+			Owns(&v1alpha1.EphemeralRunnerSet{}, builder.WithPredicates(autoscalingRunnerSetOwnedEphemeralRunnerSetPredicate())).
 			Watches(&v1alpha1.AutoscalingListener{}, handler.EnqueueRequestsFromMapFunc(
 				func(_ context.Context, o client.Object) []reconcile.Request {
 					autoscalingListener := o.(*v1alpha1.AutoscalingListener)

controllers/actions.github.com/ephemeralrunner_controller.go

@@ -36,6 +36,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
@@ -843,14 +844,7 @@ func (r *EphemeralRunnerReconciler) updateRunStatusFromPod(ctx context.Context,
 		return nil
 	}
 
-	var ready bool
-	var lastTransitionTime time.Time
-	for _, condition := range pod.Status.Conditions {
-		if condition.Type == corev1.PodReady && condition.LastTransitionTime.After(lastTransitionTime) {
-			ready = condition.Status == corev1.ConditionTrue
-			lastTransitionTime = condition.LastTransitionTime.Time
-		}
-	}
+	ready := podReady(pod)
 
 	// Publish Pending as soon as the runner is observed non-terminal, regardless of
 	// the pod phase. The controller only reaches this point once the runner
@@ -971,7 +965,7 @@ func (r *EphemeralRunnerReconciler) SetupWithManager(mgr ctrl.Manager, opts ...O
 	return builderWithOptions(
 		ctrl.NewControllerManagedBy(mgr).
 			For(&v1alpha1.EphemeralRunner{}).
-			Owns(&corev1.Pod{}).
+			Owns(&corev1.Pod{}, builder.WithPredicates(ephemeralRunnerOwnedPodPredicate())).
 			WithEventFilter(predicate.ResourceVersionChangedPredicate{}),
 		opts,
 	).Complete(r)

controllers/actions.github.com/ephemeralrunnerset_controller.go

@@ -38,6 +38,7 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/util/retry"
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
@@ -981,7 +982,7 @@ func (r *EphemeralRunnerSetReconciler) SetupWithManager(mgr ctrl.Manager, opts .
 	return builderWithOptions(
 		ctrl.NewControllerManagedBy(mgr).
 			For(&v1alpha1.EphemeralRunnerSet{}).
-			Owns(&v1alpha1.EphemeralRunner{}).
+			Owns(&v1alpha1.EphemeralRunner{}, builder.WithPredicates(ephemeralRunnerSetOwnedEphemeralRunnerPredicate())).
 			WithEventFilter(predicate.ResourceVersionChangedPredicate{}),
 		opts,
 	).Complete(r)

controllers/actions.github.com/predicates.go

@@ -0,0 +1,222 @@
+package actionsgithubcom
+
+import (
+	"slices"
+	"time"
+
+	"github.com/actions/actions-runner-controller/apis/actions.github.com/v1alpha1"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/equality"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
+)
+
+// The predicates in this file exist purely to keep work out of the workqueue.
+// They must never change what a reconciler does, so each of them is written as
+// the projection of the fields its reconciler actually reads: an update is
+// dropped only when every one of those fields is unchanged. Whenever a
+// reconciler starts reading a new field, the matching projection below has to
+// grow with it.
+//
+// Create, delete and generic events are always delivered. Only updates are
+// considered here.
+
+// autoscalingRunnerSetOwnedEphemeralRunnerSetPredicate filters updates of the
+// EphemeralRunnerSets owned by an AutoscalingRunnerSet.
+//
+// The AutoscalingRunnerSet reconciler reads the runner set's object metadata
+// (labels, annotations, finalizers and deletion timestamp), its whole spec, and,
+// through ephemeralRunnerSetOutdatedForAppliedRevision, Status.Phase together
+// with Status.AppliedActionableRevision. It never reads
+// Status.FinishedRunnerCleanupPatchID, which the EphemeralRunnerSet rewrites for
+// every listener patch id and which is the noisiest part of the object.
+func autoscalingRunnerSetOwnedEphemeralRunnerSetPredicate() predicate.Predicate {
+	return predicate.Funcs{
+		UpdateFunc: func(e event.UpdateEvent) bool {
+			oldRunnerSet, oldOk := e.ObjectOld.(*v1alpha1.EphemeralRunnerSet)
+			newRunnerSet, newOk := e.ObjectNew.(*v1alpha1.EphemeralRunnerSet)
+			if !oldOk || !newOk {
+				// Not the type we reasoned about, so we cannot claim the event
+				// is irrelevant. Let it through.
+				return true
+			}
+
+			if !equalReconciledObjectMeta(&oldRunnerSet.ObjectMeta, &newRunnerSet.ObjectMeta) ||
+				!equality.Semantic.DeepEqual(&oldRunnerSet.Spec, &newRunnerSet.Spec) {
+				return true
+			}
+
+			return oldRunnerSet.Status.Phase != newRunnerSet.Status.Phase ||
+				oldRunnerSet.Status.AppliedActionableRevision != newRunnerSet.Status.AppliedActionableRevision
+		},
+	}
+}
+
+// ephemeralRunnerSetOwnedEphemeralRunnerPredicate filters updates of the
+// EphemeralRunners owned by an EphemeralRunnerSet.
+//
+// Besides object metadata and spec, the EphemeralRunnerSet reconciler reads
+// Status.Phase, to group runners by state, Status.RunnerID, to decide whether a
+// runner still has to be removed from the service, and Status.JobID, through
+// HasJob, to skip runners that are busy serving a job. The rest of the runner
+// status (readiness, failure bookkeeping, reason, message and the remaining job
+// details written by the listener) is never read, and it is by far the noisiest
+// part of the object.
+func ephemeralRunnerSetOwnedEphemeralRunnerPredicate() predicate.Predicate {
+	return predicate.Funcs{
+		UpdateFunc: func(e event.UpdateEvent) bool {
+			oldRunner, oldOk := e.ObjectOld.(*v1alpha1.EphemeralRunner)
+			newRunner, newOk := e.ObjectNew.(*v1alpha1.EphemeralRunner)
+			if !oldOk || !newOk {
+				return true
+			}
+
+			if !equalReconciledObjectMeta(&oldRunner.ObjectMeta, &newRunner.ObjectMeta) ||
+				!equality.Semantic.DeepEqual(&oldRunner.Spec, &newRunner.Spec) {
+				return true
+			}
+
+			return oldRunner.Status.Phase != newRunner.Status.Phase ||
+				oldRunner.Status.RunnerID != newRunner.Status.RunnerID ||
+				oldRunner.Status.JobID != newRunner.Status.JobID
+		},
+	}
+}
+
+// ephemeralRunnerOwnedPodPredicate filters updates of the pod owned by an
+// EphemeralRunner.
+//
+// The EphemeralRunner reconciler reads the pod UID, its deletion timestamp, the
+// pod phase, reason and message, the container and init container statuses, and
+// the Ready condition. It never reads the pod spec or the remaining status
+// fields, which is where most pod updates land: assigned IPs, the node the pod
+// was scheduled on, start time and the conditions other than Ready.
+func ephemeralRunnerOwnedPodPredicate() predicate.Predicate {
+	return predicate.Funcs{
+		UpdateFunc: func(e event.UpdateEvent) bool {
+			oldPod, oldOk := e.ObjectOld.(*corev1.Pod)
+			newPod, newOk := e.ObjectNew.(*corev1.Pod)
+			if !oldOk || !newOk {
+				return true
+			}
+
+			if oldPod.UID != newPod.UID ||
+				!equalTime(oldPod.DeletionTimestamp, newPod.DeletionTimestamp) {
+				return true
+			}
+
+			oldStatus, newStatus := &oldPod.Status, &newPod.Status
+			if oldStatus.Phase != newStatus.Phase ||
+				oldStatus.Reason != newStatus.Reason ||
+				oldStatus.Message != newStatus.Message {
+				return true
+			}
+
+			if !equality.Semantic.DeepEqual(oldStatus.ContainerStatuses, newStatus.ContainerStatuses) ||
+				!equality.Semantic.DeepEqual(oldStatus.InitContainerStatuses, newStatus.InitContainerStatuses) {
+				return true
+			}
+
+			return podReady(oldPod) != podReady(newPod)
+		},
+	}
+}
+
+// autoscalingListenerOwnedPodPredicate filters updates of the listener pod
+// owned by an AutoscalingListener.
+//
+// The AutoscalingListener reconciler reads the pod's object metadata, because it
+// merges labels and annotations back onto the pod and compares the listener
+// config resource version annotation, and its whole spec, through
+// listenerPodSpecRequiresRecreation. Off the status it reads only the phase,
+// reason and message, to detect eviction, and the container statuses, to find
+// the listener container and branch on whether it is running or terminated. The
+// rest of the pod status is never read: assigned IPs, the node the pod landed
+// on, start time, the conditions and the init container statuses.
+func autoscalingListenerOwnedPodPredicate() predicate.Predicate {
+	return predicate.Funcs{
+		UpdateFunc: func(e event.UpdateEvent) bool {
+			oldPod, oldOk := e.ObjectOld.(*corev1.Pod)
+			newPod, newOk := e.ObjectNew.(*corev1.Pod)
+			if !oldOk || !newOk {
+				return true
+			}
+
+			if !equalReconciledObjectMeta(&oldPod.ObjectMeta, &newPod.ObjectMeta) ||
+				!equality.Semantic.DeepEqual(&oldPod.Spec, &newPod.Spec) {
+				return true
+			}
+
+			oldStatus, newStatus := &oldPod.Status, &newPod.Status
+			if oldStatus.Phase != newStatus.Phase ||
+				oldStatus.Reason != newStatus.Reason ||
+				oldStatus.Message != newStatus.Message {
+				return true
+			}
+
+			return !equality.Semantic.DeepEqual(oldStatus.ContainerStatuses, newStatus.ContainerStatuses)
+		},
+	}
+}
+
+// autoscalingListenerOwnedServiceAccountPredicate filters updates of the service
+// account owned by an AutoscalingListener.
+//
+// The AutoscalingListener reconciler only ever reads the service account's
+// labels and annotations, which it merges back onto the object. Everything else
+// the API server and the token controller write to it, the mounted secrets
+// above all, is never read.
+func autoscalingListenerOwnedServiceAccountPredicate() predicate.Predicate {
+	return predicate.Funcs{
+		UpdateFunc: func(e event.UpdateEvent) bool {
+			oldServiceAccount, oldOk := e.ObjectOld.(*corev1.ServiceAccount)
+			newServiceAccount, newOk := e.ObjectNew.(*corev1.ServiceAccount)
+			if !oldOk || !newOk {
+				return true
+			}
+
+			return !equalReconciledObjectMeta(&oldServiceAccount.ObjectMeta, &newServiceAccount.ObjectMeta)
+		},
+	}
+}
+
+// podReady reports whether the pod advertises the Ready condition. It is the
+// single source of truth for both the reconciler, which mirrors the result into
+// EphemeralRunner.Status.Ready, and the pod predicate, which has to wake the
+// reconciler whenever the result changes.
+func podReady(pod *corev1.Pod) bool {
+	var ready bool
+	var lastTransitionTime time.Time
+	for _, condition := range pod.Status.Conditions {
+		if condition.Type == corev1.PodReady && condition.LastTransitionTime.After(lastTransitionTime) {
+			ready = condition.Status == corev1.ConditionTrue
+			lastTransitionTime = condition.LastTransitionTime.Time
+		}
+	}
+	return ready
+}
+
+// equalReconciledObjectMeta compares the metadata fields the controllers in this
+// package branch on. Bookkeeping the API server owns, such as the resource
+// version, the managed fields and the timestamps outside of deletion, is
+// deliberately left out.
+func equalReconciledObjectMeta(old, new *metav1.ObjectMeta) bool {
+	return old.Generation == new.Generation &&
+		equalTime(old.DeletionTimestamp, new.DeletionTimestamp) &&
+		slices.Equal(old.Finalizers, new.Finalizers) &&
+		equality.Semantic.DeepEqual(old.Labels, new.Labels) &&
+		equality.Semantic.DeepEqual(old.Annotations, new.Annotations) &&
+		equality.Semantic.DeepEqual(old.OwnerReferences, new.OwnerReferences)
+}
+
+func equalTime(old, new *metav1.Time) bool {
+	switch {
+	case old == nil && new == nil:
+		return true
+	case old == nil || new == nil:
+		return false
+	default:
+		return old.Equal(new)
+	}
+}

controllers/actions.github.com/predicates_test.go

@@ -0,0 +1,445 @@
+package actionsgithubcom
+
+import (
+	"reflect"
+	"sort"
+	"testing"
+	"time"
+
+	"github.com/actions/actions-runner-controller/apis/actions.github.com/v1alpha1"
+	"github.com/stretchr/testify/assert"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+)
+
+func TestAutoscalingRunnerSetOwnedEphemeralRunnerSetPredicate(t *testing.T) {
+	base := func() *v1alpha1.EphemeralRunnerSet {
+		return &v1alpha1.EphemeralRunnerSet{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:        "runner-set",
+				Namespace:   "default",
+				Generation:  1,
+				Labels:      map[string]string{"app": "runner"},
+				Annotations: map[string]string{AnnotationKeyPatchID: "1"},
+				Finalizers:  []string{"finalizer"},
+			},
+			Spec: v1alpha1.EphemeralRunnerSetSpec{
+				Replicas:           2,
+				PatchID:            3,
+				ActionableRevision: 4,
+			},
+			Status: v1alpha1.EphemeralRunnerSetStatus{
+				Phase: v1alpha1.EphemeralRunnerSetPhaseRunning,
+			},
+		}
+	}
+
+	// Every field the AutoscalingRunnerSet reconciler reads off an owned
+	// EphemeralRunnerSet has to wake it up.
+	for name, mutate := range map[string]func(*v1alpha1.EphemeralRunnerSet){
+		"replicas":            func(s *v1alpha1.EphemeralRunnerSet) { s.Spec.Replicas = 5 },
+		"patch id":            func(s *v1alpha1.EphemeralRunnerSet) { s.Spec.PatchID = 9 },
+		"actionable revision": func(s *v1alpha1.EphemeralRunnerSet) { s.Spec.ActionableRevision = 7 },
+		"ephemeral runner spec": func(s *v1alpha1.EphemeralRunnerSet) {
+			s.Spec.EphemeralRunnerSpec.GitHubConfigURL = "https://github.com/org"
+		},
+		"ephemeral runner meta": func(s *v1alpha1.EphemeralRunnerSet) {
+			s.Spec.EphemeralRunnerMetadata = &v1alpha1.ResourceMeta{Labels: map[string]string{"a": "b"}}
+		},
+		"labels":             func(s *v1alpha1.EphemeralRunnerSet) { s.Labels["app"] = "changed" },
+		"annotations":        func(s *v1alpha1.EphemeralRunnerSet) { s.Annotations[AnnotationKeyPatchID] = "2" },
+		"finalizers":         func(s *v1alpha1.EphemeralRunnerSet) { s.Finalizers = nil },
+		"deletion timestamp": func(s *v1alpha1.EphemeralRunnerSet) { s.DeletionTimestamp = &metav1.Time{Time: time.Now()} },
+		"generation":         func(s *v1alpha1.EphemeralRunnerSet) { s.Generation = 2 },
+		"owner references":   func(s *v1alpha1.EphemeralRunnerSet) { s.OwnerReferences = []metav1.OwnerReference{{Name: "owner"}} },
+		"phase": func(s *v1alpha1.EphemeralRunnerSet) {
+			s.Status.Phase = v1alpha1.EphemeralRunnerSetPhaseOutdated
+		},
+		"applied actionable revision": func(s *v1alpha1.EphemeralRunnerSet) {
+			s.Status.AppliedActionableRevision = 9
+		},
+	} {
+		t.Run("reconciles on "+name, func(t *testing.T) {
+			old, updated := base(), base()
+			mutate(updated)
+			assert.True(t, autoscalingRunnerSetOwnedEphemeralRunnerSetPredicate().Update(event.UpdateEvent{ObjectOld: old, ObjectNew: updated}))
+		})
+	}
+
+	t.Run("ignores the finished runner cleanup patch id", func(t *testing.T) {
+		old, updated := base(), base()
+		updated.Status.FinishedRunnerCleanupPatchID = 3
+		updated.ResourceVersion = "2"
+		assert.False(t, autoscalingRunnerSetOwnedEphemeralRunnerSetPredicate().Update(event.UpdateEvent{ObjectOld: old, ObjectNew: updated}))
+	})
+
+	t.Run("reconciles on unexpected types", func(t *testing.T) {
+		assert.True(t, autoscalingRunnerSetOwnedEphemeralRunnerSetPredicate().Update(event.UpdateEvent{
+			ObjectOld: &corev1.Pod{},
+			ObjectNew: &corev1.Pod{},
+		}))
+	})
+
+	t.Run("does not filter create, delete or generic events", func(t *testing.T) {
+		p := autoscalingRunnerSetOwnedEphemeralRunnerSetPredicate()
+		assert.True(t, p.Create(event.CreateEvent{Object: base()}))
+		assert.True(t, p.Delete(event.DeleteEvent{Object: base()}))
+		assert.True(t, p.Generic(event.GenericEvent{Object: base()}))
+	})
+}
+
+func TestEphemeralRunnerSetOwnedEphemeralRunnerPredicate(t *testing.T) {
+	base := func() *v1alpha1.EphemeralRunner {
+		return &v1alpha1.EphemeralRunner{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:       "runner",
+				Namespace:  "default",
+				Generation: 1,
+				Annotations: map[string]string{
+					AnnotationKeyPatchID:            "1",
+					AnnotationKeyActionableRevision: "1",
+				},
+				Finalizers: []string{"finalizer"},
+			},
+			Status: v1alpha1.EphemeralRunnerStatus{
+				Phase:    v1alpha1.EphemeralRunnerPhaseRunning,
+				RunnerID: 42,
+			},
+		}
+	}
+
+	// The EphemeralRunnerSet reconciler groups runners by phase, reads the patch
+	// id and actionable revision annotations, and needs the runner id to remove
+	// the runner from the service.
+	for name, mutate := range map[string]func(*v1alpha1.EphemeralRunner){
+		"phase":               func(r *v1alpha1.EphemeralRunner) { r.Status.Phase = v1alpha1.EphemeralRunnerPhaseSucceeded },
+		"runner id":           func(r *v1alpha1.EphemeralRunner) { r.Status.RunnerID = 43 },
+		"job id":              func(r *v1alpha1.EphemeralRunner) { r.Status.JobID = "job" },
+		"patch id annotation": func(r *v1alpha1.EphemeralRunner) { r.Annotations[AnnotationKeyPatchID] = "2" },
+		"actionable revision": func(r *v1alpha1.EphemeralRunner) { r.Annotations[AnnotationKeyActionableRevision] = "2" },
+		"labels":              func(r *v1alpha1.EphemeralRunner) { r.Labels = map[string]string{"a": "b"} },
+		"finalizers":          func(r *v1alpha1.EphemeralRunner) { r.Finalizers = nil },
+		"deletion timestamp":  func(r *v1alpha1.EphemeralRunner) { r.DeletionTimestamp = &metav1.Time{Time: time.Now()} },
+		"generation":          func(r *v1alpha1.EphemeralRunner) { r.Generation = 2 },
+		"spec":                func(r *v1alpha1.EphemeralRunner) { r.Spec.GitHubConfigURL = "https://github.com/org" },
+	} {
+		t.Run("reconciles on "+name, func(t *testing.T) {
+			old, updated := base(), base()
+			mutate(updated)
+			assert.True(t, ephemeralRunnerSetOwnedEphemeralRunnerPredicate().Update(event.UpdateEvent{ObjectOld: old, ObjectNew: updated}))
+		})
+	}
+
+	t.Run("ignores runner status the runner set never reads", func(t *testing.T) {
+		old, updated := base(), base()
+		updated.Status.Ready = true
+		updated.Status.Reason = "reason"
+		updated.Status.Message = "message"
+		updated.Status.RunnerName = "runner-name"
+		updated.Status.Failures = map[string]metav1.Time{"pod": metav1.Now()}
+		updated.Status.JobRequestID = 7
+		updated.Status.JobDisplayName = "display"
+		updated.Status.JobRepositoryName = "org/repo"
+		updated.Status.JobWorkflowRef = "ref"
+		updated.Status.WorkflowRunID = 12
+		updated.ResourceVersion = "2"
+		assert.False(t, ephemeralRunnerSetOwnedEphemeralRunnerPredicate().Update(event.UpdateEvent{ObjectOld: old, ObjectNew: updated}))
+	})
+
+	t.Run("reconciles on unexpected types", func(t *testing.T) {
+		assert.True(t, ephemeralRunnerSetOwnedEphemeralRunnerPredicate().Update(event.UpdateEvent{
+			ObjectOld: &corev1.Pod{},
+			ObjectNew: &corev1.Pod{},
+		}))
+	})
+}
+
+func TestEphemeralRunnerOwnedPodPredicate(t *testing.T) {
+	base := func() *corev1.Pod {
+		return &corev1.Pod{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "runner",
+				Namespace: "default",
+				UID:       types.UID("uid"),
+			},
+			Status: corev1.PodStatus{
+				Phase: corev1.PodRunning,
+				Conditions: []corev1.PodCondition{
+					{
+						Type:               corev1.PodReady,
+						Status:             corev1.ConditionTrue,
+						LastTransitionTime: metav1.NewTime(time.Now()),
+					},
+				},
+				ContainerStatuses: []corev1.ContainerStatus{
+					{
+						Name:  v1alpha1.EphemeralRunnerContainerName,
+						Ready: true,
+						State: corev1.ContainerState{Running: &corev1.ContainerStateRunning{}},
+					},
+				},
+				InitContainerStatuses: []corev1.ContainerStatus{
+					{
+						Name:  "init",
+						State: corev1.ContainerState{Terminated: &corev1.ContainerStateTerminated{ExitCode: 0}},
+					},
+				},
+			},
+		}
+	}
+
+	// Every pod field the EphemeralRunner reconciler branches on has to wake it up.
+	for name, mutate := range map[string]func(*corev1.Pod){
+		"phase":   func(p *corev1.Pod) { p.Status.Phase = corev1.PodFailed },
+		"reason":  func(p *corev1.Pod) { p.Status.Reason = "Evicted" },
+		"message": func(p *corev1.Pod) { p.Status.Message = "evicted" },
+		"uid":     func(p *corev1.Pod) { p.UID = types.UID("other") },
+		"deletion timestamp": func(p *corev1.Pod) {
+			p.DeletionTimestamp = &metav1.Time{Time: time.Now()}
+		},
+		"runner container terminated": func(p *corev1.Pod) {
+			p.Status.ContainerStatuses[0].State = corev1.ContainerState{Terminated: &corev1.ContainerStateTerminated{ExitCode: 1}}
+		},
+		"runner container exit code": func(p *corev1.Pod) {
+			p.Status.ContainerStatuses[0].State = corev1.ContainerState{Terminated: &corev1.ContainerStateTerminated{ExitCode: 7}}
+		},
+		"runner container readiness": func(p *corev1.Pod) {
+			p.Status.ContainerStatuses[0].Ready = false
+		},
+		"init container status": func(p *corev1.Pod) {
+			p.Status.InitContainerStatuses[0].State = corev1.ContainerState{Terminated: &corev1.ContainerStateTerminated{ExitCode: 1}}
+		},
+		"ready condition": func(p *corev1.Pod) {
+			p.Status.Conditions[0].Status = corev1.ConditionFalse
+		},
+	} {
+		t.Run("reconciles on "+name, func(t *testing.T) {
+			old, updated := base(), base()
+			mutate(updated)
+			assert.True(t, ephemeralRunnerOwnedPodPredicate().Update(event.UpdateEvent{ObjectOld: old, ObjectNew: updated}))
+		})
+	}
+
+	t.Run("ignores pod noise the runner never reads", func(t *testing.T) {
+		old, updated := base(), base()
+		updated.ResourceVersion = "2"
+		updated.Labels = map[string]string{"a": "b"}
+		updated.Status.PodIP = "10.0.0.1"
+		updated.Status.PodIPs = []corev1.PodIP{{IP: "10.0.0.1"}}
+		updated.Status.HostIP = "10.0.0.2"
+		updated.Status.StartTime = &metav1.Time{Time: time.Now()}
+		updated.Status.Conditions = append(updated.Status.Conditions,
+			corev1.PodCondition{Type: corev1.PodScheduled, Status: corev1.ConditionTrue},
+			corev1.PodCondition{Type: corev1.ContainersReady, Status: corev1.ConditionTrue},
+		)
+		assert.False(t, ephemeralRunnerOwnedPodPredicate().Update(event.UpdateEvent{ObjectOld: old, ObjectNew: updated}))
+	})
+
+	t.Run("ready condition transition time alone does not reconcile", func(t *testing.T) {
+		old, updated := base(), base()
+		updated.Status.Conditions[0].LastTransitionTime = metav1.NewTime(time.Now().Add(time.Hour))
+		assert.False(t, ephemeralRunnerOwnedPodPredicate().Update(event.UpdateEvent{ObjectOld: old, ObjectNew: updated}))
+	})
+
+	t.Run("reconciles on unexpected types", func(t *testing.T) {
+		assert.True(t, ephemeralRunnerOwnedPodPredicate().Update(event.UpdateEvent{
+			ObjectOld: &v1alpha1.EphemeralRunner{},
+			ObjectNew: &v1alpha1.EphemeralRunner{},
+		}))
+	})
+}
+
+func TestAutoscalingListenerOwnedPodPredicate(t *testing.T) {
+	base := func() *corev1.Pod {
+		return &corev1.Pod{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:        "listener",
+				Namespace:   "default",
+				Generation:  1,
+				Labels:      map[string]string{"app": "listener"},
+				Annotations: map[string]string{AnnotationKeyListenerConfigResourceVersion: "1"},
+			},
+			Spec: corev1.PodSpec{
+				ServiceAccountName: "listener",
+				Containers: []corev1.Container{
+					{
+						Name:  autoscalingListenerContainerName,
+						Image: "listener:1",
+					},
+				},
+			},
+			Status: corev1.PodStatus{
+				Phase: corev1.PodRunning,
+				ContainerStatuses: []corev1.ContainerStatus{
+					{
+						Name:  autoscalingListenerContainerName,
+						State: corev1.ContainerState{Running: &corev1.ContainerStateRunning{}},
+					},
+				},
+			},
+		}
+	}
+
+	// Every listener pod field the AutoscalingListener reconciler branches on has
+	// to wake it up.
+	for name, mutate := range map[string]func(*corev1.Pod){
+		"phase":   func(p *corev1.Pod) { p.Status.Phase = corev1.PodFailed },
+		"reason":  func(p *corev1.Pod) { p.Status.Reason = "Evicted" },
+		"message": func(p *corev1.Pod) { p.Status.Message = "evicted" },
+		"labels":  func(p *corev1.Pod) { p.Labels["app"] = "changed" },
+		"listener config resource version": func(p *corev1.Pod) {
+			p.Annotations[AnnotationKeyListenerConfigResourceVersion] = "2"
+		},
+		"deletion timestamp": func(p *corev1.Pod) {
+			p.DeletionTimestamp = &metav1.Time{Time: time.Now()}
+		},
+		"generation":       func(p *corev1.Pod) { p.Generation = 2 },
+		"owner references": func(p *corev1.Pod) { p.OwnerReferences = []metav1.OwnerReference{{Name: "owner"}} },
+		"finalizers":       func(p *corev1.Pod) { p.Finalizers = []string{"finalizer"} },
+		"image":            func(p *corev1.Pod) { p.Spec.Containers[0].Image = "listener:2" },
+		"container ports": func(p *corev1.Pod) {
+			p.Spec.Containers[0].Ports = []corev1.ContainerPort{{ContainerPort: 8080}}
+		},
+		"service account name": func(p *corev1.Pod) { p.Spec.ServiceAccountName = "other" },
+		"listener container terminated": func(p *corev1.Pod) {
+			p.Status.ContainerStatuses[0].State = corev1.ContainerState{Terminated: &corev1.ContainerStateTerminated{ExitCode: 1}}
+		},
+	} {
+		t.Run("reconciles on "+name, func(t *testing.T) {
+			old, updated := base(), base()
+			mutate(updated)
+			assert.True(t, autoscalingListenerOwnedPodPredicate().Update(event.UpdateEvent{ObjectOld: old, ObjectNew: updated}))
+		})
+	}
+
+	t.Run("ignores pod noise the listener never reads", func(t *testing.T) {
+		old, updated := base(), base()
+		updated.ResourceVersion = "2"
+		updated.Status.PodIP = "10.0.0.1"
+		updated.Status.PodIPs = []corev1.PodIP{{IP: "10.0.0.1"}}
+		updated.Status.HostIP = "10.0.0.2"
+		updated.Status.StartTime = &metav1.Time{Time: time.Now()}
+		updated.Status.NominatedNodeName = "node"
+		updated.Status.QOSClass = corev1.PodQOSBestEffort
+		updated.Status.Conditions = []corev1.PodCondition{
+			{Type: corev1.PodReady, Status: corev1.ConditionTrue},
+			{Type: corev1.PodScheduled, Status: corev1.ConditionTrue},
+		}
+		updated.Status.InitContainerStatuses = []corev1.ContainerStatus{
+			{Name: "init", State: corev1.ContainerState{Terminated: &corev1.ContainerStateTerminated{ExitCode: 0}}},
+		}
+		assert.False(t, autoscalingListenerOwnedPodPredicate().Update(event.UpdateEvent{ObjectOld: old, ObjectNew: updated}))
+	})
+
+	t.Run("reconciles on unexpected types", func(t *testing.T) {
+		assert.True(t, autoscalingListenerOwnedPodPredicate().Update(event.UpdateEvent{
+			ObjectOld: &v1alpha1.AutoscalingListener{},
+			ObjectNew: &v1alpha1.AutoscalingListener{},
+		}))
+	})
+
+	t.Run("does not filter create, delete or generic events", func(t *testing.T) {
+		p := autoscalingListenerOwnedPodPredicate()
+		assert.True(t, p.Create(event.CreateEvent{Object: base()}))
+		assert.True(t, p.Delete(event.DeleteEvent{Object: base()}))
+		assert.True(t, p.Generic(event.GenericEvent{Object: base()}))
+	})
+}
+
+func TestAutoscalingListenerOwnedServiceAccountPredicate(t *testing.T) {
+	base := func() *corev1.ServiceAccount {
+		return &corev1.ServiceAccount{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:        "listener",
+				Namespace:   "default",
+				Generation:  1,
+				Labels:      map[string]string{"app": "listener"},
+				Annotations: map[string]string{"annotation": "1"},
+			},
+		}
+	}
+
+	// The reconciler merges labels and annotations back onto the service account,
+	// so a change to either has to wake it up.
+	for name, mutate := range map[string]func(*corev1.ServiceAccount){
+		"labels":      func(sa *corev1.ServiceAccount) { sa.Labels["app"] = "changed" },
+		"annotations": func(sa *corev1.ServiceAccount) { sa.Annotations["annotation"] = "2" },
+		"deletion timestamp": func(sa *corev1.ServiceAccount) {
+			sa.DeletionTimestamp = &metav1.Time{Time: time.Now()}
+		},
+		"finalizers":       func(sa *corev1.ServiceAccount) { sa.Finalizers = []string{"finalizer"} },
+		"owner references": func(sa *corev1.ServiceAccount) { sa.OwnerReferences = []metav1.OwnerReference{{Name: "owner"}} },
+	} {
+		t.Run("reconciles on "+name, func(t *testing.T) {
+			old, updated := base(), base()
+			mutate(updated)
+			assert.True(t, autoscalingListenerOwnedServiceAccountPredicate().Update(event.UpdateEvent{ObjectOld: old, ObjectNew: updated}))
+		})
+	}
+
+	t.Run("ignores service account noise the listener never reads", func(t *testing.T) {
+		old, updated := base(), base()
+		updated.ResourceVersion = "2"
+		updated.Secrets = []corev1.ObjectReference{{Name: "listener-token"}}
+		updated.ImagePullSecrets = []corev1.LocalObjectReference{{Name: "pull"}}
+		automount := true
+		updated.AutomountServiceAccountToken = &automount
+		assert.False(t, autoscalingListenerOwnedServiceAccountPredicate().Update(event.UpdateEvent{ObjectOld: old, ObjectNew: updated}))
+	})
+
+	t.Run("reconciles on unexpected types", func(t *testing.T) {
+		assert.True(t, autoscalingListenerOwnedServiceAccountPredicate().Update(event.UpdateEvent{
+			ObjectOld: &corev1.Pod{},
+			ObjectNew: &corev1.Pod{},
+		}))
+	})
+}
+
+// The predicates above are projections of the status fields their reconcilers
+// read, so a field added to either status has to be classified: either it wakes
+// the reconciler up and belongs in the tables above, or it is deliberately
+// ignored. Nothing else forces that decision, so the field names are pinned
+// here and adding one fails until somebody updates this list and the tables.
+//
+// The pin is a tripwire on the type, not a proof of correspondence: it cannot
+// tell whether a reconciler started branching on a field the matching predicate
+// still drops. Only the tables above assert the behaviour.
+func TestPredicateProjectionsCoverEveryStatusField(t *testing.T) {
+	fieldNames := func(v any) []string {
+		typ := reflect.TypeOf(v)
+		names := make([]string, 0, typ.NumField())
+		for i := 0; i < typ.NumField(); i++ {
+			names = append(names, typ.Field(i).Name)
+		}
+		sort.Strings(names)
+		return names
+	}
+
+	t.Run("ephemeral runner set status", func(t *testing.T) {
+		assert.Equal(t, []string{
+			"AppliedActionableRevision",
+			"FinishedRunnerCleanupPatchID",
+			"Phase",
+		}, fieldNames(v1alpha1.EphemeralRunnerSetStatus{}))
+	})
+
+	t.Run("ephemeral runner status", func(t *testing.T) {
+		assert.Equal(t, []string{
+			"Failures",
+			"JobDisplayName",
+			"JobID",
+			"JobRepositoryName",
+			"JobRequestID",
+			"JobWorkflowRef",
+			"Message",
+			"Phase",
+			"Ready",
+			"Reason",
+			"RunnerID",
+			"RunnerName",
+			"WorkflowRunID",
+		}, fieldNames(v1alpha1.EphemeralRunnerStatus{}))
+	})
+}