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 }