From b7d08338014c05b8cb64821dc73d298f5673e8db Mon Sep 17 00:00:00 2001 From: Luke Van Drie Date: Fri, 17 Jul 2026 11:20:58 -0700 Subject: [PATCH 1/4] fix(datastore): surface dropped endpoint registrations and requeue pod reconciles An upsert that overlaps an in-flight delete (or whose collector fails to start) observes a nil endpoint from NewEndpoint with no surviving pods-map entry. Previously this was silently swallowed: the pod stayed untracked until the next pod event, which can leave the flow control dispatch cycle with an empty candidate list and saturation pinned at 1.0. upsertEndpoint now distinguishes the benign duplicate-start race (a concurrent upsert already stored the entry) from a dropped registration, which is returned as an error. PodReconciler propagates it so controller-runtime requeues with backoff; podResyncAll propagates it through PoolSet for the same effect on the pool reconciler; discovery sources log it. Related to #2060. Signed-off-by: Luke Van Drie --- pkg/epp/controller/pod_reconciler.go | 17 ++-- pkg/epp/controller/pod_reconciler_test.go | 47 +++++++++ pkg/epp/datastore/datastore.go | 86 +++++++++++----- pkg/epp/datastore/datastore_test.go | 114 ++++++++++++++++++++++ 4 files changed, 234 insertions(+), 30 deletions(-) diff --git a/pkg/epp/controller/pod_reconciler.go b/pkg/epp/controller/pod_reconciler.go index 2d0d32c049..527cdee701 100644 --- a/pkg/epp/controller/pod_reconciler.go +++ b/pkg/epp/controller/pod_reconciler.go @@ -59,7 +59,12 @@ func (c *PodReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.R return ctrl.Result{}, fmt.Errorf("unable to get pod - %w", err) } - c.updateDatastore(ctx, pod) + // Returning the error makes controller-runtime requeue with backoff, so an endpoint + // registration dropped by an upsert racing an in-flight delete is retried instead of + // leaving the pod untracked until the next pod event. + if err := c.updateDatastore(ctx, pod); err != nil { + return ctrl.Result{}, fmt.Errorf("failed to update datastore for pod %s - %w", req.NamespacedName, err) + } return ctrl.Result{}, nil } @@ -91,16 +96,12 @@ func (c *PodReconciler) SetupWithManager(mgr ctrl.Manager) error { Complete(c) } -func (c *PodReconciler) updateDatastore(ctx context.Context, pod *corev1.Pod) { +func (c *PodReconciler) updateDatastore(ctx context.Context, pod *corev1.Pod) error { logger := log.FromContext(ctx) if !podutil.IsPodReady(pod) || !c.Datastore.PoolLabelsMatch(pod.Labels) { logger.V(logutil.DEBUG).Info("Pod removed or not added") c.Datastore.PodDelete(pod.Name) - } else { - if c.Datastore.PodUpdateOrAddIfNotExist(ctx, pod) { - logger.V(logutil.DEFAULT).Info("Pod added") - } else { - logger.V(logutil.DEFAULT).Info("Pod already exists") - } + return nil } + return c.Datastore.PodUpdateOrAddIfNotExist(ctx, pod) } diff --git a/pkg/epp/controller/pod_reconciler_test.go b/pkg/epp/controller/pod_reconciler_test.go index a8ab97556e..3aad1e7159 100644 --- a/pkg/epp/controller/pod_reconciler_test.go +++ b/pkg/epp/controller/pod_reconciler_test.go @@ -35,6 +35,7 @@ import ( "github.com/llm-d/llm-d-router/pkg/epp/datalayer" "github.com/llm-d/llm-d-router/pkg/epp/datastore" + fwkdl "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/datalayer" "github.com/llm-d/llm-d-router/pkg/epp/util/pool" testutil "github.com/llm-d/llm-d-router/pkg/epp/util/testing" ) @@ -229,3 +230,49 @@ func TestPodReconciler(t *testing.T) { } } } + +// nilEndpointFactory always fails endpoint registration, standing in for a collector that is +// still registered for the endpoint (upsert overlapping an in-flight delete) or fails to start. +type nilEndpointFactory struct{} + +func (nilEndpointFactory) NewEndpoint(_ context.Context, _ *fwkdl.EndpointMetadata) fwkdl.Endpoint { + return nil +} + +func (nilEndpointFactory) UpdateEndpoint(_ context.Context, _ fwkdl.Endpoint) {} + +func (nilEndpointFactory) ReleaseEndpoint(_ fwkdl.Endpoint) {} + +// TestPodReconciler_ErrorsOnRegistrationDrop verifies that Reconcile surfaces a dropped endpoint +// registration as an error so controller-runtime requeues the pod, rather than leaving it +// untracked until the next pod event (#2060). +func TestPodReconciler_ErrorsOnRegistrationDrop(t *testing.T) { + scheme := runtime.NewScheme() + _ = clientgoscheme.AddToScheme(scheme) + + incomingPod := testutil.FromBase(basePod1). + Labels(map[string]string{"some-key": "some-val"}). + ReadyCondition().ObjRef() + fakeClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(incomingPod).Build() + + testPool := &v1.InferencePool{ + Spec: v1.InferencePoolSpec{ + TargetPorts: []v1.Port{{Number: v1.PortNumber(int32(8000))}}, + Selector: v1.LabelSelector{ + MatchLabels: map[v1.LabelKey]v1.LabelValue{"some-key": "some-val"}, + }, + }, + } + store := datastore.NewDatastore(t.Context(), nilEndpointFactory{}) + _ = store.PoolSet(t.Context(), fakeClient, pool.InferencePoolToEndpointPool(testPool)) + + podReconciler := &PodReconciler{Reader: fakeClient, Datastore: store} + req := ctrl.Request{NamespacedName: types.NamespacedName{Name: incomingPod.Name, Namespace: incomingPod.Namespace}} + + if _, err := podReconciler.Reconcile(context.Background(), req); err == nil { + t.Error("expected Reconcile to return an error for a dropped endpoint registration, got nil") + } + if pods := store.PodList(datastore.AllPodsPredicate); len(pods) != 0 { + t.Errorf("expected no pods in datastore, got %d", len(pods)) + } +} diff --git a/pkg/epp/datastore/datastore.go b/pkg/epp/datastore/datastore.go index b7260474a4..c34c3764ef 100644 --- a/pkg/epp/datastore/datastore.go +++ b/pkg/epp/datastore/datastore.go @@ -88,10 +88,15 @@ type Datastore interface { // PodList lists pods matching the given predicate. PodList(predicate func(fwkdl.Endpoint) bool) []fwkdl.Endpoint - PodUpdateOrAddIfNotExist(ctx context.Context, pod *corev1.Pod) bool + // PodUpdateOrAddIfNotExist stores or updates the endpoints for the given pod. It returns an + // error when an endpoint registration was dropped (see upsertEndpoint); the pod is then not + // tracked by the datastore and the caller must retry (e.g. by requeueing the reconcile). + PodUpdateOrAddIfNotExist(ctx context.Context, pod *corev1.Pod) error PodDelete(podName string) // EndpointUpsert adds or updates an endpoint from a non-Kubernetes discovery source. + // A dropped registration is logged; the endpoint stays untracked until the discovery + // source re-emits it. EndpointUpsert(ctx context.Context, meta *fwkdl.EndpointMetadata) // EndpointDelete removes the endpoint with the given namespaced name. EndpointDelete(id types.NamespacedName) @@ -282,7 +287,7 @@ func (ds *datastore) PodList(predicate func(fwkdl.Endpoint) bool) []fwkdl.Endpoi return res } -func (ds *datastore) PodUpdateOrAddIfNotExist(ctx context.Context, pod *corev1.Pod) bool { +func (ds *datastore) PodUpdateOrAddIfNotExist(ctx context.Context, pod *corev1.Pod) error { // Take a reference to pool under read lock to avoid racing with PoolSet(). // This is safe because PoolSet() replaces the entire pool struct rather than // updating it in-place. @@ -290,15 +295,27 @@ func (ds *datastore) PodUpdateOrAddIfNotExist(ctx context.Context, pod *corev1.P pool := ds.pool ds.mu.RUnlock() - return ds.podUpdateOrAddIfNotExist(ctx, pod, pool) + added, err := ds.podUpdateOrAddIfNotExist(ctx, pod, pool) + if err != nil { + return err + } + logger := log.FromContext(ctx) + if added { + logger.V(logutil.DEFAULT).Info("Pod added", "name", pod.Name) + } else { + logger.V(logutil.DEFAULT).Info("Pod already exists", "name", pod.Name) + } + return nil } // podUpdateOrAddIfNotExist is the lock-free inner implementation. // Callers must ensure pool is a consistent snapshot (either read under lock // or already held, as in podResyncAll which runs under ds.mu.Lock via PoolSet). -func (ds *datastore) podUpdateOrAddIfNotExist(ctx context.Context, pod *corev1.Pod, pool *datalayer.EndpointPool) bool { +// It returns whether any endpoint was newly created, and a joined error covering every +// endpoint of the pod whose registration was dropped. +func (ds *datastore) podUpdateOrAddIfNotExist(ctx context.Context, pod *corev1.Pod, pool *datalayer.EndpointPool) (bool, error) { if pool == nil { - return true + return false, nil } labels := make(map[string]string, len(pod.GetLabels())) @@ -328,12 +345,18 @@ func (ds *datastore) podUpdateOrAddIfNotExist(ctx context.Context, pod *corev1.P "pod", pod.Name, "namespace", pod.Namespace, "targetPorts", pool.TargetPorts) } - result := true + added := false + var errs []error existingEpSet := sets.Set[types.NamespacedName]{} for _, endpointMetadata := range pods { existingEpSet.Insert(endpointMetadata.NamespacedName) - if ds.upsertEndpoint(ctx, endpointMetadata) { - result = false + created, err := ds.upsertEndpoint(ctx, endpointMetadata) + if err != nil { + errs = append(errs, err) + continue + } + if created { + added = true } } @@ -350,7 +373,7 @@ func (ds *datastore) podUpdateOrAddIfNotExist(ctx context.Context, pod *corev1.P } } - return result + return added, errors.Join(errs...) } func (ds *datastore) PodDelete(podName string) { @@ -365,7 +388,11 @@ func (ds *datastore) PodDelete(podName string) { } func (ds *datastore) EndpointUpsert(ctx context.Context, meta *fwkdl.EndpointMetadata) { - ds.upsertEndpoint(ctx, meta) + if _, err := ds.upsertEndpoint(ctx, meta); err != nil { + // Discovery sources have no requeue mechanism, so surface the drop loudly; the endpoint + // stays untracked until the source re-emits it. + log.FromContext(ctx).Error(err, "failed to register endpoint", "endpoint", meta.NamespacedName) + } } func (ds *datastore) EndpointDelete(id types.NamespacedName) { @@ -375,29 +402,37 @@ func (ds *datastore) EndpointDelete(id types.NamespacedName) { } // upsertEndpoint stores or updates a single endpoint in the pods map. -// Returns true if the endpoint was newly created, false if it already existed -// or if NewEndpoint returned nil (duplicate-start race). +// Returns true if the endpoint was newly created, false if it already existed. // Shared by EndpointUpsert and podUpdateOrAddIfNotExist. -func (ds *datastore) upsertEndpoint(ctx context.Context, meta *fwkdl.EndpointMetadata) bool { +// +// It returns an error when the endpoint's registration was dropped: NewEndpoint returns nil when +// a collector is already registered for this endpoint or its collector failed to start. If a +// concurrent upsert has stored the entry, the datastore is consistent and the nil is benign. But +// when the pods map has no entry either — the upsert overlapped an in-flight delete that removed +// the entry before deregistering the collector, or the collector failed to start — the endpoint +// is untracked and the caller must retry. +func (ds *datastore) upsertEndpoint(ctx context.Context, meta *fwkdl.EndpointMetadata) (bool, error) { existing, ok := ds.pods.Load(meta.NamespacedName) if !ok { ep := ds.epf.NewEndpoint(ds.parentCtx, meta) if ep == nil { - // NewEndpoint returns nil when a collector is already running for this - // endpoint (duplicate reconcile race). The existing entry in ds.pods - // is still valid; skip re-registering it. - return false + if _, ok := ds.pods.Load(meta.NamespacedName); ok { + return false, nil + } + return false, fmt.Errorf( + "endpoint %s registration dropped: collector already registered or failed to start", + meta.NamespacedName) } ds.pods.Store(meta.NamespacedName, ep) - return true + return true, nil } ep := existing.(fwkdl.Endpoint) if ep.GetMetadata().Equal(meta) { - return false + return false, nil } ep.UpdateMetadata(meta) ds.epf.UpdateEndpoint(ctx, ep) - return false + return false, nil } func (ds *datastore) podResyncAll(ctx context.Context, reader client.Reader) error { @@ -413,6 +448,7 @@ func (ds *datastore) podResyncAll(ctx context.Context, reader client.Reader) err // Track active endpoints by their full name (including rank suffix). // This ensures orphaned rank endpoints are removed when targetPorts shrinks. activeEndpoints := sets.New[types.NamespacedName]() + var errs []error for _, pod := range podList.Items { if !podutil.IsPodReady(&pod) { continue @@ -422,7 +458,13 @@ func (ds *datastore) podResyncAll(ctx context.Context, reader client.Reader) err for idx := range ds.pool.TargetPorts { activeEndpoints.Insert(createEndpointNamespacedName(&pod, idx)) } - if !ds.podUpdateOrAddIfNotExist(ctx, &pod, ds.pool) { + added, err := ds.podUpdateOrAddIfNotExist(ctx, &pod, ds.pool) + if err != nil { + // Propagate so PoolSet fails and the pool reconciler requeues the resync. + errs = append(errs, err) + continue + } + if added { logger.V(logutil.DEFAULT).Info("Pod added", "name", namespacedName) } else { logger.V(logutil.DEFAULT).Info("Pod already exists", "name", namespacedName) @@ -441,7 +483,7 @@ func (ds *datastore) podResyncAll(ctx context.Context, reader client.Reader) err return true }) - return nil + return errors.Join(errs...) } // extractActivePorts extracts the active ports from a pod's annotations. diff --git a/pkg/epp/datastore/datastore_test.go b/pkg/epp/datastore/datastore_test.go index 7492ff31a1..6d051af7b5 100644 --- a/pkg/epp/datastore/datastore_test.go +++ b/pkg/epp/datastore/datastore_test.go @@ -29,6 +29,7 @@ import ( "github.com/google/go-cmp/cmp" "github.com/google/go-cmp/cmp/cmpopts" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/labels" @@ -1454,6 +1455,119 @@ func TestEndpointDelete_Missing(t *testing.T) { }) } +// storeThenNilFactory simulates the benign duplicate-start race: a concurrent upsert stores the +// endpoint between the caller's pods-map miss and NewEndpoint returning nil. +type storeThenNilFactory struct { + ds *datastore +} + +func (f *storeThenNilFactory) NewEndpoint(_ context.Context, meta *fwkdl.EndpointMetadata) fwkdl.Endpoint { + f.ds.pods.Store(meta.NamespacedName, fwkdl.NewEndpoint(meta, fwkdl.NewMetrics())) + return nil +} + +func (f *storeThenNilFactory) UpdateEndpoint(_ context.Context, _ fwkdl.Endpoint) {} + +func (f *storeThenNilFactory) ReleaseEndpoint(_ fwkdl.Endpoint) {} + +// collectorRegistryFactory mimics Runtime's collector registry semantics: NewEndpoint returns nil +// while a collector is still registered for the endpoint, and ReleaseEndpoint deregisters only +// after a delay, mirroring the delete-event dispatch that runs before collector removal in +// Runtime.ReleaseEndpoint. An upsert landing in that window observes a pods-map miss and a nil +// endpoint at once — the dropped-registration case. +type collectorRegistryFactory struct { + mu sync.Mutex + registered map[types.NamespacedName]bool +} + +func (f *collectorRegistryFactory) NewEndpoint(_ context.Context, meta *fwkdl.EndpointMetadata) fwkdl.Endpoint { + f.mu.Lock() + defer f.mu.Unlock() + if f.registered[meta.NamespacedName] { + return nil + } + f.registered[meta.NamespacedName] = true + return fwkdl.NewEndpoint(meta, fwkdl.NewMetrics()) +} + +func (f *collectorRegistryFactory) UpdateEndpoint(_ context.Context, _ fwkdl.Endpoint) {} + +func (f *collectorRegistryFactory) ReleaseEndpoint(ep fwkdl.Endpoint) { + time.Sleep(50 * time.Microsecond) + f.mu.Lock() + defer f.mu.Unlock() + delete(f.registered, ep.GetMetadata().NamespacedName) +} + +// TestPodUpdateOrAddIfNotExist_RegistrationDropReturnsError verifies that a dropped endpoint +// registration (NewEndpoint returns nil with no surviving pods-map entry) is reported to the +// caller instead of leaving the pod silently untracked, and that a retry succeeds once the +// collector can start again. Regression test for the silent-drop mechanism in #2060. +func TestPodUpdateOrAddIfNotExist_RegistrationDropReturnsError(t *testing.T) { + ctx := context.Background() + factory := &mockEndpointFactory{returnNil: true} + ds := NewDatastore(ctx, factory) + require.NoError(t, ds.PoolSet(ctx, fake.NewFakeClient(), poolutil.InferencePoolToEndpointPool(inferencePool))) + + err := ds.PodUpdateOrAddIfNotExist(ctx, pod1) + require.Error(t, err) + assert.Empty(t, ds.PodList(AllPodsPredicate)) + + // The retry (the reconciler's requeue) succeeds once the collector can start. + factory.returnNil = false + require.NoError(t, ds.PodUpdateOrAddIfNotExist(ctx, pod1)) + assert.Len(t, ds.PodList(AllPodsPredicate), 1) +} + +// TestUpsertEndpoint_ConcurrentStoreDuringNilIsBenign verifies the duplicate-start race is not +// reported as an error: when a concurrent upsert has stored the endpoint by the time NewEndpoint +// returns nil, the datastore is consistent and the upsert is a no-op. +func TestUpsertEndpoint_ConcurrentStoreDuringNilIsBenign(t *testing.T) { + ctx := context.Background() + factory := &storeThenNilFactory{} + ds := NewDatastore(ctx, factory).(*datastore) + factory.ds = ds + id := types.NamespacedName{Name: "ep1", Namespace: "default"} + + created, err := ds.upsertEndpoint(ctx, &fwkdl.EndpointMetadata{NamespacedName: id, Address: "10.0.0.1"}) + + require.NoError(t, err) + assert.False(t, created) + assert.Len(t, ds.PodList(AllPodsPredicate), 1) +} + +// TestDatastore_ConcurrentAddRemoveCompleteness hammers concurrent delete/upsert of the same pod +// and asserts the completeness invariant: an upsert either stores the pod or returns an error the +// caller can retry on. Without the dropped-registration error, an upsert overlapping an in-flight +// delete leaves the pod silently missing until the next pod event (#2060). +func TestDatastore_ConcurrentAddRemoveCompleteness(t *testing.T) { + ctx := context.Background() + ds := NewDatastore(ctx, &collectorRegistryFactory{registered: map[types.NamespacedName]bool{}}) + require.NoError(t, ds.PoolSet(ctx, fake.NewFakeClient(), poolutil.InferencePoolToEndpointPool(inferencePool))) + + for i := 0; i < 200; i++ { + var wg sync.WaitGroup + wg.Add(2) + go func() { + defer wg.Done() + ds.PodDelete(pod1.Name) + }() + go func() { + defer wg.Done() + // The racing upsert may observe the dropped-registration error; the retry below is + // what must converge. + _ = ds.PodUpdateOrAddIfNotExist(ctx, pod1) + }() + wg.Wait() + + // The reconciler's requeue loop: retry until the registration lands. + require.Eventually(t, func() bool { + return ds.PodUpdateOrAddIfNotExist(ctx, pod1) == nil + }, 5*time.Second, time.Millisecond, "iteration %d: upsert never converged", i) + require.Len(t, ds.PodList(AllPodsPredicate), 1, "iteration %d: pod missing after successful upsert", i) + } +} + func TestDiscoveryNotifier_WorksAlongsideDirectUpsert(t *testing.T) { ctx := context.Background() ds := NewDatastore(ctx, &mockEndpointFactory{}) From 640cad266382341fc43fa6566d5c13eeb8dc3986 Mon Sep 17 00:00:00 2001 From: Luke Van Drie Date: Mon, 20 Jul 2026 12:04:58 -0700 Subject: [PATCH 2/4] fix(datastore): retry resync after registration drop and apply metadata on upsert races Close gaps in the dropped-registration handling: - PoolSet stores the pool before resyncing, so a PoolSet retried after a resync failure compared the incoming pool against the already-stored identical pool and skipped the resync entirely. A needsResync flag now forces the retried PoolSet to resync until one succeeds. - upsertEndpoint applies the caller's metadata through the update path when a concurrent upsert wins the registration race, instead of silently discarding it. - The dropped-registration error is a sentinel (errRegistrationDropped) so callers and tests can match it with errors.Is. - PodUpdateOrAddIfNotExist skips and logs at DEBUG when the pool is not synced instead of logging a misleading 'Pod already exists'; the added/exists logging lives in one place. - Unchecked calls of the now error-returning PodUpdateOrAddIfNotExist are handled or explicitly discarded to satisfy errcheck. - The bespoke test endpoint factories collapse into a configurable datalayer.FakeEndpointFactory. Related to #2060. Signed-off-by: Luke Van Drie --- pkg/epp/controller/pod_reconciler.go | 3 - pkg/epp/controller/pod_reconciler_test.go | 23 +--- pkg/epp/datalayer/fake_factory.go | 53 +++++++ pkg/epp/datastore/datastore.go | 103 +++++++------- pkg/epp/datastore/datastore_test.go | 160 ++++++++++++---------- pkg/epp/requestcontrol/director_test.go | 6 +- test/utils/server.go | 4 +- 7 files changed, 208 insertions(+), 144 deletions(-) create mode 100644 pkg/epp/datalayer/fake_factory.go diff --git a/pkg/epp/controller/pod_reconciler.go b/pkg/epp/controller/pod_reconciler.go index 527cdee701..ea2065fbfb 100644 --- a/pkg/epp/controller/pod_reconciler.go +++ b/pkg/epp/controller/pod_reconciler.go @@ -59,9 +59,6 @@ func (c *PodReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.R return ctrl.Result{}, fmt.Errorf("unable to get pod - %w", err) } - // Returning the error makes controller-runtime requeue with backoff, so an endpoint - // registration dropped by an upsert racing an in-flight delete is retried instead of - // leaving the pod untracked until the next pod event. if err := c.updateDatastore(ctx, pod); err != nil { return ctrl.Result{}, fmt.Errorf("failed to update datastore for pod %s - %w", req.NamespacedName, err) } diff --git a/pkg/epp/controller/pod_reconciler_test.go b/pkg/epp/controller/pod_reconciler_test.go index 3aad1e7159..ae62d6be75 100644 --- a/pkg/epp/controller/pod_reconciler_test.go +++ b/pkg/epp/controller/pod_reconciler_test.go @@ -203,7 +203,7 @@ func TestPodReconciler(t *testing.T) { store := datastore.NewDatastore(t.Context(), epf) _ = store.PoolSet(t.Context(), fakeClient, pool.InferencePoolToEndpointPool(test.pool)) for _, pod := range test.existingPods { - store.PodUpdateOrAddIfNotExist(t.Context(), pod) + _ = store.PodUpdateOrAddIfNotExist(t.Context(), pod) } podReconciler := &PodReconciler{Reader: fakeClient, Datastore: store} @@ -231,21 +231,8 @@ func TestPodReconciler(t *testing.T) { } } -// nilEndpointFactory always fails endpoint registration, standing in for a collector that is -// still registered for the endpoint (upsert overlapping an in-flight delete) or fails to start. -type nilEndpointFactory struct{} - -func (nilEndpointFactory) NewEndpoint(_ context.Context, _ *fwkdl.EndpointMetadata) fwkdl.Endpoint { - return nil -} - -func (nilEndpointFactory) UpdateEndpoint(_ context.Context, _ fwkdl.Endpoint) {} - -func (nilEndpointFactory) ReleaseEndpoint(_ fwkdl.Endpoint) {} - // TestPodReconciler_ErrorsOnRegistrationDrop verifies that Reconcile surfaces a dropped endpoint -// registration as an error so controller-runtime requeues the pod, rather than leaving it -// untracked until the next pod event (#2060). +// registration as an error so controller-runtime requeues the pod (#2060). func TestPodReconciler_ErrorsOnRegistrationDrop(t *testing.T) { scheme := runtime.NewScheme() _ = clientgoscheme.AddToScheme(scheme) @@ -263,7 +250,11 @@ func TestPodReconciler_ErrorsOnRegistrationDrop(t *testing.T) { }, }, } - store := datastore.NewDatastore(t.Context(), nilEndpointFactory{}) + // The factory stands in for a collector that is still registered for the endpoint (upsert + // overlapping an in-flight delete) or fails to start. + store := datastore.NewDatastore(t.Context(), &datalayer.FakeEndpointFactory{ + NewEndpointFn: func(_ context.Context, _ *fwkdl.EndpointMetadata) fwkdl.Endpoint { return nil }, + }) _ = store.PoolSet(t.Context(), fakeClient, pool.InferencePoolToEndpointPool(testPool)) podReconciler := &PodReconciler{Reader: fakeClient, Datastore: store} diff --git a/pkg/epp/datalayer/fake_factory.go b/pkg/epp/datalayer/fake_factory.go new file mode 100644 index 0000000000..5aa7d2485e --- /dev/null +++ b/pkg/epp/datalayer/fake_factory.go @@ -0,0 +1,53 @@ +/* +Copyright 2026 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package datalayer + +import ( + "context" + + fwkdl "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/datalayer" +) + +// FakeEndpointFactory is a configurable EndpointFactory for tests. Each hook overrides the +// corresponding method. A nil NewEndpointFn creates a plain endpoint with fresh metrics; nil +// UpdateEndpointFn and ReleaseEndpointFn hooks are no-ops. +type FakeEndpointFactory struct { + NewEndpointFn func(ctx context.Context, meta *fwkdl.EndpointMetadata) fwkdl.Endpoint + UpdateEndpointFn func(ctx context.Context, ep fwkdl.Endpoint) + ReleaseEndpointFn func(ep fwkdl.Endpoint) +} + +var _ EndpointFactory = (*FakeEndpointFactory)(nil) + +func (f *FakeEndpointFactory) NewEndpoint(ctx context.Context, meta *fwkdl.EndpointMetadata) fwkdl.Endpoint { + if f.NewEndpointFn != nil { + return f.NewEndpointFn(ctx, meta) + } + return fwkdl.NewEndpoint(meta, fwkdl.NewMetrics()) +} + +func (f *FakeEndpointFactory) UpdateEndpoint(ctx context.Context, ep fwkdl.Endpoint) { + if f.UpdateEndpointFn != nil { + f.UpdateEndpointFn(ctx, ep) + } +} + +func (f *FakeEndpointFactory) ReleaseEndpoint(ep fwkdl.Endpoint) { + if f.ReleaseEndpointFn != nil { + f.ReleaseEndpointFn(ep) + } +} diff --git a/pkg/epp/datastore/datastore.go b/pkg/epp/datastore/datastore.go index c34c3764ef..67cdfe925e 100644 --- a/pkg/epp/datastore/datastore.go +++ b/pkg/epp/datastore/datastore.go @@ -44,7 +44,11 @@ import ( var ( errPoolNotSynced = errors.New("InferencePool is not initialized in data store") - AllPodsPredicate = func(_ fwkdl.Endpoint) bool { return true } + // errRegistrationDropped reports an endpoint that could not be tracked: its collector is + // still registered from an earlier registration (an upsert overlapping an in-flight delete) + // or failed to start. Callers match it with errors.Is to decide whether to retry. + errRegistrationDropped = errors.New("endpoint registration dropped: collector already registered or failed to start") + AllPodsPredicate = func(_ fwkdl.Endpoint) bool { return true } ) const ( @@ -135,6 +139,11 @@ type datastore struct { // key: types.NamespacedName, value: fwkdl.Endpoint pods *sync.Map epf datalayer.EndpointFactory + // needsResync forces the next PoolSet to run podResyncAll even when the pool is unchanged. + // PoolSet stores the pool before resyncing, so without this flag a PoolSet retried after a + // resync failure would compare the incoming pool against the already-stored identical pool + // and skip the resync. Guarded by mu. + needsResync bool } func (ds *datastore) WithEndpointPool(pool *datalayer.EndpointPool) Datastore { @@ -172,7 +181,7 @@ func (ds *datastore) PoolSet(ctx context.Context, reader client.Reader, endpoint selectorChanged := oldEndpointPool == nil || !selectorEqual(oldEndpointPool.Selector, endpointPool.Selector) targetPortsChanged := oldEndpointPool != nil && !slices.Equal(oldEndpointPool.TargetPorts, endpointPool.TargetPorts) - if selectorChanged || targetPortsChanged { + if selectorChanged || targetPortsChanged || ds.needsResync { logger.V(logutil.DEFAULT).Info("Updating endpoints", "selector", endpointPool.Selector, "targetPortsChanged", targetPortsChanged) // A full resync is required to address the following cases: // 1) At startup, the pod events may get processed before the pool is synced with the datastore, @@ -183,8 +192,10 @@ func (ds *datastore) PoolSet(ctx context.Context, reader client.Reader, endpoint // 3) If the targetPorts changed, we need to resync to remove orphaned rank endpoints that no longer // exist in the new targetPorts configuration. if err := ds.podResyncAll(ctx, reader); err != nil { + ds.needsResync = true return fmt.Errorf("failed to update pods according to the pool selector - %w", err) } + ds.needsResync = false } return nil @@ -295,27 +306,22 @@ func (ds *datastore) PodUpdateOrAddIfNotExist(ctx context.Context, pod *corev1.P pool := ds.pool ds.mu.RUnlock() - added, err := ds.podUpdateOrAddIfNotExist(ctx, pod, pool) - if err != nil { - return err - } - logger := log.FromContext(ctx) - if added { - logger.V(logutil.DEFAULT).Info("Pod added", "name", pod.Name) - } else { - logger.V(logutil.DEFAULT).Info("Pod already exists", "name", pod.Name) + if pool == nil { + // Without the pool's target ports the pod cannot be mapped to endpoints; the resync + // triggered when the pool syncs picks the pod up. + log.FromContext(ctx).V(logutil.DEBUG).Info("Skipping pod upsert, InferencePool not synced", "name", pod.Name) + return nil } - return nil + return ds.podUpdateOrAddIfNotExist(ctx, pod, pool) } // podUpdateOrAddIfNotExist is the lock-free inner implementation. -// Callers must ensure pool is a consistent snapshot (either read under lock +// Callers must ensure pool is a non-nil consistent snapshot (either read under lock // or already held, as in podResyncAll which runs under ds.mu.Lock via PoolSet). -// It returns whether any endpoint was newly created, and a joined error covering every -// endpoint of the pod whose registration was dropped. -func (ds *datastore) podUpdateOrAddIfNotExist(ctx context.Context, pod *corev1.Pod, pool *datalayer.EndpointPool) (bool, error) { +// It returns a joined error covering every endpoint of the pod whose registration was dropped. +func (ds *datastore) podUpdateOrAddIfNotExist(ctx context.Context, pod *corev1.Pod, pool *datalayer.EndpointPool) error { if pool == nil { - return false, nil + return nil } labels := make(map[string]string, len(pod.GetLabels())) @@ -359,6 +365,14 @@ func (ds *datastore) podUpdateOrAddIfNotExist(ctx context.Context, pod *corev1.P added = true } } + logger := log.FromContext(ctx) + if len(errs) == 0 { + if added { + logger.V(logutil.DEFAULT).Info("Pod added", "name", pod.Name) + } else { + logger.V(logutil.DEFAULT).Info("Pod already exists", "name", pod.Name) + } + } // remove endpoints that are no longer active in the pool for idx, port := range pool.TargetPorts { @@ -373,7 +387,7 @@ func (ds *datastore) podUpdateOrAddIfNotExist(ctx context.Context, pod *corev1.P } } - return added, errors.Join(errs...) + return errors.Join(errs...) } func (ds *datastore) PodDelete(podName string) { @@ -389,8 +403,6 @@ func (ds *datastore) PodDelete(podName string) { func (ds *datastore) EndpointUpsert(ctx context.Context, meta *fwkdl.EndpointMetadata) { if _, err := ds.upsertEndpoint(ctx, meta); err != nil { - // Discovery sources have no requeue mechanism, so surface the drop loudly; the endpoint - // stays untracked until the source re-emits it. log.FromContext(ctx).Error(err, "failed to register endpoint", "endpoint", meta.NamespacedName) } } @@ -405,34 +417,35 @@ func (ds *datastore) EndpointDelete(id types.NamespacedName) { // Returns true if the endpoint was newly created, false if it already existed. // Shared by EndpointUpsert and podUpdateOrAddIfNotExist. // -// It returns an error when the endpoint's registration was dropped: NewEndpoint returns nil when -// a collector is already registered for this endpoint or its collector failed to start. If a -// concurrent upsert has stored the entry, the datastore is consistent and the nil is benign. But -// when the pods map has no entry either — the upsert overlapped an in-flight delete that removed -// the entry before deregistering the collector, or the collector failed to start — the endpoint -// is untracked and the caller must retry. +// It returns an error wrapping errRegistrationDropped when the endpoint cannot be tracked: +// NewEndpoint returns nil when a collector is still registered for this endpoint or the +// collector failed to start. When a concurrent upsert has stored an entry for the key, this +// call's metadata is applied through the update path; when no entry exists (the upsert +// overlapped an in-flight delete that removed the entry before deregistering the collector, +// or the collector failed to start), the endpoint is untracked and the caller must retry. func (ds *datastore) upsertEndpoint(ctx context.Context, meta *fwkdl.EndpointMetadata) (bool, error) { - existing, ok := ds.pods.Load(meta.NamespacedName) - if !ok { + for { + if existing, ok := ds.pods.Load(meta.NamespacedName); ok { + ep := existing.(fwkdl.Endpoint) + if ep.GetMetadata().Equal(meta) { + return false, nil + } + ep.UpdateMetadata(meta) + ds.epf.UpdateEndpoint(ctx, ep) + return false, nil + } ep := ds.epf.NewEndpoint(ds.parentCtx, meta) if ep == nil { if _, ok := ds.pods.Load(meta.NamespacedName); ok { - return false, nil + // A concurrent upsert won the registration; apply this call's metadata through + // the update path above. + continue } - return false, fmt.Errorf( - "endpoint %s registration dropped: collector already registered or failed to start", - meta.NamespacedName) + return false, fmt.Errorf("endpoint %s: %w", meta.NamespacedName, errRegistrationDropped) } ds.pods.Store(meta.NamespacedName, ep) return true, nil } - ep := existing.(fwkdl.Endpoint) - if ep.GetMetadata().Equal(meta) { - return false, nil - } - ep.UpdateMetadata(meta) - ds.epf.UpdateEndpoint(ctx, ep) - return false, nil } func (ds *datastore) podResyncAll(ctx context.Context, reader client.Reader) error { @@ -453,21 +466,13 @@ func (ds *datastore) podResyncAll(ctx context.Context, reader client.Reader) err if !podutil.IsPodReady(&pod) { continue } - namespacedName := types.NamespacedName{Name: pod.Name, Namespace: pod.Namespace} // Calculate expected endpoint names based on current targetPorts. for idx := range ds.pool.TargetPorts { activeEndpoints.Insert(createEndpointNamespacedName(&pod, idx)) } - added, err := ds.podUpdateOrAddIfNotExist(ctx, &pod, ds.pool) - if err != nil { - // Propagate so PoolSet fails and the pool reconciler requeues the resync. + if err := ds.podUpdateOrAddIfNotExist(ctx, &pod, ds.pool); err != nil { + // Propagate so PoolSet fails; needsResync makes the retried PoolSet resync again. errs = append(errs, err) - continue - } - if added { - logger.V(logutil.DEFAULT).Info("Pod added", "name", namespacedName) - } else { - logger.V(logutil.DEFAULT).Info("Pod already exists", "name", namespacedName) } } diff --git a/pkg/epp/datastore/datastore_test.go b/pkg/epp/datastore/datastore_test.go index 6d051af7b5..73d1bd5819 100644 --- a/pkg/epp/datastore/datastore_test.go +++ b/pkg/epp/datastore/datastore_test.go @@ -423,7 +423,7 @@ func TestMetrics(t *testing.T) { ds := NewDatastore(ctx, epf) _ = ds.PoolSet(ctx, fakeClient, poolutil.InferencePoolToEndpointPool(inferencePool)) for _, pod := range test.storePods { - ds.PodUpdateOrAddIfNotExist(ctx, pod) + _ = ds.PodUpdateOrAddIfNotExist(ctx, pod) } time.Sleep(1 * time.Second) // Give some time for the metrics to be fetched. if test.predict == nil { @@ -456,7 +456,7 @@ func TestPods(t *testing.T) { existingPods: []*corev1.Pod{}, wantPods: []*corev1.Pod{pod1}, op: func(ctx context.Context, ds Datastore) { - ds.PodUpdateOrAddIfNotExist(ctx, pod1) + _ = ds.PodUpdateOrAddIfNotExist(ctx, pod1) }, }, { @@ -464,7 +464,7 @@ func TestPods(t *testing.T) { existingPods: []*corev1.Pod{pod1}, wantPods: []*corev1.Pod{pod1, pod2}, op: func(ctx context.Context, ds Datastore) { - ds.PodUpdateOrAddIfNotExist(ctx, pod2) + _ = ds.PodUpdateOrAddIfNotExist(ctx, pod2) }, }, { @@ -498,7 +498,7 @@ func TestPods(t *testing.T) { t.Error(err) } for _, pod := range test.existingPods { - ds.PodUpdateOrAddIfNotExist(ctx, pod) + _ = ds.PodUpdateOrAddIfNotExist(ctx, pod) } test.op(ctx, ds) @@ -656,7 +656,7 @@ func TestEndpointMetadata(t *testing.T) { }, }, op: func(ctx context.Context, ds Datastore) { - ds.PodUpdateOrAddIfNotExist(ctx, pod1) + _ = ds.PodUpdateOrAddIfNotExist(ctx, pod1) }, pool: inferencePool, }, @@ -691,7 +691,7 @@ func TestEndpointMetadata(t *testing.T) { }, }, op: func(ctx context.Context, ds Datastore) { - ds.PodUpdateOrAddIfNotExist(ctx, pod1) + _ = ds.PodUpdateOrAddIfNotExist(ctx, pod1) }, pool: inferencePoolMultiTarget, }, @@ -751,7 +751,7 @@ func TestEndpointMetadata(t *testing.T) { }, }, op: func(ctx context.Context, ds Datastore) { - ds.PodUpdateOrAddIfNotExist(ctx, pod2) + _ = ds.PodUpdateOrAddIfNotExist(ctx, pod2) }, pool: inferencePoolMultiTarget, }, @@ -806,7 +806,7 @@ func TestEndpointMetadata(t *testing.T) { t.Error(err) } for _, pod := range test.existingPods { - ds.PodUpdateOrAddIfNotExist(ctx, pod) + _ = ds.PodUpdateOrAddIfNotExist(ctx, pod) } test.op(ctx, ds) @@ -965,7 +965,7 @@ func TestActivePortFiltering(t *testing.T) { // Add all pods for _, pod := range test.pods { - ds.PodUpdateOrAddIfNotExist(ctx, pod) + _ = ds.PodUpdateOrAddIfNotExist(ctx, pod) } // Check final endpoint count @@ -1065,7 +1065,7 @@ func TestActivePortEndpointRemoval(t *testing.T) { operations: []func(Datastore){ // Update the pod to reduce active ports from 3 to 1 func(ds Datastore) { - ds.PodUpdateOrAddIfNotExist(context.Background(), updatedPod1) + _ = ds.PodUpdateOrAddIfNotExist(context.Background(), updatedPod1) }, }, wantEndpointCount: 1, // Only port 8000 should remain active @@ -1081,7 +1081,7 @@ func TestActivePortEndpointRemoval(t *testing.T) { operations: []func(Datastore){ // Update the pod to have no active ports func(ds Datastore) { - ds.PodUpdateOrAddIfNotExist(context.Background(), inactivePod1) + _ = ds.PodUpdateOrAddIfNotExist(context.Background(), inactivePod1) }, }, wantEndpointCount: 0, // No ports should remain active @@ -1113,7 +1113,7 @@ func TestActivePortEndpointRemoval(t *testing.T) { } // Add the initial pod - ds.PodUpdateOrAddIfNotExist(ctx, test.initialPod) + _ = ds.PodUpdateOrAddIfNotExist(ctx, test.initialPod) // Wait a bit for the datastore to process the pod time.Sleep(100 * time.Millisecond) @@ -1200,7 +1200,7 @@ func TestPodUpdateOrAddIfNotExist_ConcurrentPoolSet(t *testing.T) { go func() { defer wg.Done() for range 1000 { - ds.PodUpdateOrAddIfNotExist(ctx, pod) + _ = ds.PodUpdateOrAddIfNotExist(ctx, pod) } }() @@ -1455,54 +1455,10 @@ func TestEndpointDelete_Missing(t *testing.T) { }) } -// storeThenNilFactory simulates the benign duplicate-start race: a concurrent upsert stores the -// endpoint between the caller's pods-map miss and NewEndpoint returning nil. -type storeThenNilFactory struct { - ds *datastore -} - -func (f *storeThenNilFactory) NewEndpoint(_ context.Context, meta *fwkdl.EndpointMetadata) fwkdl.Endpoint { - f.ds.pods.Store(meta.NamespacedName, fwkdl.NewEndpoint(meta, fwkdl.NewMetrics())) - return nil -} - -func (f *storeThenNilFactory) UpdateEndpoint(_ context.Context, _ fwkdl.Endpoint) {} - -func (f *storeThenNilFactory) ReleaseEndpoint(_ fwkdl.Endpoint) {} - -// collectorRegistryFactory mimics Runtime's collector registry semantics: NewEndpoint returns nil -// while a collector is still registered for the endpoint, and ReleaseEndpoint deregisters only -// after a delay, mirroring the delete-event dispatch that runs before collector removal in -// Runtime.ReleaseEndpoint. An upsert landing in that window observes a pods-map miss and a nil -// endpoint at once — the dropped-registration case. -type collectorRegistryFactory struct { - mu sync.Mutex - registered map[types.NamespacedName]bool -} - -func (f *collectorRegistryFactory) NewEndpoint(_ context.Context, meta *fwkdl.EndpointMetadata) fwkdl.Endpoint { - f.mu.Lock() - defer f.mu.Unlock() - if f.registered[meta.NamespacedName] { - return nil - } - f.registered[meta.NamespacedName] = true - return fwkdl.NewEndpoint(meta, fwkdl.NewMetrics()) -} - -func (f *collectorRegistryFactory) UpdateEndpoint(_ context.Context, _ fwkdl.Endpoint) {} - -func (f *collectorRegistryFactory) ReleaseEndpoint(ep fwkdl.Endpoint) { - time.Sleep(50 * time.Microsecond) - f.mu.Lock() - defer f.mu.Unlock() - delete(f.registered, ep.GetMetadata().NamespacedName) -} - // TestPodUpdateOrAddIfNotExist_RegistrationDropReturnsError verifies that a dropped endpoint // registration (NewEndpoint returns nil with no surviving pods-map entry) is reported to the // caller instead of leaving the pod silently untracked, and that a retry succeeds once the -// collector can start again. Regression test for the silent-drop mechanism in #2060. +// collector can start again (#2060). func TestPodUpdateOrAddIfNotExist_RegistrationDropReturnsError(t *testing.T) { ctx := context.Background() factory := &mockEndpointFactory{returnNil: true} @@ -1510,7 +1466,7 @@ func TestPodUpdateOrAddIfNotExist_RegistrationDropReturnsError(t *testing.T) { require.NoError(t, ds.PoolSet(ctx, fake.NewFakeClient(), poolutil.InferencePoolToEndpointPool(inferencePool))) err := ds.PodUpdateOrAddIfNotExist(ctx, pod1) - require.Error(t, err) + require.ErrorIs(t, err, errRegistrationDropped) assert.Empty(t, ds.PodList(AllPodsPredicate)) // The retry (the reconciler's requeue) succeeds once the collector can start. @@ -1519,30 +1475,90 @@ func TestPodUpdateOrAddIfNotExist_RegistrationDropReturnsError(t *testing.T) { assert.Len(t, ds.PodList(AllPodsPredicate), 1) } -// TestUpsertEndpoint_ConcurrentStoreDuringNilIsBenign verifies the duplicate-start race is not -// reported as an error: when a concurrent upsert has stored the endpoint by the time NewEndpoint -// returns nil, the datastore is consistent and the upsert is a no-op. -func TestUpsertEndpoint_ConcurrentStoreDuringNilIsBenign(t *testing.T) { +// TestPoolSet_ResyncRetriedAfterDropError verifies that a PoolSet retried with an identical pool +// after a failed resync runs the resync again rather than skipping it because the pool compares +// equal to the one already stored (#2060). +func TestPoolSet_ResyncRetriedAfterDropError(t *testing.T) { + ctx := context.Background() + readyPod := testutil.FromBase(pod1).ReadyCondition().ObjRef() + fakeClient := fake.NewClientBuilder().WithObjects(readyPod).Build() + + fail := true + factory := &datalayer.FakeEndpointFactory{ + NewEndpointFn: func(_ context.Context, meta *fwkdl.EndpointMetadata) fwkdl.Endpoint { + if fail { + return nil + } + return fwkdl.NewEndpoint(meta, fwkdl.NewMetrics()) + }, + } + ds := NewDatastore(ctx, factory) + + err := ds.PoolSet(ctx, fakeClient, poolutil.InferencePoolToEndpointPool(inferencePool)) + require.ErrorIs(t, err, errRegistrationDropped) + assert.Empty(t, ds.PodList(AllPodsPredicate)) + + fail = false + require.NoError(t, ds.PoolSet(ctx, fakeClient, poolutil.InferencePoolToEndpointPool(inferencePool))) + assert.Len(t, ds.PodList(AllPodsPredicate), 1) +} + +// TestUpsertEndpoint_ConcurrentStoreDuringNilAppliesMetadata verifies the duplicate-start race +// is not reported as an error and does not lose this call's metadata: when a concurrent upsert +// has stored an entry by the time NewEndpoint returns nil, the upsert applies its metadata to +// that entry through the update path. +func TestUpsertEndpoint_ConcurrentStoreDuringNilAppliesMetadata(t *testing.T) { ctx := context.Background() - factory := &storeThenNilFactory{} - ds := NewDatastore(ctx, factory).(*datastore) - factory.ds = ds id := types.NamespacedName{Name: "ep1", Namespace: "default"} - created, err := ds.upsertEndpoint(ctx, &fwkdl.EndpointMetadata{NamespacedName: id, Address: "10.0.0.1"}) + var ds *datastore + factory := &datalayer.FakeEndpointFactory{ + NewEndpointFn: func(_ context.Context, _ *fwkdl.EndpointMetadata) fwkdl.Endpoint { + staleMeta := &fwkdl.EndpointMetadata{NamespacedName: id, Address: "10.0.0.1"} + ds.pods.Store(id, fwkdl.NewEndpoint(staleMeta, fwkdl.NewMetrics())) + return nil + }, + } + ds = NewDatastore(ctx, factory).(*datastore) + + created, err := ds.upsertEndpoint(ctx, &fwkdl.EndpointMetadata{NamespacedName: id, Address: "10.0.0.2"}) require.NoError(t, err) assert.False(t, created) - assert.Len(t, ds.PodList(AllPodsPredicate), 1) + eps := ds.PodList(AllPodsPredicate) + require.Len(t, eps, 1) + assert.Equal(t, "10.0.0.2", eps[0].GetMetadata().Address) } // TestDatastore_ConcurrentAddRemoveCompleteness hammers concurrent delete/upsert of the same pod -// and asserts the completeness invariant: an upsert either stores the pod or returns an error the -// caller can retry on. Without the dropped-registration error, an upsert overlapping an in-flight -// delete leaves the pod silently missing until the next pod event (#2060). +// and asserts the completeness invariant: an upsert either stores the pod or returns an error +// the caller can retry on (#2060). The fake mirrors Runtime's collector registry: NewEndpoint +// fails while a collector is registered for the endpoint, and ReleaseEndpoint deregisters only +// after a delay, so an upsert can observe a pods-map miss while the collector is still +// registered. func TestDatastore_ConcurrentAddRemoveCompleteness(t *testing.T) { ctx := context.Background() - ds := NewDatastore(ctx, &collectorRegistryFactory{registered: map[types.NamespacedName]bool{}}) + + var mu sync.Mutex + registered := map[types.NamespacedName]bool{} + factory := &datalayer.FakeEndpointFactory{ + NewEndpointFn: func(_ context.Context, meta *fwkdl.EndpointMetadata) fwkdl.Endpoint { + mu.Lock() + defer mu.Unlock() + if registered[meta.NamespacedName] { + return nil + } + registered[meta.NamespacedName] = true + return fwkdl.NewEndpoint(meta, fwkdl.NewMetrics()) + }, + ReleaseEndpointFn: func(ep fwkdl.Endpoint) { + time.Sleep(50 * time.Microsecond) + mu.Lock() + defer mu.Unlock() + delete(registered, ep.GetMetadata().NamespacedName) + }, + } + ds := NewDatastore(ctx, factory) require.NoError(t, ds.PoolSet(ctx, fake.NewFakeClient(), poolutil.InferencePoolToEndpointPool(inferencePool))) for i := 0; i < 200; i++ { diff --git a/pkg/epp/requestcontrol/director_test.go b/pkg/epp/requestcontrol/director_test.go index 7c79bdcdba..d1b2677510 100644 --- a/pkg/epp/requestcontrol/director_test.go +++ b/pkg/epp/requestcontrol/director_test.go @@ -888,7 +888,7 @@ func TestDirector_HandleRequest(t *testing.T) { Conditions: []corev1.PodCondition{{Type: corev1.PodReady, Status: corev1.ConditionTrue}}, }, } - ds.PodUpdateOrAddIfNotExist(ctx, testPod) + _ = ds.PodUpdateOrAddIfNotExist(ctx, testPod) } for _, test := range tests { @@ -1055,7 +1055,7 @@ func TestGetRandomEndpoint(t *testing.T) { t.Errorf("unexpected error setting pool: %s", err) } for _, pod := range test.storePods { - ds.PodUpdateOrAddIfNotExist(context.Background(), pod) + _ = ds.PodUpdateOrAddIfNotExist(context.Background(), pod) } d := &Director{datastore: ds} gotEndpoint := d.GetRandomEndpoint() @@ -1887,7 +1887,7 @@ func newConditionalDecodeDirector(t *testing.T, scheduleResult *fwksched.Schedul if err := ds.PoolSet(ctx, fakeClient, poolutil.InferencePoolToEndpointPool(pool)); err != nil { t.Fatalf("PoolSet: %v", err) } - ds.PodUpdateOrAddIfNotExist(ctx, &corev1.Pod{ + _ = ds.PodUpdateOrAddIfNotExist(ctx, &corev1.Pod{ ObjectMeta: metav1.ObjectMeta{Name: "pod1", Namespace: "default", Labels: map[string]string{"app": "inference"}}, Status: corev1.PodStatus{ PodIP: "192.168.1.100", diff --git a/test/utils/server.go b/test/utils/server.go index c0feccebf5..710277daa0 100644 --- a/test/utils/server.go +++ b/test/utils/server.go @@ -59,7 +59,9 @@ func PrepareForTestStreamingServer(t *testing.T, objectives []*v1alpha2.Inferenc } for _, pod := range pods { initObjs = append(initObjs, pod) - ds.PodUpdateOrAddIfNotExist(ctx, pod) + if err := ds.PodUpdateOrAddIfNotExist(ctx, pod); err != nil { + t.Fatalf("failed to add pod %s to datastore: %v", pod.Name, err) + } } scheme := runtime.NewScheme() From 06946f2adaa5de0b66719116feba20ac2ae78f22 Mon Sep 17 00:00:00 2001 From: Luke Van Drie Date: Mon, 20 Jul 2026 15:12:28 -0700 Subject: [PATCH 3/4] chore: use llm-d Authors copyright in fake_factory.go Signed-off-by: Luke Van Drie --- pkg/epp/datalayer/fake_factory.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/epp/datalayer/fake_factory.go b/pkg/epp/datalayer/fake_factory.go index 5aa7d2485e..dde55d538d 100644 --- a/pkg/epp/datalayer/fake_factory.go +++ b/pkg/epp/datalayer/fake_factory.go @@ -1,5 +1,5 @@ /* -Copyright 2026 The Kubernetes Authors. +Copyright 2026 The llm-d Authors. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. From d1372410ce70393f1d9b7fee8699b7e0ef493670 Mon Sep 17 00:00:00 2001 From: Luke Van Drie Date: Mon, 20 Jul 2026 15:21:16 -0700 Subject: [PATCH 4/4] chore: fix requeueing typo flagged by CI Signed-off-by: Luke Van Drie --- pkg/epp/datastore/datastore.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/epp/datastore/datastore.go b/pkg/epp/datastore/datastore.go index 67cdfe925e..eaba703b58 100644 --- a/pkg/epp/datastore/datastore.go +++ b/pkg/epp/datastore/datastore.go @@ -94,7 +94,7 @@ type Datastore interface { PodList(predicate func(fwkdl.Endpoint) bool) []fwkdl.Endpoint // PodUpdateOrAddIfNotExist stores or updates the endpoints for the given pod. It returns an // error when an endpoint registration was dropped (see upsertEndpoint); the pod is then not - // tracked by the datastore and the caller must retry (e.g. by requeueing the reconcile). + // tracked by the datastore and the caller must retry (e.g. by requeuing the reconcile). PodUpdateOrAddIfNotExist(ctx context.Context, pod *corev1.Pod) error PodDelete(podName string)