Skip to content

Commit b8d69c0

Browse files
kushnaidukushnaidu
authored
Evict pod based on GRPC response instead of force check on every call (kptdev#1055)
* Evict pods only on grpc connection failure with retry in function-runner Signed-off-by: Kushal Harish Naidu <kushal.harish.naidu@ericsson.com> * revert gc-interval Signed-off-by: Kushal Harish Naidu <kushal.harish.naidu@ericsson.com> * revert maxWaitlist to default value Signed-off-by: Kushal Harish Naidu <kushal.harish.naidu@ericsson.com> * Remove backoff as WaitForReady(true) is enabled on retries Signed-off-by: Kushal Harish Naidu <kushal.harish.naidu@ericsson.com> * decrement counter before continue and guard it Signed-off-by: Kushal Harish Naidu <kushal.harish.naidu@ericsson.com> * use pod.image for eviction Signed-off-by: Kushal Harish Naidu <kushal.harish.naidu@ericsson.com> * Update eviction to cleanup k8s objects Signed-off-by: Kushal Harish Naidu <kushal.harish.naidu@ericsson.com> * Introduce pod eviction acknowledgement and update test to get config from function runner deployment Signed-off-by: Kushal Harish Naidu <kushal.harish.naidu@ericsson.com> * Update comment to include evictionCh Signed-off-by: Kushal Harish Naidu <kushal.harish.naidu@ericsson.com> * Add back eviction for targetted pod Signed-off-by: Kushal Harish Naidu <kushal.harish.naidu@ericsson.com> * Make evictionCh ctx aware for both send and ack Signed-off-by: Kushal Harish Naidu <kushal.harish.naidu@ericsson.com> * Address co-pilot comments Signed-off-by: Kushal Harish Naidu <kushal.harish.naidu@ericsson.com> * Make grpc retries configurable and move code for readability Signed-off-by: Kushal Harish Naidu <kushal.harish.naidu@ericsson.com> * Add unit test to improve code coverage Signed-off-by: Kushal Harish Naidu <kushal.harish.naidu@ericsson.com> * Extract default pod evaluator parameters into named constants 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 51e8126 commit b8d69c0

8 files changed

Lines changed: 502 additions & 61 deletions

‎func/internal/podcachemanager.go‎

Lines changed: 35 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,7 @@ import (
3535

3636
// podCacheManager manages the cache of the pods and the corresponding GRPC clients.
3737
// It also does the garbage collection after pods' TTL.
38-
// It has 2 receive-only channels: connectionRequestCh and podReadyCh.
38+
// It has 3 receive-only channels: connectionRequestCh, podReadyCh, and evictionCh.
3939
// It listens to the connectionRequestCh channel and receives clientConnRequest from the
4040
// GRPC request handlers and add them in the waitlists.
4141
// It also listens to the podReadyCh channel. If a pod is ready, it notifies the
@@ -48,6 +48,8 @@ type podCacheManager struct {
4848
connectionRequestCh <-chan *connectionRequest
4949
// podReadyCh is a channel to receive the information when a pod is ready.
5050
podReadyCh <-chan *podReadyResponse
51+
// evictionCh receives requests to remove specific dead pods from cache
52+
evictionCh <-chan *podEvictionRequest
5153

5254
// functions maps KRM function image names to its pods and waitlist information.
5355
functions map[string]*functionInfo
@@ -59,6 +61,15 @@ type podCacheManager struct {
5961
functionConfigMap *fnconf.FunctionConfigStore
6062
}
6163

64+
// podEvictionRequest is sent after an Unavailable gRPC error to remove a dead pod from cache.
65+
type podEvictionRequest struct {
66+
image string
67+
podKey client.ObjectKey
68+
// doneCh is closed by the cache manager once the pod has been removed from cache,
69+
// allowing the caller to wait for eviction completion before retrying.
70+
doneCh chan struct{}
71+
}
72+
6273
// functionInfo holds the list of all pod instances for the same KRM function image.
6374
type functionInfo struct {
6475
// status of all pods belonging to the same KRM function image
@@ -114,7 +125,6 @@ func (pcm *podCacheManager) podCacheManager(ctx context.Context) {
114125
fn := pcm.FunctionInfo(req.image)
115126

116127
shouldScaleUp := false
117-
pcm.removeUnhealthyPods(fn, false)
118128
bestPodIndex, bestWaitlistLen := pcm.findBestPod(fn)
119129
_, maxWaitlist, maxPods := pcm.getParamsForImage(req.image)
120130
if bestPodIndex == -1 {
@@ -194,6 +204,29 @@ func (pcm *podCacheManager) podCacheManager(ctx context.Context) {
194204
}
195205
pod.waitlist = nil
196206

207+
case evict := <-pcm.evictionCh:
208+
fn, ok := pcm.functions[evict.image]
209+
if !ok {
210+
if evict.doneCh != nil {
211+
close(evict.doneCh)
212+
}
213+
continue
214+
}
215+
idx := slices.IndexFunc(fn.pods, func(pod functionPodInfo) bool {
216+
return pod.podData != nil && pod.podKey != nil && *pod.podKey == evict.podKey
217+
})
218+
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)
225+
}
226+
if evict.doneCh != nil {
227+
close(evict.doneCh)
228+
}
229+
197230
case <-tick:
198231
pcm.garbageCollector()
199232
case <-ctx.Done():

‎func/internal/podcachemanager_eventloop_test.go‎

Lines changed: 135 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -38,14 +38,16 @@ import (
3838
// newTestEventLoopPCM creates a podCacheManager with unbuffered channels suitable
3939
// for deterministic event loop testing. The podManager's podReadyCh is the same as
4040
// the pcm's podReadyCh so that getFuncEvalPodClient sends results to the event loop.
41-
func newTestEventLoopPCM(kubeClient client.Client) (*podCacheManager, chan *connectionRequest, chan *podReadyResponse) {
41+
func newTestEventLoopPCM(kubeClient client.Client) (*podCacheManager, chan *connectionRequest, chan *podReadyResponse, chan *podEvictionRequest) {
4242
reqCh := make(chan *connectionRequest)
4343
readyCh := make(chan *podReadyResponse)
44+
evictCh := make(chan *podEvictionRequest)
4445
pcm := &podCacheManager{
4546
gcScanInterval: 5 * time.Minute,
4647
podTTL: 10 * time.Minute,
4748
connectionRequestCh: reqCh,
4849
podReadyCh: readyCh,
50+
evictionCh: evictCh,
4951
functions: map[string]*functionInfo{},
5052
maxWaitlistLength: 2,
5153
maxParallelPodsPerFunction: 1,
@@ -60,14 +62,14 @@ func newTestEventLoopPCM(kubeClient client.Client) (*podCacheManager, chan *conn
6062
managerNamespace: defaultNamespace,
6163
},
6264
}
63-
return pcm, reqCh, readyCh
65+
return pcm, reqCh, readyCh, evictCh
6466
}
6567

6668
// ---------- Event Loop Tests ----------
6769

6870
func TestEventLoop_PodReadyEmptyImage(t *testing.T) {
6971
kubeClient := fake.NewClientBuilder().Build()
70-
pcm, _, readyCh := newTestEventLoopPCM(kubeClient)
72+
pcm, _, readyCh, _ := newTestEventLoopPCM(kubeClient)
7173

7274
// Pre-populate a pending pod for "test-image" BEFORE starting the event loop
7375
waitCh := make(chan *connectionResponse, 1)
@@ -104,7 +106,7 @@ func TestEventLoop_PodReadyEmptyImage(t *testing.T) {
104106

105107
func TestEventLoop_PodReadyUnknownFunction(t *testing.T) {
106108
kubeClient := fake.NewClientBuilder().Build()
107-
pcm, _, readyCh := newTestEventLoopPCM(kubeClient)
109+
pcm, _, readyCh, _ := newTestEventLoopPCM(kubeClient)
108110

109111
// Pre-populate a pending pod for "known-image" BEFORE starting the event loop
110112
waitCh := make(chan *connectionResponse, 1)
@@ -163,7 +165,7 @@ func TestEventLoop_PodReadyNoPendingPod(t *testing.T) {
163165
}
164166

165167
kubeClient := fake.NewClientBuilder().WithObjects(k8sPod, k8sSvc).Build()
166-
pcm, reqCh, readyCh := newTestEventLoopPCM(kubeClient)
168+
pcm, reqCh, readyCh, _ := newTestEventLoopPCM(kubeClient)
167169

168170
readyPod := makeReadyPodInfo("test-image", podKey, serviceKey, conn, 0)
169171
pcm.functions["test-image"] = &functionInfo{
@@ -196,7 +198,7 @@ func TestEventLoop_PodReadyNoPendingPod(t *testing.T) {
196198

197199
func TestEventLoop_QueueOnPendingPod(t *testing.T) {
198200
kubeClient := fake.NewClientBuilder().Build()
199-
pcm, reqCh, readyCh := newTestEventLoopPCM(kubeClient)
201+
pcm, reqCh, readyCh, _ := newTestEventLoopPCM(kubeClient)
200202

201203
// Pre-populate with pending pod that has one initial waiter
202204
initialCh := make(chan *connectionResponse, 1)
@@ -251,7 +253,7 @@ func TestEventLoop_PodFailedNoRedistribution(t *testing.T) {
251253
},
252254
}).Build()
253255

254-
pcm, reqCh, _ := newTestEventLoopPCM(kubeClient)
256+
pcm, reqCh, _, _ := newTestEventLoopPCM(kubeClient)
255257

256258
// Pre-populate imageMetadataCache so imageDigestAndEntrypoint returns instantly
257259
pcm.podManager.imageMetadataCache.Store("ghcr.io/kptdev/krm-functions-catalog/test-fn:latest", &digestAndEntrypoint{
@@ -318,3 +320,129 @@ func TestRetrieveFunctionPods_EmptyPodList(t *testing.T) {
318320
assert.NoError(t, err)
319321
assert.Empty(t, pcm.functions)
320322
}
323+
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,
437+
}
438+
439+
select {
440+
case <-doneCh:
441+
case <-time.After(5 * time.Second):
442+
t.Fatal("eviction with non-matching podKey should still close doneCh")
443+
}
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")
448+
}

0 commit comments

Comments
 (0)