From b8523e12a374b82c36ce843864009379ed1ed1d7 Mon Sep 17 00:00:00 2001 From: Troy Chiu Date: Fri, 31 Jul 2026 22:31:35 +0000 Subject: [PATCH 1/2] ateapi/controlapi: process syncer events via rate-limiting workqueue --- cmd/ateapi/internal/controlapi/syncer.go | 272 ++++++++++++++--------- 1 file changed, 163 insertions(+), 109 deletions(-) diff --git a/cmd/ateapi/internal/controlapi/syncer.go b/cmd/ateapi/internal/controlapi/syncer.go index 5e1d58961..1ded8acb4 100644 --- a/cmd/ateapi/internal/controlapi/syncer.go +++ b/cmd/ateapi/internal/controlapi/syncer.go @@ -17,25 +17,46 @@ package controlapi import ( "context" "errors" + "fmt" "log/slog" "maps" + "time" "github.com/agent-substrate/substrate/cmd/ateapi/internal/store" "github.com/agent-substrate/substrate/internal/resources" listersv1alpha1 "github.com/agent-substrate/substrate/pkg/client/listers/api/v1alpha1" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" - "google.golang.org/grpc/codes" - "google.golang.org/grpc/status" corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/util/wait" "k8s.io/client-go/tools/cache" + "k8s.io/client-go/util/workqueue" ) +// syncerWorkerCount is the number of goroutines draining the work queue. The +// queue never hands the same key to two workers concurrently, so per-key +// ordering is preserved. +const syncerWorkerCount = 2 + +// workerKey identifies a worker row in the store. pool is captured at enqueue +// time because once the pod is gone from the informer cache, the +// ate.dev/worker-pool label cannot be recovered from a namespace/name key. +type workerKey struct { + namespace string + pool string + name string +} + // WorkerPoolSyncer reconciles the state of worker pods from Kubernetes Informer // into the store. +// +// Informer event handlers only enqueue keys; worker goroutines reconcile each +// key against the current informer cache state, requeuing with rate-limited +// backoff on transient failures such as store.ErrVersionConflict. type WorkerPoolSyncer struct { persistence store.Interface workerInformer cache.SharedIndexInformer workerPoolLister listersv1alpha1.WorkerPoolLister + queue workqueue.TypedRateLimitingInterface[workerKey] } // NewWorkerPoolSyncer creates a new WorkerPoolSyncer. @@ -44,19 +65,28 @@ func NewWorkerPoolSyncer(persistence store.Interface, workerInformer cache.Share persistence: persistence, workerInformer: workerInformer, workerPoolLister: workerPoolLister, + queue: workqueue.NewTypedRateLimitingQueue(workqueue.DefaultTypedControllerRateLimiter[workerKey]()), } } -// Start starts the background reconciliation loop. +// Start registers the event handlers and starts the background workers. The +// informer's initial list synthesizes Add events for every existing pod, so no +// explicit startup re-list is needed as long as Start is called before the +// informer factory is started. func (s *WorkerPoolSyncer) Start(ctx context.Context) { s.workerInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{ AddFunc: func(obj interface{}) { - pod := obj.(*corev1.Pod) - s.syncWorkerToStore(ctx, pod) + s.enqueuePod(obj.(*corev1.Pod)) }, UpdateFunc: func(oldObj, newObj interface{}) { - pod := newObj.(*corev1.Pod) - s.syncWorkerToStore(ctx, pod) + oldPod := oldObj.(*corev1.Pod) + newPod := newObj.(*corev1.Pod) + // If the pool label changed, enqueue the old key too so its + // now-stale store row gets cleaned up. + if oldPod.Labels[workerPodLabel] != newPod.Labels[workerPodLabel] { + s.enqueuePod(oldPod) + } + s.enqueuePod(newPod) }, DeleteFunc: func(obj interface{}) { var pod *corev1.Pod @@ -74,40 +104,76 @@ func (s *WorkerPoolSyncer) Start(ctx context.Context) { slog.ErrorContext(ctx, "Unknown object type in delete handler", slog.Any("obj", obj)) return } - slog.InfoContext(ctx, "Syncer: removing worker from store (pod deleted)", slog.String("worker", pod.Namespace+"/"+pod.Name)) - // TODO: make this more robust. Informer event handlers cannot signal failure and - // the informer never retries them, so a cleanup that fails here is not retried - // until the next startup reconcile. The canonical fix is a rate-limited workqueue: - // enqueue the worker key and run this from a worker goroutine that requeues with - // backoff on error. - if err := s.reconcileDeadWorker(ctx, pod.Namespace, pod.Labels[workerPodLabel], pod.Name); err != nil { - slog.ErrorContext(ctx, "Failed to reconcile deleted worker", slog.String("worker", pod.Namespace+"/"+pod.Name), slog.Any("err", err)) - } + s.enqueuePod(pod) }, }) go func() { + defer s.queue.ShutDown() if !cache.WaitForCacheSync(ctx.Done(), s.workerInformer.HasSynced) { slog.ErrorContext(ctx, "Syncer: failed to sync informer cache") return } - - slog.InfoContext(ctx, "Syncer: performing initial sync on startup") - objs := s.workerInformer.GetIndexer().List() - for _, obj := range objs { - pod := obj.(*corev1.Pod) - s.syncWorkerToStore(ctx, pod) + for range syncerWorkerCount { + go wait.UntilWithContext(ctx, s.runWorker, time.Second) } - // Reconcile the other direction: clean up stored workers whose pods no - // longer exist. This recovers delete events missed while ate-api-server - // was down — neither the watch relist nor the resync period can replay a - // delete across a process restart, because the informer cache starts empty. - s.reconcileOrphanedWorkers(ctx) + // Reconcile the other direction: enqueue every stored worker so records + // whose pods no longer exist are cleaned up. This recovers delete events + // missed while ate-api-server was down — neither the watch relist nor + // the resync period can replay a delete across a process restart, + // because the informer cache starts empty. Runs after the cache sync so + // the indexer is an authoritative snapshot of live pods. + s.enqueueStoredWorkers(ctx) + + <-ctx.Done() }() } -func (s *WorkerPoolSyncer) syncWorkerToStore(ctx context.Context, pod *corev1.Pod) { +func (s *WorkerPoolSyncer) enqueuePod(pod *corev1.Pod) { + s.queue.Add(workerKey{namespace: pod.Namespace, pool: pod.Labels[workerPodLabel], name: pod.Name}) +} + +func (s *WorkerPoolSyncer) runWorker(ctx context.Context) { + for s.processNextWorkItem(ctx) { + } +} + +func (s *WorkerPoolSyncer) processNextWorkItem(ctx context.Context) bool { + key, quit := s.queue.Get() + if quit { + return false + } + defer s.queue.Done(key) + + if err := s.reconcile(ctx, key); err != nil { + slog.ErrorContext(ctx, "Syncer: reconcile failed, requeueing", + slog.String("worker", key.namespace+"/"+key.name), + slog.String("pool", key.pool), + slog.Any("err", err)) + s.queue.AddRateLimited(key) + return true + } + s.queue.Forget(key) + return true +} + +// reconcile converges the store row for key with the current pod state in the +// informer cache. Returning an error requeues the key with backoff. +func (s *WorkerPoolSyncer) reconcile(ctx context.Context, key workerKey) error { + obj, exists, err := s.workerInformer.GetIndexer().GetByKey(key.namespace + "/" + key.name) + if err != nil { + return err + } + if !exists { + slog.InfoContext(ctx, "Syncer: removing worker from store (pod deleted)", slog.String("worker", key.namespace+"/"+key.name)) + return s.reconcileDeadWorker(ctx, key.namespace, key.pool, key.name) + } + pod := obj.(*corev1.Pod) + if pod.Labels[workerPodLabel] != key.pool { + // The pod moved to a different pool; this key's store row is stale. + return s.reconcileDeadWorker(ctx, key.namespace, key.pool, key.name) + } // Checked before eligibility: draining works off the stored record by name and // never reads the pod IP, while a Terminating pod can legitimately report no // IP once its sandbox is torn down. Gating on the IP first would drop the @@ -118,79 +184,88 @@ func (s *WorkerPoolSyncer) syncWorkerToStore(ctx context.Context, pod *corev1.Po // the bound actor here — inside the pod ateom has received SIGTERM and is // gracefully shutting the actor down. Actor cleanup happens on the Pod // Deleted event. - if err := s.markWorkerDraining(ctx, pod.Namespace, pod.Labels[workerPodLabel], pod.Name); err != nil { - slog.ErrorContext(ctx, "Failed to mark worker draining", slog.String("worker", pod.Namespace+"/"+pod.Name), slog.Any("err", err)) - } - return + return s.markWorkerDraining(ctx, key.namespace, key.pool, key.name) } - if !isWorkerEligible(pod) { - return + // No IP yet; a later update event re-enqueues the pod. + return nil } + return s.createOrUpdateWorker(ctx, key, pod) +} - poolName := pod.Labels[workerPodLabel] - pool, err := s.workerPoolLister.WorkerPools(pod.Namespace).Get(poolName) +func (s *WorkerPoolSyncer) createOrUpdateWorker(ctx context.Context, key workerKey, pod *corev1.Pod) error { + pool, err := s.workerPoolLister.WorkerPools(key.namespace).Get(key.pool) if err != nil { - slog.ErrorContext(ctx, "Failed to get WorkerPool for worker pod", slog.String("worker", pod.Namespace+"/"+pod.Name), slog.String("pool", poolName), slog.Any("err", err)) - return + return fmt.Errorf("getting WorkerPool %s/%s: %w", key.namespace, key.pool, err) } - w, err := s.persistence.GetWorker(ctx, pod.Namespace, poolName, pod.Name) + w, err := s.persistence.GetWorker(ctx, key.namespace, key.pool, key.name) if err != nil { - if errors.Is(err, store.ErrNotFound) { - slog.InfoContext(ctx, "Syncer: creating worker in store", slog.String("worker", pod.Namespace+"/"+pod.Name)) - worker := &ateapipb.Worker{ - WorkerNamespace: pod.Namespace, - WorkerPool: poolName, - WorkerPod: pod.Name, - Ip: pod.Status.PodIP, - WorkerPodUid: string(pod.UID), - NodeName: pod.Spec.NodeName, - SandboxClass: string(pool.Spec.SandboxClass), - Labels: pool.GetLabels(), - State: ateapipb.Worker_STATE_ACTIVE, - } - // TODO(thockin): for now this is the only place Workers are - // created. If/when this becomes a regular API, validation should - // move there. - if errs := resources.ValidateWorker(worker, nil); len(errs) > 0 { - err := status.Error(codes.InvalidArgument, errs.ToAggregate().Error()) - slog.ErrorContext(ctx, "Invalid worker", slog.Any("err", err)) - return - } - err = s.persistence.CreateWorker(ctx, worker) - if err != nil && !errors.Is(err, store.ErrAlreadyExists) { - slog.ErrorContext(ctx, "Failed to create worker in store", slog.Any("err", err)) - } - return + if !errors.Is(err, store.ErrNotFound) { + return fmt.Errorf("getting worker from store: %w", err) + } + slog.InfoContext(ctx, "Syncer: creating worker in store", slog.String("worker", key.namespace+"/"+key.name)) + worker := &ateapipb.Worker{ + WorkerNamespace: pod.Namespace, + WorkerPool: key.pool, + WorkerPod: pod.Name, + Ip: pod.Status.PodIP, + WorkerPodUid: string(pod.UID), + NodeName: pod.Spec.NodeName, + SandboxClass: string(pool.Spec.SandboxClass), + Labels: pool.GetLabels(), + State: ateapipb.Worker_STATE_ACTIVE, + } + // TODO(thockin): for now this is the only place Workers are + // created. If/when this becomes a regular API, validation should + // move there. + if errs := resources.ValidateWorker(worker, nil); len(errs) > 0 { + // Terminal: the inputs are deterministic, retrying cannot help. A + // future pod event re-enqueues the key. + slog.ErrorContext(ctx, "Invalid worker", slog.Any("err", errs.ToAggregate())) + return nil + } + // ErrAlreadyExists means we lost a create race; requeue and converge + // via the update path. + return s.persistence.CreateWorker(ctx, worker) + } + + if w.WorkerPodUid != string(pod.UID) { + // The pod was deleted and recreated under the same name, and the queue + // coalesced the two events. The store row belongs to the dead pod: + // clean it up (releasing any actor bound to the old incarnation) and + // requeue to create a fresh row. + slog.InfoContext(ctx, "Syncer: worker in store belongs to a replaced pod, deleting", slog.String("worker", key.namespace+"/"+key.name)) + if err := s.reconcileDeadWorker(ctx, key.namespace, key.pool, key.name); err != nil { + return err } - slog.ErrorContext(ctx, "Failed to get worker from store", slog.Any("err", err)) - return + return fmt.Errorf("worker %s/%s/%s belonged to replaced pod UID %s; requeueing to recreate", key.namespace, key.pool, key.name, w.WorkerPodUid) } changed := false if w.Ip != pod.Status.PodIP { // TODO: I don't think this is possible, but handling this case so we can log it just in case we can reproduce it. - slog.InfoContext(ctx, "Syncer: updating worker in store (IP changed)", slog.String("worker", pod.Namespace+"/"+pod.Name)) + slog.InfoContext(ctx, "Syncer: updating worker in store (IP changed)", slog.String("worker", key.namespace+"/"+key.name)) w.Ip = pod.Status.PodIP changed = true } if w.SandboxClass != string(pool.Spec.SandboxClass) { - slog.InfoContext(ctx, "Syncer: updating worker in store (SandboxClass changed)", slog.String("worker", pod.Namespace+"/"+pod.Name)) + slog.InfoContext(ctx, "Syncer: updating worker in store (SandboxClass changed)", slog.String("worker", key.namespace+"/"+key.name)) w.SandboxClass = string(pool.Spec.SandboxClass) changed = true } if !maps.Equal(w.Labels, pool.GetLabels()) { - slog.InfoContext(ctx, "Syncer: updating worker in store (labels changed)", slog.String("worker", pod.Namespace+"/"+pod.Name)) + slog.InfoContext(ctx, "Syncer: updating worker in store (labels changed)", slog.String("worker", key.namespace+"/"+key.name)) w.Labels = pool.GetLabels() changed = true } - - if changed { - if err = s.persistence.UpdateWorker(ctx, w, w.Version); err != nil { - slog.ErrorContext(ctx, "Failed to update worker in store", slog.Any("err", err)) - } + if !changed { + return nil } + + // ErrVersionConflict requeues the key; the retry re-fetches the worker at + // its new version. + return s.persistence.UpdateWorker(ctx, w, w.Version) } func isWorkerEligible(pod *corev1.Pod) bool { @@ -198,9 +273,10 @@ func isWorkerEligible(pod *corev1.Pod) bool { } // markWorkerDraining transitions a worker to STATE_DRAINING so the scheduler -// stops routing new actors to it while its pod is Terminating. Best-effort: if -// the worker is already gone or already draining, or a concurrent update wins, -// there is nothing more to do — the Pod Deleted event will clean up the record. +// stops routing new actors to it while its pod is Terminating. If the worker is +// already gone or already draining there is nothing more to do — the Pod +// Deleted event will clean up the record. A version conflict is returned so the +// caller requeues and retries against the updated record. func (s *WorkerPoolSyncer) markWorkerDraining(ctx context.Context, namespace, pool, podName string) error { worker, err := s.persistence.GetWorker(ctx, namespace, pool, podName) if err != nil { @@ -230,47 +306,25 @@ func (s *WorkerPoolSyncer) reconcileDeadWorker(ctx context.Context, namespace, p return s.persistence.DeleteWorker(ctx, namespace, pool, podName) } -// reconcileOrphanedWorkers cleans up stored worker records whose pods no longer -// exist. It runs once after the informer cache has synced, when the indexer is a -// fresh, authoritative snapshot of the live worker pods, so a worker missing -// from the indexer (or present under a different pod UID, i.e. name reuse) is -// stale. -func (s *WorkerPoolSyncer) reconcileOrphanedWorkers(ctx context.Context) { - var workers []*ateapipb.Worker +// enqueueStoredWorkers enqueues a key for every worker record in the store. +// Records whose pods are live and unchanged reconcile to a no-op; orphaned +// records (pod gone, or its name reused by a new pod UID) get cleaned up. +func (s *WorkerPoolSyncer) enqueueStoredWorkers(ctx context.Context) { var pageToken string for { - wPage, nextToken, err := s.persistence.ListWorkers(ctx, 1000, pageToken) + workers, nextToken, err := s.persistence.ListWorkers(ctx, 1000, pageToken) if err != nil { slog.ErrorContext(ctx, "Syncer: failed to list workers for orphan reconcile", slog.Any("err", err)) return } - workers = append(workers, wPage...) + for _, w := range workers { + s.queue.Add(workerKey{namespace: w.GetWorkerNamespace(), pool: w.GetWorkerPool(), name: w.GetWorkerPod()}) + } if nextToken == "" { - break + return } pageToken = nextToken } - indexer := s.workerInformer.GetIndexer() - for _, w := range workers { - key := w.GetWorkerNamespace() + "/" + w.GetWorkerPod() - obj, exists, err := indexer.GetByKey(key) - if err != nil { - slog.ErrorContext(ctx, "Syncer: indexer lookup failed during orphan reconcile", slog.String("worker", key), slog.Any("err", err)) - continue - } - // The pod is still live only if it is present under the same UID the - // worker recorded; a different UID means the name was reused by a new pod - // and this record belongs to a dead incarnation. - if exists { - if pod, ok := obj.(*corev1.Pod); ok && string(pod.UID) == w.GetWorkerPodUid() { - continue - } - } - slog.InfoContext(ctx, "Syncer: reconciling orphaned worker (pod gone)", slog.String("worker", key)) - if err := s.reconcileDeadWorker(ctx, w.GetWorkerNamespace(), w.GetWorkerPool(), w.GetWorkerPod()); err != nil { - slog.ErrorContext(ctx, "Syncer: failed to reconcile orphaned worker", slog.String("worker", key), slog.Any("err", err)) - } - } } // releaseActorOnDeadWorker resets the actor bound to a vanishing worker pod. An From 484f5adf63bc2c6cc11863a0de8d41be022843c6 Mon Sep 17 00:00:00 2001 From: Troy Chiu Date: Fri, 31 Jul 2026 22:31:45 +0000 Subject: [PATCH 2/2] ateapi/controlapi: add unit tests --- cmd/ateapi/internal/controlapi/syncer_test.go | 617 +++++++++++++++++- 1 file changed, 591 insertions(+), 26 deletions(-) diff --git a/cmd/ateapi/internal/controlapi/syncer_test.go b/cmd/ateapi/internal/controlapi/syncer_test.go index f2050ae4a..06342db44 100644 --- a/cmd/ateapi/internal/controlapi/syncer_test.go +++ b/cmd/ateapi/internal/controlapi/syncer_test.go @@ -19,6 +19,7 @@ import ( "errors" "fmt" "maps" + "sync/atomic" "testing" "time" @@ -32,15 +33,24 @@ import ( 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/apimachinery/pkg/util/wait" "k8s.io/client-go/kubernetes/fake" ) // setupSyncerTest sets up a real store with fake Redis and a fake K8s client with informer. func setupSyncerTest(t *testing.T, ctx context.Context, initPools ...*atev1alpha1.WorkerPool) (store.Interface, *fake.Clientset, *atefake.Clientset, func()) { + persistence, fakeK8s, fakeAte, _, cleanup := setupSyncerTestWithStore(t, ctx, nil, initPools...) + return persistence, fakeK8s, fakeAte, cleanup +} + +func setupSyncerTestWithStore(t *testing.T, ctx context.Context, wrapStore func(store.Interface) store.Interface, initPools ...*atev1alpha1.WorkerPool) (store.Interface, *fake.Clientset, *atefake.Clientset, *WorkerPoolSyncer, func()) { t.Helper() persistence, cleanup := storetest.SetupTestStore(t) + if wrapStore != nil { + persistence = wrapStore(persistence) + } fakeK8s := fake.NewSimpleClientset() workerFactory, workerInformer := WorkerPodInformer(fakeK8s) @@ -63,7 +73,7 @@ func setupSyncerTest(t *testing.T, ctx context.Context, initPools ...*atev1alpha workerFactory.WaitForCacheSync(ctx.Done()) ateInformerFactory.WaitForCacheSync(ctx.Done()) - return persistence, fakeK8s, fakeAte, cleanup + return persistence, fakeK8s, fakeAte, syncer, cleanup } func TestSyncer_Lifecycle(t *testing.T) { @@ -105,7 +115,7 @@ func TestSyncer_Lifecycle(t *testing.T) { ObjectMeta: metav1.ObjectMeta{ Name: podName, Namespace: ns, - UID: "08675309-4a65-6e6e-7973-6e756d626572", + UID: "11111111-1111-1111-1111-111111111111", Labels: map[string]string{ workerPodLabel: poolName, }, @@ -222,7 +232,7 @@ func TestSyncer_DeleteBoundWorker_ClearsActor(t *testing.T) { ObjectMeta: metav1.ObjectMeta{ Name: pod, Namespace: ns, - UID: "08675309-4a65-6e6e-7973-6e756d626572", + UID: "11111111-1111-1111-1111-111111111111", Labels: map[string]string{workerPodLabel: pool}, }, Spec: corev1.PodSpec{ @@ -321,7 +331,7 @@ func TestSyncer_OmittedFields(t *testing.T) { ObjectMeta: metav1.ObjectMeta{ Name: podName, Namespace: ns, - UID: "08675309-4a65-6e6e-7973-6e756d626572", + UID: "11111111-1111-1111-1111-111111111111", Labels: map[string]string{ workerPodLabel: poolName, }, @@ -367,6 +377,30 @@ func TestSyncer_OmittedFields(t *testing.T) { } } +// setupReconcileTest builds a syncer whose informer indexer can be seeded +// directly, for tests that drive reconcile synchronously without starting +// factories or worker goroutines. +func setupReconcileTest(t *testing.T, persistence store.Interface, initPools ...*atev1alpha1.WorkerPool) *WorkerPoolSyncer { + t.Helper() + fakeK8s := fake.NewSimpleClientset() + _, workerInformer := WorkerPodInformer(fakeK8s) + + objects := make([]runtime.Object, len(initPools)) + for i, pool := range initPools { + objects[i] = pool + } + //nolint:staticcheck // NewSimpleClientset is the only available fake clientset for versioned CRDs. + fakeAte := atefake.NewSimpleClientset(objects...) + ateInformerFactory := externalversions.NewSharedInformerFactory(fakeAte, 0) + workerPoolLister := ateInformerFactory.Api().V1alpha1().WorkerPools().Lister() + stopCh := make(chan struct{}) + t.Cleanup(func() { close(stopCh) }) + ateInformerFactory.Start(stopCh) + ateInformerFactory.WaitForCacheSync(stopCh) + + return NewWorkerPoolSyncer(persistence, workerInformer, workerPoolLister) +} + // TestSyncer_SoftDelete_MarksDraining verifies that a pod entering Terminating // (DeletionTimestamp set) flips its worker to STATE_DRAINING without deleting the // worker record or touching the bound actor — the actor is still gracefully @@ -375,12 +409,12 @@ func TestSyncer_SoftDelete_MarksDraining(t *testing.T) { ctx := context.Background() persistence, cleanup := storetest.SetupTestStore(t) defer cleanup() - s := &WorkerPoolSyncer{persistence: persistence} + s := setupReconcileTest(t, persistence) ns, pool, pod, ip := "ns-drain", "pool1", "worker-drain", "10.0.0.2" if err := persistence.CreateWorker(ctx, &ateapipb.Worker{ WorkerNamespace: ns, WorkerPool: pool, WorkerPod: pod, Ip: ip, - WorkerPodUid: "08675309-4a65-6e6e-7973-6e756d626572", NodeName: "node1", + WorkerPodUid: "11111111-1111-1111-1111-111111111111", NodeName: "node1", State: ateapipb.Worker_STATE_ACTIVE, }); err != nil { t.Fatalf("create worker: %v", err) @@ -395,7 +429,12 @@ func TestSyncer_SoftDelete_MarksDraining(t *testing.T) { }, Status: corev1.PodStatus{PodIP: ip, PodIPs: []corev1.PodIP{{IP: ip}}}, } - s.syncWorkerToStore(ctx, deleting) + if err := s.workerInformer.GetIndexer().Add(deleting); err != nil { + t.Fatalf("seed indexer: %v", err) + } + if err := s.reconcile(ctx, workerKey{namespace: ns, pool: pool, name: pod}); err != nil { + t.Fatalf("reconcile: %v", err) + } w, err := persistence.GetWorker(ctx, ns, pool, pod) if err != nil { @@ -413,12 +452,12 @@ func TestSyncer_SoftDelete_NoPodIP(t *testing.T) { ctx := context.Background() persistence, cleanup := storetest.SetupTestStore(t) defer cleanup() - s := &WorkerPoolSyncer{persistence: persistence} + s := setupReconcileTest(t, persistence) ns, pool, pod := "ns-drain-noip", "pool1", "worker-drain-noip" if err := persistence.CreateWorker(ctx, &ateapipb.Worker{ WorkerNamespace: ns, WorkerPool: pool, WorkerPod: pod, Ip: "10.0.0.3", - WorkerPodUid: "08675309-4a65-6e6e-7973-6e756d626572", NodeName: "node1", + WorkerPodUid: "11111111-1111-1111-1111-111111111111", NodeName: "node1", State: ateapipb.Worker_STATE_ACTIVE, }); err != nil { t.Fatalf("create worker: %v", err) @@ -434,7 +473,12 @@ func TestSyncer_SoftDelete_NoPodIP(t *testing.T) { // Same object as the draining test above, minus the pod IP. Status: corev1.PodStatus{}, } - s.syncWorkerToStore(ctx, deleting) + if err := s.workerInformer.GetIndexer().Add(deleting); err != nil { + t.Fatalf("seed indexer: %v", err) + } + if err := s.reconcile(ctx, workerKey{namespace: ns, pool: pool, name: pod}); err != nil { + t.Fatalf("reconcile: %v", err) + } w, err := persistence.GetWorker(ctx, ns, pool, pod) if err != nil { @@ -455,7 +499,7 @@ func TestMarkWorkerDraining(t *testing.T) { newWorker := func(state ateapipb.Worker_State) *ateapipb.Worker { return &ateapipb.Worker{ WorkerNamespace: ns, WorkerPool: pool, WorkerPod: pod, Ip: "10.0.0.4", - WorkerPodUid: "08675309-4a65-6e6e-7973-6e756d626572", NodeName: "node1", + WorkerPodUid: "11111111-1111-1111-1111-111111111111", NodeName: "node1", State: state, } } @@ -519,7 +563,7 @@ func TestReconcileDeadWorker(t *testing.T) { } if err := persistence.CreateWorker(ctx, &ateapipb.Worker{ WorkerNamespace: ns, WorkerPool: pool, WorkerPod: pod, Ip: "10.0.0.5", - WorkerPodUid: "08675309-4a65-6e6e-7973-6e756d626572", NodeName: "node1", + WorkerPodUid: "11111111-1111-1111-1111-111111111111", NodeName: "node1", State: ateapipb.Worker_STATE_DRAINING, Assignment: &ateapipb.Assignment{ ActorTemplate: &ateapipb.KubeNamespacedObjectRef{Namespace: ns, Name: "tmpl"}, @@ -555,31 +599,29 @@ func TestSyncer_ReconcileOrphanedWorkers(t *testing.T) { persistence, cleanup := storetest.SetupTestStore(t) defer cleanup() - fakeK8s := fake.NewSimpleClientset() - workerFactory, workerInformer := WorkerPodInformer(fakeK8s) - ns, pool := "ns-recon", "pool1" + workerPool := &atev1alpha1.WorkerPool{ + ObjectMeta: metav1.ObjectMeta{Name: pool, Namespace: ns}, + Spec: atev1alpha1.WorkerPoolSpec{}, + } + s := setupReconcileTest(t, persistence, workerPool) + liveUID := "11111111-1111-1111-1111-111111111111" livePod := &corev1.Pod{ ObjectMeta: metav1.ObjectMeta{ Name: "worker-live", Namespace: ns, - UID: "11111111-1111-1111-1111-111111111111", + UID: types.UID(liveUID), Labels: map[string]string{workerPodLabel: pool}, }, Spec: corev1.PodSpec{NodeName: "node1", Containers: []corev1.Container{{Name: "main", Image: "nginx"}}}, Status: corev1.PodStatus{Phase: corev1.PodRunning, PodIP: "10.0.0.9", PodIPs: []corev1.PodIP{{IP: "10.0.0.9"}}}, } - if _, err := fakeK8s.CoreV1().Pods(ns).Create(ctx, livePod, metav1.CreateOptions{}); err != nil { - t.Fatalf("create live pod: %v", err) + // Seed the indexer so it is an authoritative snapshot of live pods. + if err := s.workerInformer.GetIndexer().Add(livePod); err != nil { + t.Fatalf("seed indexer: %v", err) } - // Sync the informer so its indexer is an authoritative snapshot of live pods. - workerFactory.Start(ctx.Done()) - workerFactory.WaitForCacheSync(ctx.Done()) - - s := &WorkerPoolSyncer{persistence: persistence, workerInformer: workerInformer} - // A worker whose pod is live must be preserved. if err := persistence.CreateWorker(ctx, &ateapipb.Worker{ WorkerNamespace: ns, WorkerPool: pool, WorkerPod: "worker-live", Ip: "10.0.0.9", @@ -609,7 +651,12 @@ func TestSyncer_ReconcileOrphanedWorkers(t *testing.T) { t.Fatalf("create orphan worker: %v", err) } - s.reconcileOrphanedWorkers(ctx) + // Drive the startup pass synchronously: enqueue all stored workers and + // drain the queue in this goroutine. + s.enqueueStoredWorkers(ctx) + for s.queue.Len() > 0 { + s.processNextWorkItem(ctx) + } if _, err := persistence.GetWorker(ctx, ns, pool, "worker-orphan"); !errors.Is(err, store.ErrNotFound) { t.Errorf("orphan worker not removed: err=%v", err) @@ -659,7 +706,7 @@ func TestReleaseActorOnDeadWorker_StatusTransitions(t *testing.T) { } if err := persistence.CreateWorker(ctx, &ateapipb.Worker{ WorkerNamespace: ns, WorkerPool: pool, WorkerPod: pod, Ip: ip, - WorkerPodUid: "08675309-4a65-6e6e-7973-6e756d626572", NodeName: "node1", + WorkerPodUid: "11111111-1111-1111-1111-111111111111", NodeName: "node1", State: ateapipb.Worker_STATE_ACTIVE, Assignment: &ateapipb.Assignment{ ActorTemplate: &ateapipb.KubeNamespacedObjectRef{Namespace: ns, Name: "tmpl"}, @@ -686,3 +733,521 @@ func TestReleaseActorOnDeadWorker_StatusTransitions(t *testing.T) { }) } } + +type conflictStore struct { + store.Interface + conflictTriggered atomic.Bool + onUpdate func(ctx context.Context, worker *ateapipb.Worker) +} + +func (c *conflictStore) UpdateWorker(ctx context.Context, worker *ateapipb.Worker, expectedVersion int64) error { + if c.onUpdate != nil && c.conflictTriggered.CompareAndSwap(false, true) { + c.onUpdate(ctx, worker) + } + return c.Interface.UpdateWorker(ctx, worker, expectedVersion) +} + +func TestSyncer_UpdateWorker_RetryOnVersionConflict(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + + ns := "ns-syncer-conflict" + podName := "worker-unit-conflict" + poolName := "pool-conflict" + + pool := &atev1alpha1.WorkerPool{ + ObjectMeta: metav1.ObjectMeta{ + Name: poolName, + Namespace: ns, + Labels: map[string]string{"foo": "bar"}, + }, + Spec: atev1alpha1.WorkerPoolSpec{ + SandboxClass: "gvisor", + }, + } + + var cs *conflictStore + persistence, fakeK8s, fakeAte, syncer, cleanup := setupSyncerTestWithStore(t, ctx, func(s store.Interface) store.Interface { + cs = &conflictStore{Interface: s} + return cs + }, pool) + defer func() { + // Stop syncer before closing store to prevent panics on closed miniredis. + cancel() + cleanup() + }() + + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: podName, + Namespace: ns, + UID: "11111111-1111-1111-1111-111111111111", + Labels: map[string]string{ + workerPodLabel: poolName, + }, + }, + Spec: corev1.PodSpec{ + NodeName: "node1", + Containers: []corev1.Container{{Name: "main", Image: "nginx"}}, + }, + Status: corev1.PodStatus{ + Phase: corev1.PodRunning, + PodIP: "10.0.0.1", + PodIPs: []corev1.PodIP{{IP: "10.0.0.1"}}, + }, + } + + _, err := fakeK8s.CoreV1().Pods(ns).Create(context.Background(), pod, metav1.CreateOptions{}) + if err != nil { + t.Fatalf("failed to create pod: %v", err) + } + + err = wait.PollUntilContextTimeout(context.Background(), 100*time.Millisecond, 5*time.Second, true, func(ctx context.Context) (bool, error) { + w, err := persistence.GetWorker(ctx, ns, poolName, podName) + if err != nil { + if errors.Is(err, store.ErrNotFound) { + return false, nil + } + return false, err + } + return w.Ip == "10.0.0.1", nil + }) + if err != nil { + t.Fatalf("Worker state check failed: %v", err) + } + + // Update the pool SandboxClass in K8s (a mutable worker field). + updatedPool, err := fakeAte.ApiV1alpha1().WorkerPools(ns).Get(context.Background(), poolName, metav1.GetOptions{}) + if err != nil { + t.Fatalf("failed to get pool: %v", err) + } + updatedPool.Spec.SandboxClass = "microvm" + if _, err := fakeAte.ApiV1alpha1().WorkerPools(ns).Update(context.Background(), updatedPool, metav1.UpdateOptions{}); err != nil { + t.Fatalf("failed to update pool: %v", err) + } + + // Wait until the WorkerPool informer cache reflects the updated SandboxClass before triggering the pod update. + err = wait.PollUntilContextTimeout(context.Background(), 50*time.Millisecond, 5*time.Second, true, func(ctx context.Context) (bool, error) { + p, err := syncer.workerPoolLister.WorkerPools(ns).Get(poolName) + if err != nil { + return false, nil + } + return p.Spec.SandboxClass == "microvm", nil + }) + if err != nil { + t.Fatalf("pool informer cache failed to update: %v", err) + } + + // Configure conflictStore to inject a concurrent version bump in Redis when the syncer calls UpdateWorker. + cs.onUpdate = func(c context.Context, w *ateapipb.Worker) { + if cw, err := cs.Interface.GetWorker(c, ns, poolName, podName); err == nil { + cw.NodeName = "node2" + _ = cs.Interface.UpdateWorker(c, cw, cw.Version) + } + } + + // Touch the pod ONCE in K8s so the syncer reconciles it. The first reconcile's + // UpdateWorker hits ErrVersionConflict (injected by conflictStore), which requeues + // the key with backoff; the retry re-fetches the latest version from Redis. + updatedPod, err := fakeK8s.CoreV1().Pods(ns).Get(context.Background(), podName, metav1.GetOptions{}) + if err != nil { + t.Fatalf("failed to get pod: %v", err) + } + if updatedPod.Annotations == nil { + updatedPod.Annotations = make(map[string]string) + } + updatedPod.Annotations["trigger"] = "update" + if _, err := fakeK8s.CoreV1().Pods(ns).Update(context.Background(), updatedPod, metav1.UpdateOptions{}); err != nil { + t.Fatalf("failed to update pod: %v", err) + } + + // Verify that the worker in Redis eventually gets updated to the new SandboxClass despite the version conflict. + err = wait.PollUntilContextTimeout(context.Background(), 100*time.Millisecond, 5*time.Second, true, func(ctx context.Context) (bool, error) { + w, err := persistence.GetWorker(ctx, ns, poolName, podName) + if err != nil { + if errors.Is(err, store.ErrNotFound) { + return false, nil + } + return false, err + } + return w.SandboxClass == "microvm" && w.NodeName == "node2", nil + }) + if err != nil { + t.Fatalf("Worker failed to update SandboxClass after version conflict: %v", err) + } +} + +// TestSyncer_RequeueOnMissingWorkerPool verifies that a pod whose WorkerPool is +// not yet in the lister cache is requeued rather than dropped, and converges +// once the pool appears. +func TestSyncer_RequeueOnMissingWorkerPool(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + + ns := "ns-syncer-latepool" + podName := "worker-late-1" + poolName := "pool-late" + + persistence, fakeK8s, fakeAte, syncer, cleanup := setupSyncerTestWithStore(t, ctx, nil) // no pools yet + defer func() { + // Stop syncer before closing store to prevent panics on closed miniredis. + cancel() + cleanup() + }() + + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: podName, + Namespace: ns, + UID: "11111111-1111-1111-1111-111111111111", + Labels: map[string]string{ + workerPodLabel: poolName, + }, + }, + Spec: corev1.PodSpec{ + NodeName: "node1", + Containers: []corev1.Container{{Name: "main", Image: "nginx"}}, + }, + Status: corev1.PodStatus{ + Phase: corev1.PodRunning, + PodIP: "10.0.0.5", + PodIPs: []corev1.PodIP{{IP: "10.0.0.5"}}, + }, + } + if _, err := fakeK8s.CoreV1().Pods(ns).Create(context.Background(), pod, metav1.CreateOptions{}); err != nil { + t.Fatalf("failed to create pod: %v", err) + } + + // Wait until the syncer has attempted to reconcile the pod and requeued it due to the missing pool. + key := workerKey{namespace: ns, pool: poolName, name: podName} + err := wait.PollUntilContextTimeout(context.Background(), 10*time.Millisecond, 5*time.Second, true, func(ctx context.Context) (bool, error) { + return syncer.queue.NumRequeues(key) > 0, nil + }) + if err != nil { + t.Fatalf("syncer did not requeue pod on missing WorkerPool: %v", err) + } + + pool := &atev1alpha1.WorkerPool{ + ObjectMeta: metav1.ObjectMeta{ + Name: poolName, + Namespace: ns, + }, + Spec: atev1alpha1.WorkerPoolSpec{ + SandboxClass: "gvisor", + }, + } + if _, err := fakeAte.ApiV1alpha1().WorkerPools(ns).Create(context.Background(), pool, metav1.CreateOptions{}); err != nil { + t.Fatalf("failed to create pool: %v", err) + } + + err = wait.PollUntilContextTimeout(context.Background(), 100*time.Millisecond, 5*time.Second, true, func(ctx context.Context) (bool, error) { + w, err := persistence.GetWorker(ctx, ns, poolName, podName) + if err != nil { + if errors.Is(err, store.ErrNotFound) { + return false, nil + } + return false, err + } + return w.SandboxClass == "gvisor", nil + }) + if err != nil { + t.Fatalf("Worker not created after pool appeared: %v", err) + } +} + +// TestSyncer_SoftDelete_ViaInformer verifies end-to-end (through informer events +// and the queue) that a pod entering graceful termination flips its worker to +// STATE_DRAINING without deleting the record. +func TestSyncer_SoftDelete_ViaInformer(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + + ns := "ns-syncer-softdelete" + podName := "worker-soft-1" + poolName := "pool1" + + pool := &atev1alpha1.WorkerPool{ + ObjectMeta: metav1.ObjectMeta{ + Name: poolName, + Namespace: ns, + }, + Spec: atev1alpha1.WorkerPoolSpec{ + SandboxClass: "gvisor", + }, + } + + persistence, fakeK8s, _, cleanup := setupSyncerTest(t, ctx, pool) + defer func() { + // Stop syncer before closing store to prevent panics on closed miniredis. + cancel() + cleanup() + }() + + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: podName, + Namespace: ns, + UID: "11111111-1111-1111-1111-111111111111", + Labels: map[string]string{ + workerPodLabel: poolName, + }, + }, + Spec: corev1.PodSpec{ + NodeName: "node1", + Containers: []corev1.Container{{Name: "main", Image: "nginx"}}, + }, + Status: corev1.PodStatus{ + Phase: corev1.PodRunning, + PodIP: "10.0.0.6", + PodIPs: []corev1.PodIP{{IP: "10.0.0.6"}}, + }, + } + if _, err := fakeK8s.CoreV1().Pods(ns).Create(context.Background(), pod, metav1.CreateOptions{}); err != nil { + t.Fatalf("failed to create pod: %v", err) + } + + if err := wait.PollUntilContextTimeout(context.Background(), 50*time.Millisecond, 2*time.Second, true, func(ctx context.Context) (bool, error) { + _, err := persistence.GetWorker(ctx, ns, poolName, podName) + return err == nil, nil + }); err != nil { + t.Fatalf("worker row not materialised: %v", err) + } + + // The fake clientset stores updates verbatim, so setting DeletionTimestamp + // simulates a pod in graceful termination that is still in the cache. + deleting := pod.DeepCopy() + now := metav1.Now() + deleting.DeletionTimestamp = &now + if _, err := fakeK8s.CoreV1().Pods(ns).Update(context.Background(), deleting, metav1.UpdateOptions{}); err != nil { + t.Fatalf("failed to update pod: %v", err) + } + + if err := wait.PollUntilContextTimeout(context.Background(), 100*time.Millisecond, 2*time.Second, true, func(ctx context.Context) (bool, error) { + w, err := persistence.GetWorker(ctx, ns, poolName, podName) + if err != nil { + return false, err + } + return w.GetState() == ateapipb.Worker_STATE_DRAINING, nil + }); err != nil { + t.Fatalf("Worker not marked DRAINING after DeletionTimestamp set: %v", err) + } +} + +// TestSyncer_PodRecreatedWithNewUID verifies that when a pod is deleted and +// recreated under the same name, the store row converges to the new pod's UID +// and IP even if the queue coalesces the delete and add events. +func TestSyncer_PodRecreatedWithNewUID(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + + ns := "ns-syncer-recreate" + podName := "worker-recreate-1" + poolName := "pool1" + + pool := &atev1alpha1.WorkerPool{ + ObjectMeta: metav1.ObjectMeta{ + Name: poolName, + Namespace: ns, + }, + Spec: atev1alpha1.WorkerPoolSpec{ + SandboxClass: "gvisor", + }, + } + + persistence, fakeK8s, _, cleanup := setupSyncerTest(t, ctx, pool) + defer func() { + // Stop syncer before closing store to prevent panics on closed miniredis. + cancel() + cleanup() + }() + + makePod := func(uid, ip string) *corev1.Pod { + return &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: podName, + Namespace: ns, + UID: types.UID(uid), + Labels: map[string]string{ + workerPodLabel: poolName, + }, + }, + Spec: corev1.PodSpec{ + NodeName: "node1", + Containers: []corev1.Container{{Name: "main", Image: "nginx"}}, + }, + Status: corev1.PodStatus{ + Phase: corev1.PodRunning, + PodIP: ip, + PodIPs: []corev1.PodIP{{IP: ip}}, + }, + } + } + + oldUID := "11111111-1111-1111-1111-111111111111" + newUID := "22222222-2222-2222-2222-222222222222" + + if _, err := fakeK8s.CoreV1().Pods(ns).Create(context.Background(), makePod(oldUID, "10.0.0.7"), metav1.CreateOptions{}); err != nil { + t.Fatalf("failed to create pod: %v", err) + } + if err := wait.PollUntilContextTimeout(context.Background(), 50*time.Millisecond, 2*time.Second, true, func(ctx context.Context) (bool, error) { + w, err := persistence.GetWorker(ctx, ns, poolName, podName) + if err != nil { + if errors.Is(err, store.ErrNotFound) { + return false, nil + } + return false, err + } + return w.WorkerPodUid == oldUID, nil + }); err != nil { + t.Fatalf("worker row not materialised: %v", err) + } + + // Update the pod in K8s directly to the new UID (simulating coalesced Delete+Create events where + // the syncer sees the new Pod incarnation while the old UID record is still in the store). + if _, err := fakeK8s.CoreV1().Pods(ns).Update(context.Background(), makePod(newUID, "10.0.0.8"), metav1.UpdateOptions{}); err != nil { + t.Fatalf("failed to update pod with new UID: %v", err) + } + + if err := wait.PollUntilContextTimeout(context.Background(), 100*time.Millisecond, 5*time.Second, true, func(ctx context.Context) (bool, error) { + w, err := persistence.GetWorker(ctx, ns, poolName, podName) + if err != nil { + if errors.Is(err, store.ErrNotFound) { + return false, nil + } + return false, err + } + return w.WorkerPodUid == newUID && w.Ip == "10.0.0.8", nil + }); err != nil { + t.Fatalf("Worker row did not converge to recreated pod: %v", err) + } + + // Verify via ListWorkers that no stale worker record with oldUID remains in the store. + workers, _, err := persistence.ListWorkers(context.Background(), 100, "") + if err != nil { + t.Fatalf("failed to list workers: %v", err) + } + if len(workers) != 1 { + t.Fatalf("expected exactly 1 worker in store, got %d", len(workers)) + } + if workers[0].GetWorkerPodUid() != newUID { + t.Fatalf("expected worker in store to have UID %q, got %q", newUID, workers[0].GetWorkerPodUid()) + } +} + +// TestSyncer_DeleteNeverEligiblePod verifies that deleting a pod that never got +// an IP (and thus never had a store row) is a no-op and does not error-loop. +func TestSyncer_DeleteNeverEligiblePod(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + + ns := "ns-syncer-neverip" + podName := "worker-noip-1" + poolName := "pool1" + + pool := &atev1alpha1.WorkerPool{ + ObjectMeta: metav1.ObjectMeta{ + Name: poolName, + Namespace: ns, + }, + Spec: atev1alpha1.WorkerPoolSpec{ + SandboxClass: "gvisor", + }, + } + + persistence, fakeK8s, _, cleanup := setupSyncerTest(t, ctx, pool) + defer func() { + // Stop syncer before closing store to prevent panics on closed miniredis. + cancel() + cleanup() + }() + + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: podName, + Namespace: ns, + UID: "11111111-1111-1111-1111-111111111111", + Labels: map[string]string{ + workerPodLabel: poolName, + }, + }, + Spec: corev1.PodSpec{ + NodeName: "node1", + Containers: []corev1.Container{{Name: "main", Image: "nginx"}}, + }, + } + if _, err := fakeK8s.CoreV1().Pods(ns).Create(context.Background(), pod, metav1.CreateOptions{}); err != nil { + t.Fatalf("failed to create pod: %v", err) + } + if err := fakeK8s.CoreV1().Pods(ns).Delete(context.Background(), podName, metav1.DeleteOptions{}); err != nil { + t.Fatalf("failed to delete pod: %v", err) + } + + // The store must stay empty throughout: the pod never had an IP so no row + // was created, and the delete path is an idempotent no-op. + err := wait.PollUntilContextTimeout(context.Background(), 50*time.Millisecond, 500*time.Millisecond, true, func(ctx context.Context) (bool, error) { + workers, _, err := persistence.ListWorkers(ctx, 1000, "") + if err != nil { + return false, err + } + if len(workers) != 0 { + return false, fmt.Errorf("expected 0 workers, got %d", len(workers)) + } + return false, nil // keep polling until timeout + }) + if err != nil && !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("store did not stay empty: %v", err) + } +} + +// TestSyncer_InvalidWorkerIsTerminal drives reconcile directly and verifies +// that a validation failure is terminal (nil error, no requeue) and writes +// nothing to the store. +func TestSyncer_InvalidWorkerIsTerminal(t *testing.T) { + ctx := context.Background() + + ns := "ns-syncer-invalid" + podName := "worker-invalid-1" + poolName := "pool1" + + persistence, cleanup := storetest.SetupTestStore(t) + defer cleanup() + + pool := &atev1alpha1.WorkerPool{ + ObjectMeta: metav1.ObjectMeta{ + Name: poolName, + Namespace: ns, + }, + Spec: atev1alpha1.WorkerPoolSpec{ + SandboxClass: "gvisor", + }, + } + s := setupReconcileTest(t, persistence, pool) + + // Seed the indexer directly (no factories started, no workers running) so + // reconcile can be driven synchronously. + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: podName, + Namespace: ns, + UID: "not-a-valid-uuid", // fails ValidateWorker + Labels: map[string]string{ + workerPodLabel: poolName, + }, + }, + Spec: corev1.PodSpec{ + NodeName: "node1", + Containers: []corev1.Container{{Name: "main", Image: "nginx"}}, + }, + Status: corev1.PodStatus{ + Phase: corev1.PodRunning, + PodIP: "10.0.0.9", + PodIPs: []corev1.PodIP{{IP: "10.0.0.9"}}, + }, + } + if err := s.workerInformer.GetIndexer().Add(pod); err != nil { + t.Fatalf("failed to seed indexer: %v", err) + } + + key := workerKey{namespace: ns, pool: poolName, name: podName} + if err := s.reconcile(ctx, key); err != nil { + t.Fatalf("reconcile returned error for invalid worker (should be terminal): %v", err) + } + if _, err := persistence.GetWorker(ctx, ns, poolName, podName); !errors.Is(err, store.ErrNotFound) { + t.Fatalf("expected no worker row for invalid worker, got err=%v", err) + } +}