From c43768946eb87d4c443e9b82fd8738a695d8919a Mon Sep 17 00:00:00 2001 From: DualFroz Date: Sun, 18 Oct 2026 10:54:32 +0200 Subject: [PATCH] executor: treat a cached function service as invalid during scale-in IsValid accepted a deployment as long as its status reported an available replica. When the idle reaper scales the deployment to zero, Spec.Replicas drops right away but the status can still count the pod that is shutting down, so a cache hit kept routing requests to it. Export the availability check from executor/util, make it look at the spec as well, and use it in the container and newdeploy IsValid paths. --- .../executortype/container/containermgr.go | 2 +- .../executortype/container/isvalid_test.go | 10 ++++++++++ .../executortype/newdeploy/isvalid_test.go | 10 ++++++++++ .../executortype/newdeploy/newdeploymgr.go | 2 +- pkg/executor/util/deployment.go | 19 +++++++++++++++---- 5 files changed, 37 insertions(+), 6 deletions(-) diff --git a/pkg/executor/executortype/container/containermgr.go b/pkg/executor/executortype/container/containermgr.go index 25d2761..eb80caa 100644 --- a/pkg/executor/executortype/container/containermgr.go +++ b/pkg/executor/executortype/container/containermgr.go @@ -209,7 +209,7 @@ func (caaf *Container) IsValid(ctx context.Context, fsvc *fscache.FuncSvc) bool } return false } - if currentDeploy.Status.AvailableReplicas < 1 { + if !executorUtils.DeploymentAvailable(currentDeploy, 1) { return false } } diff --git a/pkg/executor/executortype/container/isvalid_test.go b/pkg/executor/executortype/container/isvalid_test.go index 2e362cd..4880d01 100644 --- a/pkg/executor/executortype/container/isvalid_test.go +++ b/pkg/executor/executortype/container/isvalid_test.go @@ -59,6 +59,16 @@ func TestContainerIsValid(t *testing.T) { assert.False(t, c.IsValid(t.Context(), fsvcWith(svcRef, deplRef))) }) + t.Run("invalid while the scale-in is not reflected in the status yet", func(t *testing.T) { + zero := int32(0) + depl := deplObj("default", "fn-dep", 1) + depl.Generation = 2 + depl.Spec.Replicas = &zero + depl.Status.ObservedGeneration = 1 + c := newContainer(svcObj("default", "fn-svc"), depl) + assert.False(t, c.IsValid(t.Context(), fsvcWith(svcRef, deplRef))) + }) + t.Run("invalid when the service is missing from the cache", func(t *testing.T) { c := newContainer(deplObj("default", "fn-dep", 1)) // no service assert.False(t, c.IsValid(t.Context(), fsvcWith(svcRef, deplRef))) diff --git a/pkg/executor/executortype/newdeploy/isvalid_test.go b/pkg/executor/executortype/newdeploy/isvalid_test.go index 47c6cd8..394b5b6 100644 --- a/pkg/executor/executortype/newdeploy/isvalid_test.go +++ b/pkg/executor/executortype/newdeploy/isvalid_test.go @@ -62,6 +62,16 @@ func TestNewdeployIsValid(t *testing.T) { assert.False(t, d.IsValid(t.Context(), fsvcWith(svcRef, deplRef))) }) + t.Run("invalid while the scale-in is not reflected in the status yet", func(t *testing.T) { + zero := int32(0) + depl := deplObj("default", "fn-dep", 1) + depl.Generation = 2 + depl.Spec.Replicas = &zero + depl.Status.ObservedGeneration = 1 + d := newDeploy(svcObj("default", "fn-svc"), depl) + assert.False(t, d.IsValid(t.Context(), fsvcWith(svcRef, deplRef))) + }) + t.Run("invalid when the service is missing from the cache", func(t *testing.T) { d := newDeploy(deplObj("default", "fn-dep", 1)) // no service assert.False(t, d.IsValid(t.Context(), fsvcWith(svcRef, deplRef))) diff --git a/pkg/executor/executortype/newdeploy/newdeploymgr.go b/pkg/executor/executortype/newdeploy/newdeploymgr.go index 15f4d90..e7d3389 100644 --- a/pkg/executor/executortype/newdeploy/newdeploymgr.go +++ b/pkg/executor/executortype/newdeploy/newdeploymgr.go @@ -232,7 +232,7 @@ func (deploy *NewDeploy) IsValid(ctx context.Context, fsvc *fscache.FuncSvc) boo } return false } - if currentDeploy.Status.AvailableReplicas < 1 { + if !executorUtils.DeploymentAvailable(currentDeploy, 1) { return false } } diff --git a/pkg/executor/util/deployment.go b/pkg/executor/util/deployment.go index 7ca6130..5e67fed 100644 --- a/pkg/executor/util/deployment.go +++ b/pkg/executor/util/deployment.go @@ -62,7 +62,7 @@ func nextWaitPollDelay(cur time.Duration) time.Duration { return min(cur*3/2, waitPollMaxDelay) } -// deploymentAvailable reports whether depl has at least replicas available, +// DeploymentAvailable reports whether depl has at least replicas available, // as observed by the deployment controller for the spec currently stored. // // The generation gate matters on the scale-from-zero path. UpdateScale only @@ -73,8 +73,19 @@ func nextWaitPollDelay(cur time.Duration) time.Duration { // report a serving deployment while the Service has no endpoint. Until // ObservedGeneration catches up, the status says nothing about this scale // request. Same gate as WaitForDeploymentRollout in the integration framework. -func deploymentAvailable(depl *appsv1.Deployment, replicas int32) bool { - return depl.Status.ObservedGeneration >= depl.Generation && depl.Status.AvailableReplicas >= replicas +// +// The spec check covers the way down. Once the idle reaper scales the +// deployment in, Spec.Replicas drops at once while the status can still count +// the pod that is going away, which would keep a cached function service +// looking valid. +func DeploymentAvailable(depl *appsv1.Deployment, replicas int32) bool { + specReplicas := int32(1) + if depl.Spec.Replicas != nil { + specReplicas = *depl.Spec.Replicas + } + return specReplicas >= replicas && + depl.Status.ObservedGeneration >= depl.Generation && + depl.Status.AvailableReplicas >= replicas } // WaitForDeployment polls the deployment until the deployment controller has @@ -108,7 +119,7 @@ func WaitForDeployment(ctx context.Context, kubeClient kubernetes.Interface, log return nil, err } // TODO check for imagePullerror - if deploymentAvailable(latestDepl, replicas) { + if DeploymentAvailable(latestDepl, replicas) { otelUtils.SpanTrackEvent(ctx, "deploymentAvailable", otelUtils.GetAttributesForDeployment(latestDepl)...) return latestDepl, err }