Skip to content

Commit 610aff9

Browse files
kushnaidukushnaidu
authored
Remove grpc.WaitForReady to make eviction handler more robust (#1069)
* Remove grpc.WaitForReady to make eviction handler more robust Signed-off-by: Kushal Harish Naidu <kushal.harish.naidu@ericsson.com> * Address copilot comments Signed-off-by: Kushal Harish Naidu <kushal.harish.naidu@ericsson.com> * Address copilot comments and add waitForGrpcReady before podReadyCh Signed-off-by: Kushal Harish Naidu <kushal.harish.naidu@ericsson.com> * Increment concurrency counter in redistributeLoad Signed-off-by: Kushal Harish Naidu <kushal.harish.naidu@ericsson.com> * Address copilot comments Signed-off-by: Kushal Harish Naidu <kushal.harish.naidu@ericsson.com> --------- Signed-off-by: Kushal Harish Naidu <kushal.harish.naidu@ericsson.com>
1 parent a2f8749 commit 610aff9

6 files changed

Lines changed: 304 additions & 137 deletions

‎func/internal/podcachemanager.go‎

Lines changed: 34 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -97,7 +97,8 @@ func (pcm *podCacheManager) redistributeLoad(image string, fn *functionInfo, con
9797
for _, ch := range connections {
9898
bestPodIndex, _ := pcm.findBestPod(fn)
9999
if bestPodIndex != -1 {
100-
pod := pcm.functions[image].pods[bestPodIndex]
100+
pod := &pcm.functions[image].pods[bestPodIndex]
101+
pod.concurrentEvaluations.Add(1)
101102
if pod.podData != nil {
102103
pod.SendResponse(ch, nil)
103104
} else {
@@ -216,12 +217,29 @@ func (pcm *podCacheManager) podCacheManager(ctx context.Context) {
216217
return pod.podData != nil && pod.podKey != nil && *pod.podKey == evict.podKey
217218
})
218219
if idx != -1 {
219-
klog.Infof("Evicting dead pod %s from cache for image %s (Unavailable)", evict.podKey.Name, evict.image)
220-
pcm.DeletePodWithServiceInBackgroundByObjectKey(*fn.pods[idx].podData)
221-
fn.pods = slices.Delete(fn.pods, idx, idx+1)
222-
} else {
223-
// Best-effort cleanup of any other stale entries for this image.
224-
pcm.removeUnhealthyPods(fn, false)
220+
// Check if the pod still exists and is healthy in k8s.
221+
// Use a bounded context to avoid blocking the event loop on API-server issues.
222+
k8sPod := &corev1.Pod{}
223+
getCtx, cancel := context.WithTimeout(ctx, 15*time.Second)
224+
err := pcm.podManager.kubeClient.Get(getCtx, *fn.pods[idx].podKey, k8sPod)
225+
cancel()
226+
if apierrors.IsNotFound(err) {
227+
klog.Infof("Evicting missing pod %s from cache for image %s (Unavailable)", evict.podKey.Name, evict.image)
228+
if fn.pods[idx].grpcConnection != nil {
229+
fn.pods[idx].grpcConnection.Close()
230+
}
231+
fn.pods = slices.Delete(fn.pods, idx, idx+1)
232+
} else if err != nil {
233+
// Transient API error — keep the pod in cache rather than evicting a healthy pod.
234+
klog.Warningf("Failed to confirm pod health for %s/%s; keeping it in cache: %v", evict.podKey.Namespace, evict.podKey.Name, err)
235+
} else if k8sPod.Status.Phase != corev1.PodRunning || k8sPod.DeletionTimestamp != nil {
236+
klog.Infof("Evicting dead pod %s from cache for image %s (Unavailable)", evict.podKey.Name, evict.image)
237+
if fn.pods[idx].grpcConnection != nil {
238+
fn.pods[idx].grpcConnection.Close()
239+
}
240+
pcm.DeletePodInBackground(k8sPod)
241+
fn.pods = slices.Delete(fn.pods, idx, idx+1)
242+
}
225243
}
226244
if evict.doneCh != nil {
227245
close(evict.doneCh)
@@ -314,6 +332,15 @@ func (pcm *podCacheManager) retrieveFunctionPods(ctx context.Context) error {
314332
if len(fn.pods) < pcm.maxParallelPodsPerFunction && pod.Status.Phase == corev1.PodRunning {
315333
pData, err := pcm.podManager.createPodData(ctx, serviceKey, podKey, image)
316334
if err == nil {
335+
// Verify gRPC is reachable before adding to cache
336+
if !pcm.podManager.skipGrpcReadyCheck {
337+
if grpcErr := pcm.podManager.waitForGrpcReady(ctx, pData.grpcConnection); grpcErr != nil {
338+
klog.Warningf("retrieved pod %s/%s for %s but gRPC not ready, deleting: %v", pod.Namespace, pod.Name, image, grpcErr)
339+
pData.grpcConnection.Close()
340+
pcm.DeletePodInBackground(&pod)
341+
continue
342+
}
343+
}
317344
klog.Infof("retrieved function evaluator pod %s/%s for %s", pod.Namespace, pod.Name, image)
318345
fn.pods = append(fn.pods, NewPodInfo(nil))
319346
pcm.podManager.podReadyCh <- &podReadyResponse{

‎func/internal/podcachemanager_eventloop_test.go‎

Lines changed: 132 additions & 121 deletions
Original file line numberDiff line numberDiff line change
@@ -60,6 +60,7 @@ func newTestEventLoopPCM(kubeClient client.Client) (*podCacheManager, chan *conn
6060
podReadyCh: readyCh,
6161
podReadyTimeout: 2 * time.Second,
6262
managerNamespace: defaultNamespace,
63+
skipGrpcReadyCheck: true,
6364
},
6465
}
6566
return pcm, reqCh, readyCh, evictCh
@@ -321,128 +322,138 @@ func TestRetrieveFunctionPods_EmptyPodList(t *testing.T) {
321322
assert.Empty(t, pcm.functions)
322323
}
323324

324-
func TestEventLoop_EvictionRemovesPodByKey(t *testing.T) {
325-
podKey := client.ObjectKey{Name: "evict-pod", Namespace: defaultNamespace}
326-
serviceKey := client.ObjectKey{Name: "evict-svc", Namespace: defaultNamespace}
327-
serviceUrl := serviceKey.Name + "." + serviceKey.Namespace + serviceDnsNameSuffix
328-
address := net.JoinHostPort(serviceUrl, defaultWrapperServerPort)
329-
conn, _ := grpc.NewClient(address, grpc.WithTransportCredentials(insecure.NewCredentials()))
330-
331-
k8sPod := &corev1.Pod{
332-
ObjectMeta: metav1.ObjectMeta{Name: "evict-pod", Namespace: defaultNamespace},
333-
Status: corev1.PodStatus{Phase: corev1.PodRunning},
334-
}
335-
k8sSvc := &corev1.Service{
336-
ObjectMeta: metav1.ObjectMeta{Name: "evict-svc", Namespace: defaultNamespace},
337-
}
338-
339-
kubeClient := fake.NewClientBuilder().WithObjects(k8sPod, k8sSvc).Build()
340-
pcm, _, _, evictCh := newTestEventLoopPCM(kubeClient)
341-
342-
readyPod := makeReadyPodInfo("test-image", podKey, serviceKey, conn, 0)
343-
pcm.functions["test-image"] = &functionInfo{
344-
pods: []functionPodInfo{readyPod},
345-
}
346-
347-
go pcm.podCacheManager(t.Context())
348-
349-
// Send eviction for the specific pod
350-
doneCh := make(chan struct{})
351-
evictCh <- &podEvictionRequest{
352-
image: "test-image",
353-
podKey: podKey,
354-
doneCh: doneCh,
355-
}
356-
357-
select {
358-
case <-doneCh:
359-
case <-time.After(5 * time.Second):
360-
t.Fatal("eviction did not complete")
361-
}
362-
363-
// Verify pod was removed from cache
364-
fn := pcm.functions["test-image"]
365-
assert.Empty(t, fn.pods, "evicted pod should be removed from cache")
366-
367-
// Verify k8s pod was deleted (background delete)
368-
deadline := time.Now().Add(2 * time.Second)
369-
for {
370-
var pod corev1.Pod
371-
err := kubeClient.Get(t.Context(), podKey, &pod)
372-
if apierrors.IsNotFound(err) {
373-
break
374-
}
375-
if time.Now().After(deadline) {
376-
assert.True(t, apierrors.IsNotFound(err), "k8s pod should be deleted")
377-
break
378-
}
379-
time.Sleep(20 * time.Millisecond)
380-
}
381-
}
382-
383-
func TestEventLoop_EvictionUnknownImage(t *testing.T) {
384-
kubeClient := fake.NewClientBuilder().Build()
385-
pcm, _, _, evictCh := newTestEventLoopPCM(kubeClient)
386-
387-
go pcm.podCacheManager(t.Context())
388-
389-
// Send eviction for an image not in the cache
390-
doneCh := make(chan struct{})
391-
evictCh <- &podEvictionRequest{
392-
image: "unknown-image",
393-
podKey: client.ObjectKey{Name: "no-pod", Namespace: defaultNamespace},
394-
doneCh: doneCh,
395-
}
396-
397-
select {
398-
case <-doneCh:
399-
// doneCh closed even for unknown image — no hang
400-
case <-time.After(5 * time.Second):
401-
t.Fatal("eviction for unknown image should still close doneCh")
402-
}
403-
}
404-
405-
func TestEventLoop_EvictionPodKeyNotFound(t *testing.T) {
406-
// Pod in cache has a different key than the eviction request
407-
podKey := client.ObjectKey{Name: "real-pod", Namespace: defaultNamespace}
408-
serviceKey := client.ObjectKey{Name: "real-svc", Namespace: defaultNamespace}
409-
serviceUrl := serviceKey.Name + "." + serviceKey.Namespace + serviceDnsNameSuffix
410-
address := net.JoinHostPort(serviceUrl, defaultWrapperServerPort)
411-
conn, _ := grpc.NewClient(address, grpc.WithTransportCredentials(insecure.NewCredentials()))
412-
413-
k8sPod := &corev1.Pod{
414-
ObjectMeta: metav1.ObjectMeta{Name: "real-pod", Namespace: defaultNamespace},
415-
Status: corev1.PodStatus{Phase: corev1.PodRunning},
416-
}
417-
k8sSvc := &corev1.Service{
418-
ObjectMeta: metav1.ObjectMeta{Name: "real-svc", Namespace: defaultNamespace},
419-
}
420-
421-
kubeClient := fake.NewClientBuilder().WithObjects(k8sPod, k8sSvc).Build()
422-
pcm, _, _, evictCh := newTestEventLoopPCM(kubeClient)
423-
424-
readyPod := makeReadyPodInfo("test-image", podKey, serviceKey, conn, 0)
425-
pcm.functions["test-image"] = &functionInfo{
426-
pods: []functionPodInfo{readyPod},
427-
}
428-
429-
go pcm.podCacheManager(t.Context())
430-
431-
// Send eviction with a non-matching podKey → falls back to removeUnhealthyPods
432-
doneCh := make(chan struct{})
433-
evictCh <- &podEvictionRequest{
434-
image: "test-image",
435-
podKey: client.ObjectKey{Name: "wrong-pod", Namespace: defaultNamespace},
436-
doneCh: doneCh,
325+
func TestEventLoop_Eviction(t *testing.T) {
326+
now := metav1.Now()
327+
328+
tests := []struct {
329+
name string
330+
k8sObjects []client.Object
331+
evictImage string
332+
evictPodKey client.ObjectKey
333+
expectRemoved bool
334+
expectMessage string
335+
}{
336+
{
337+
name: "pod not found in k8s is removed from cache",
338+
k8sObjects: nil,
339+
evictImage: "test-image",
340+
evictPodKey: client.ObjectKey{Name: "evict-pod", Namespace: defaultNamespace},
341+
expectRemoved: true,
342+
},
343+
{
344+
name: "pod in Failed state is removed from cache",
345+
k8sObjects: []client.Object{
346+
&corev1.Pod{
347+
ObjectMeta: metav1.ObjectMeta{Name: "evict-pod", Namespace: defaultNamespace},
348+
Status: corev1.PodStatus{Phase: corev1.PodFailed},
349+
},
350+
},
351+
evictImage: "test-image",
352+
evictPodKey: client.ObjectKey{Name: "evict-pod", Namespace: defaultNamespace},
353+
expectRemoved: true,
354+
},
355+
{
356+
name: "pod with DeletionTimestamp is removed from cache",
357+
k8sObjects: []client.Object{
358+
&corev1.Pod{
359+
ObjectMeta: metav1.ObjectMeta{
360+
Name: "evict-pod",
361+
Namespace: defaultNamespace,
362+
DeletionTimestamp: &now,
363+
Finalizers: []string{"test-finalizer"},
364+
},
365+
Status: corev1.PodStatus{Phase: corev1.PodRunning},
366+
},
367+
},
368+
evictImage: "test-image",
369+
evictPodKey: client.ObjectKey{Name: "evict-pod", Namespace: defaultNamespace},
370+
expectRemoved: true,
371+
},
372+
{
373+
name: "healthy Running pod is kept in cache",
374+
k8sObjects: []client.Object{
375+
&corev1.Pod{
376+
ObjectMeta: metav1.ObjectMeta{Name: "evict-pod", Namespace: defaultNamespace},
377+
Status: corev1.PodStatus{Phase: corev1.PodRunning},
378+
},
379+
},
380+
evictImage: "test-image",
381+
evictPodKey: client.ObjectKey{Name: "evict-pod", Namespace: defaultNamespace},
382+
expectRemoved: false,
383+
},
384+
{
385+
name: "unknown image closes doneCh without error",
386+
k8sObjects: nil,
387+
evictImage: "unknown-image",
388+
evictPodKey: client.ObjectKey{Name: "no-pod", Namespace: defaultNamespace},
389+
expectRemoved: false, // no function entry, nothing to remove
390+
},
391+
{
392+
name: "non-matching podKey is a no-op",
393+
k8sObjects: []client.Object{
394+
&corev1.Pod{
395+
ObjectMeta: metav1.ObjectMeta{Name: "real-pod", Namespace: defaultNamespace},
396+
Status: corev1.PodStatus{Phase: corev1.PodRunning},
397+
},
398+
},
399+
evictImage: "test-image",
400+
evictPodKey: client.ObjectKey{Name: "wrong-pod", Namespace: defaultNamespace},
401+
expectRemoved: false,
402+
},
437403
}
438404

439-
select {
440-
case <-doneCh:
441-
case <-time.After(5 * time.Second):
442-
t.Fatal("eviction with non-matching podKey should still close doneCh")
405+
for _, tt := range tests {
406+
t.Run(tt.name, func(t *testing.T) {
407+
podKey := client.ObjectKey{Name: "evict-pod", Namespace: defaultNamespace}
408+
serviceKey := client.ObjectKey{Name: "evict-svc", Namespace: defaultNamespace}
409+
serviceUrl := serviceKey.Name + "." + serviceKey.Namespace + serviceDnsNameSuffix
410+
address := net.JoinHostPort(serviceUrl, defaultWrapperServerPort)
411+
conn, err := grpc.NewClient(address, grpc.WithTransportCredentials(insecure.NewCredentials()))
412+
if !assert.NoError(t, err) {
413+
return
414+
}
415+
t.Cleanup(func() { _ = conn.Close() })
416+
417+
builder := fake.NewClientBuilder()
418+
if len(tt.k8sObjects) > 0 {
419+
builder = builder.WithObjects(tt.k8sObjects...)
420+
}
421+
kubeClient := builder.Build()
422+
pcm, _, _, evictCh := newTestEventLoopPCM(kubeClient)
423+
424+
// Only set up cache entry if the eviction targets "test-image"
425+
if tt.evictImage == "test-image" {
426+
readyPod := makeReadyPodInfo("test-image", podKey, serviceKey, conn, 0)
427+
pcm.functions["test-image"] = &functionInfo{
428+
pods: []functionPodInfo{readyPod},
429+
}
430+
}
431+
432+
go pcm.podCacheManager(t.Context())
433+
434+
doneCh := make(chan struct{})
435+
evictCh <- &podEvictionRequest{
436+
image: tt.evictImage,
437+
podKey: tt.evictPodKey,
438+
doneCh: doneCh,
439+
}
440+
441+
select {
442+
case <-doneCh:
443+
case <-time.After(5 * time.Second):
444+
t.Fatal("eviction did not complete")
445+
}
446+
447+
fn := pcm.functions["test-image"]
448+
if tt.expectRemoved {
449+
if fn != nil {
450+
assert.Empty(t, fn.pods, "pod should be removed from cache")
451+
}
452+
} else {
453+
if fn != nil {
454+
assert.Len(t, fn.pods, 1, "pod should remain in cache")
455+
}
456+
}
457+
})
443458
}
444-
445-
// Pod should still be in cache (it's healthy, removeUnhealthyPods won't remove it)
446-
fn := pcm.functions["test-image"]
447-
assert.Len(t, fn.pods, 1, "healthy pod should remain in cache when podKey doesn't match")
448459
}

‎func/internal/podevaluator.go‎

Lines changed: 4 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -236,14 +236,10 @@ func (pe *podEvaluator) EvaluateFunction(ctx context.Context, req *evaluator.Eva
236236
}
237237
}()
238238

239-
// First attempt: fail fast if pod is dead (no WaitForReady).
240-
// Retries: use WaitForReady since eviction cleaned stale pods and
241-
// the retry may get a newly-created pod that is still starting.
242-
var callOpts []grpc.CallOption
243-
if attempt > 0 {
244-
callOpts = append(callOpts, grpc.WaitForReady(true))
245-
}
246-
resp, err := evaluator.NewFunctionEvaluatorClient(pod.grpcConnection).EvaluateFunction(ctx, req, callOpts...)
239+
// Pod is guaranteed to have an active gRPC connection (verified
240+
// during pod readiness via waitForGrpcReady). Unavailable means
241+
// the pod died after being connected.
242+
resp, err := evaluator.NewFunctionEvaluatorClient(pod.grpcConnection).EvaluateFunction(ctx, req)
247243
if err != nil {
248244
// Retry only on Unavailable — indicates the pod is dead/unreachable:
249245
// connection refused (pod deleted), connection reset (pod crashed),

‎func/internal/podevaluator_podcachemanager_test.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -336,6 +336,7 @@ func TestPodCacheManager(t *testing.T) {
336336
managerNamespace: defaultNamespace,
337337
enablePrivateRegistries: false,
338338
podReadyCh: podReadyCh,
339+
skipGrpcReadyCheck: true,
339340
}
340341

341342
for k, v := range defaultImageMetadataCache {

0 commit comments

Comments
 (0)