newdeploy: wait for the replica after scaling a function in from zero
createOrGetDeployment decided whether to wait for readiness from the Status.AvailableReplicas of the copy it read before ScaleDeployment, the same pattern the container manager had. On the scale-from-zero path that copy can still count the pod the idle reaper is tearing down, so WaitForDeployment, and with it the ObservedGeneration gate, was skipped entirely. Track whether we issued the scale-up and always wait in that case. The test covers both the pre-scale read and a status the controller has not caught up with yet.
This commit is contained in:
1 parent
180cbd38f6
commit
a8bc9d02d5
2 files changed
+104
-1
No files matched your search
@@ -65,14 +65,18 @@ func (deploy *NewDeploy) createOrGetDeployment(ctx context.Context, fn *fv1.Func
|
|||||||
return existingDepl, nil
|
return existingDepl, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// existingDepl was read before the scale request, and on the scale-from-zero path
|
||||||
|
// its status can still count the pod the idle reaper is tearing down.
|
||||||
|
scaledUp := false
|
||||||
if *existingDepl.Spec.Replicas < minScale {
|
if *existingDepl.Spec.Replicas < minScale {
|
||||||
err = util.ScaleDeployment(ctx, deploy.kubernetesClient, deploy.logger, existingDepl.Namespace, existingDepl.Name, minScale)
|
err = util.ScaleDeployment(ctx, deploy.kubernetesClient, deploy.logger, existingDepl.Namespace, existingDepl.Name, minScale)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
deploy.logger.Error(err, "error scaling up function deployment", "function", fn.Name)
|
deploy.logger.Error(err, "error scaling up function deployment", "function", fn.Name)
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
scaledUp = true
|
||||||
}
|
}
|
||||||
if existingDepl.Status.AvailableReplicas < minScale {
|
if scaledUp || existingDepl.Status.AvailableReplicas < minScale {
|
||||||
existingDepl, err = util.WaitForDeployment(ctx, deploy.kubernetesClient, deploy.logger, existingDepl, minScale, specializationTimeout)
|
existingDepl, err = util.WaitForDeployment(ctx, deploy.kubernetesClient, deploy.logger, existingDepl, minScale, specializationTimeout)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -5,14 +5,18 @@
|
|||||||
package newdeploy
|
package newdeploy
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"context"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/stretchr/testify/assert"
|
"github.com/stretchr/testify/assert"
|
||||||
"github.com/stretchr/testify/require"
|
"github.com/stretchr/testify/require"
|
||||||
|
appsv1 "k8s.io/api/apps/v1"
|
||||||
apiv1 "k8s.io/api/core/v1"
|
apiv1 "k8s.io/api/core/v1"
|
||||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||||
|
"k8s.io/apimachinery/pkg/runtime"
|
||||||
"k8s.io/client-go/kubernetes/fake"
|
"k8s.io/client-go/kubernetes/fake"
|
||||||
|
k8stesting "k8s.io/client-go/testing"
|
||||||
|
|
||||||
fv1 "github.com/fission/fission/pkg/apis/core/v1"
|
fv1 "github.com/fission/fission/pkg/apis/core/v1"
|
||||||
ferror "github.com/fission/fission/pkg/error"
|
ferror "github.com/fission/fission/pkg/error"
|
||||||
@@ -639,3 +643,98 @@ func TestGetDeploymentSpecNonOCIUnchanged(t *testing.T) {
|
|||||||
}
|
}
|
||||||
assert.Empty(t, dep.Spec.Template.Spec.ImagePullSecrets)
|
assert.Empty(t, dep.Spec.Template.Spec.ImagePullSecrets)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestNewDeployScaleFromZeroWaitsForAvailability is the newdeploy side of the
|
||||||
|
// scale-to-zero race reported in #3183. On the scale-from-zero path the
|
||||||
|
// deployment status can still describe the pod the idle reaper is tearing
|
||||||
|
// down, either because it was read before the scale request or because the
|
||||||
|
// controller has not written a status for that request yet. A caller that
|
||||||
|
// trusts such a status reports success while the Service has no endpoint and
|
||||||
|
// the router forwards into it.
|
||||||
|
func TestNewDeployScaleFromZeroWaitsForAvailability(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
// Generation of the spec after the scale request.
|
||||||
|
const generation = 2
|
||||||
|
|
||||||
|
tests := []struct {
|
||||||
|
name string
|
||||||
|
// status is the deployment status served by the nth read, 1-based.
|
||||||
|
status func(read int) appsv1.DeploymentStatus
|
||||||
|
}{
|
||||||
|
{
|
||||||
|
// The copy createOrGetDeployment read before scaling up.
|
||||||
|
name: "pre-scale read still counts the terminating pod",
|
||||||
|
status: func(read int) appsv1.DeploymentStatus {
|
||||||
|
available := int32(0)
|
||||||
|
if read == 1 {
|
||||||
|
available = 1
|
||||||
|
}
|
||||||
|
return appsv1.DeploymentStatus{ObservedGeneration: generation, AvailableReplicas: available}
|
||||||
|
},
|
||||||
|
},
|
||||||
|
{
|
||||||
|
// Status is written asynchronously by the deployment controller, so
|
||||||
|
// every read up to the deadline can still describe the generation
|
||||||
|
// before the scale request.
|
||||||
|
name: "status has not caught up with the scale request",
|
||||||
|
status: func(int) appsv1.DeploymentStatus {
|
||||||
|
return appsv1.DeploymentStatus{ObservedGeneration: generation - 1, AvailableReplicas: 1}
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, tc := range tests {
|
||||||
|
t.Run(tc.name, func(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
env := newTestNewDeployEnv()
|
||||||
|
fn := newTestNewDeployFunction()
|
||||||
|
fn.Spec.InvokeStrategy.ExecutionStrategy.MinScale = 0
|
||||||
|
|
||||||
|
ns := utils.DefaultNSResolver().GetFunctionNS(fn.Namespace)
|
||||||
|
zero := int32(0)
|
||||||
|
scaledIn := &appsv1.Deployment{
|
||||||
|
ObjectMeta: metav1.ObjectMeta{
|
||||||
|
Name: "newdeploy-test-fn",
|
||||||
|
Namespace: ns,
|
||||||
|
Generation: generation,
|
||||||
|
Annotations: map[string]string{fv1.EXECUTOR_INSTANCEID_LABEL: "test-instance"},
|
||||||
|
},
|
||||||
|
Spec: appsv1.DeploymentSpec{Replicas: &zero},
|
||||||
|
}
|
||||||
|
|
||||||
|
kubeClient := fake.NewClientset(scaledIn)
|
||||||
|
var reads int
|
||||||
|
kubeClient.PrependReactor("get", "deployments", func(k8stesting.Action) (bool, runtime.Object, error) {
|
||||||
|
depl := scaledIn.DeepCopy()
|
||||||
|
reads++
|
||||||
|
depl.Status = tc.status(reads)
|
||||||
|
return true, depl, nil
|
||||||
|
})
|
||||||
|
// The fake clientset has no built-in scale subresource for deployments.
|
||||||
|
kubeClient.PrependReactor("update", "deployments", func(action k8stesting.Action) (bool, runtime.Object, error) {
|
||||||
|
if action.GetSubresource() != "scale" {
|
||||||
|
return false, nil, nil
|
||||||
|
}
|
||||||
|
// SAFETY: this reactor is registered for "update" and returns early unless the subresource is scale, so the action is an UpdateAction.
|
||||||
|
return true, action.(k8stesting.UpdateAction).GetObject(), nil
|
||||||
|
})
|
||||||
|
|
||||||
|
deploy := newTestNewDeploy(t)
|
||||||
|
deploy.kubernetesClient = kubeClient
|
||||||
|
deploy.instanceID = "test-instance"
|
||||||
|
|
||||||
|
// The deployment never becomes available in the fake cluster, so a
|
||||||
|
// correct implementation runs the readiness wait out to this deadline.
|
||||||
|
ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second)
|
||||||
|
defer cancel()
|
||||||
|
|
||||||
|
_, err := deploy.createOrGetDeployment(ctx, fn, env, "newdeploy-test-fn",
|
||||||
|
map[string]string{"app": "newdeploy-test"}, map[string]string{}, ns)
|
||||||
|
require.Error(t, err, "scale-from-zero must not report success while no replica is available")
|
||||||
|
assert.ErrorIs(t, err, context.DeadlineExceeded,
|
||||||
|
"expected the readiness wait to run to its deadline, got: %v", err)
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in new issue
Block a user