dualfroz
fd83c655ca
WaitForDeployment returned as soon as AvailableReplicas was high enough, without checking that the status belonged to the current spec. UpdateScale only writes the spec, so a Get issued right after it can still carry the previous generation's status, counting the pod that is being torn down. Gate AvailableReplicas behind ObservedGeneration, the same condition WaitForDeploymentRollout already uses in the integration framework.
271 lines
8.6 KiB
Go
271 lines
8.6 KiB
Go
// SPDX-FileCopyrightText: The Fission Authors
|
||
//
|
||
// SPDX-License-Identifier: Apache-2.0
|
||
|
||
package util
|
||
|
||
import (
|
||
"context"
|
||
"testing"
|
||
"time"
|
||
|
||
"github.com/stretchr/testify/assert"
|
||
"github.com/stretchr/testify/require"
|
||
appsv1 "k8s.io/api/apps/v1"
|
||
autoscalingv1 "k8s.io/api/autoscaling/v1"
|
||
apiv1 "k8s.io/api/core/v1"
|
||
"k8s.io/apimachinery/pkg/runtime"
|
||
"k8s.io/client-go/kubernetes/fake"
|
||
k8stesting "k8s.io/client-go/testing"
|
||
|
||
fv1 "github.com/fission/fission/pkg/apis/core/v1"
|
||
"github.com/fission/fission/pkg/utils/loggerfactory"
|
||
)
|
||
|
||
func secret(ns, name, rv string) *apiv1.Secret {
|
||
return &apiv1.Secret{Namespace: ns, Name: name, ResourceVersion: rv}
|
||
}
|
||
|
||
func configMap(ns, name, rv string) *apiv1.ConfigMap {
|
||
return &apiv1.ConfigMap{Namespace: ns, Name: name, ResourceVersion: rv}
|
||
}
|
||
|
||
func secretRef(ns, name string) fv1.SecretReference {
|
||
return fv1.SecretReference{Namespace: ns, Name: name}
|
||
}
|
||
|
||
func configMapRef(ns, name string) fv1.ConfigMapReference {
|
||
return fv1.ConfigMapReference{Namespace: ns, Name: name}
|
||
}
|
||
|
||
func TestReferencedResourcesRVSum(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
const ns = "default"
|
||
|
||
tests := []struct {
|
||
name string
|
||
objects []runtime.Object
|
||
spec fv1.FunctionSpec
|
||
want int64
|
||
}{
|
||
{
|
||
name: "sums referenced secret and configmap resource versions",
|
||
objects: []runtime.Object{secret(ns, "s1", "10"), configMap(ns, "c1", "20")},
|
||
spec: fv1.FunctionSpec{
|
||
Secrets: []fv1.SecretReference{secretRef(ns, "s1")},
|
||
ConfigMaps: []fv1.ConfigMapReference{configMapRef(ns, "c1")},
|
||
},
|
||
want: 30,
|
||
},
|
||
{
|
||
name: "ignores unreferenced objects in the namespace",
|
||
objects: []runtime.Object{
|
||
secret(ns, "s1", "10"), secret(ns, "s2", "99"),
|
||
configMap(ns, "c1", "20"), configMap(ns, "c2", "77"),
|
||
},
|
||
spec: fv1.FunctionSpec{
|
||
Secrets: []fv1.SecretReference{secretRef(ns, "s1")},
|
||
ConfigMaps: []fv1.ConfigMapReference{configMapRef(ns, "c1")},
|
||
},
|
||
want: 30,
|
||
},
|
||
{
|
||
name: "empty references return zero without listing",
|
||
objects: []runtime.Object{secret(ns, "s1", "10")},
|
||
want: 0,
|
||
},
|
||
{
|
||
name: "missing referenced object contributes zero and does not error",
|
||
objects: []runtime.Object{secret(ns, "s1", "10")},
|
||
spec: fv1.FunctionSpec{
|
||
Secrets: []fv1.SecretReference{secretRef(ns, "s1"), secretRef(ns, "ghost")},
|
||
},
|
||
want: 10,
|
||
},
|
||
{
|
||
name: "cross-namespace reference is skipped",
|
||
objects: []runtime.Object{secret("other", "s1", "10")},
|
||
spec: fv1.FunctionSpec{
|
||
Secrets: []fv1.SecretReference{secretRef("other", "s1")},
|
||
},
|
||
want: 0,
|
||
},
|
||
{
|
||
name: "env valueFrom references contribute (RFC-0030)",
|
||
objects: []runtime.Object{secret(ns, "s1", "10"), configMap(ns, "c1", "20")},
|
||
spec: fv1.FunctionSpec{
|
||
Env: []apiv1.EnvVar{
|
||
{Name: "A", ValueFrom: &apiv1.EnvVarSource{SecretKeyRef: &apiv1.SecretKeySelector{
|
||
Name: "s1", Key: "k"}}},
|
||
{Name: "B", ValueFrom: &apiv1.EnvVarSource{ConfigMapKeyRef: &apiv1.ConfigMapKeySelector{
|
||
Name: "c1", Key: "k"}}},
|
||
},
|
||
},
|
||
want: 30,
|
||
},
|
||
{
|
||
name: "envFrom references contribute (RFC-0030)",
|
||
objects: []runtime.Object{secret(ns, "s1", "10"), configMap(ns, "c1", "20")},
|
||
spec: fv1.FunctionSpec{
|
||
EnvFrom: []apiv1.EnvFromSource{
|
||
{SecretRef: &apiv1.SecretEnvSource{Name: "s1"}},
|
||
{ConfigMapRef: &apiv1.ConfigMapEnvSource{Name: "c1"}},
|
||
},
|
||
},
|
||
want: 30,
|
||
},
|
||
{
|
||
name: "object referenced via both channels contributes once",
|
||
objects: []runtime.Object{secret(ns, "s1", "10")},
|
||
spec: fv1.FunctionSpec{
|
||
Secrets: []fv1.SecretReference{secretRef(ns, "s1")},
|
||
EnvFrom: []apiv1.EnvFromSource{
|
||
{SecretRef: &apiv1.SecretEnvSource{Name: "s1"}},
|
||
},
|
||
},
|
||
want: 10,
|
||
},
|
||
}
|
||
|
||
for _, tc := range tests {
|
||
t.Run(tc.name, func(t *testing.T) {
|
||
t.Parallel()
|
||
client := fake.NewClientset(tc.objects...)
|
||
got, err := ReferencedResourcesRVSum(t.Context(), client, ns, tc.spec)
|
||
require.NoError(t, err)
|
||
assert.Equal(t, tc.want, got)
|
||
})
|
||
}
|
||
}
|
||
|
||
func TestScaleDeployment(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
const ns, name = "default", "dep1"
|
||
replicas := int32(1)
|
||
dep := &appsv1.Deployment{
|
||
Namespace: ns, Name: name,
|
||
Spec: appsv1.DeploymentSpec{Replicas: &replicas},
|
||
}
|
||
client := fake.NewClientset(dep)
|
||
|
||
// Capture the scale subresource update directly: the fake clientset's
|
||
// GetScale read-back has a known scheme-conversion bug, so we assert on
|
||
// the object the helper sends rather than reading it back.
|
||
var captured *autoscalingv1.Scale
|
||
client.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" on the scale subresource, so the action is an UpdateAction carrying a *Scale.
|
||
obj := action.(k8stesting.UpdateAction).GetObject()
|
||
// SAFETY: see above; the update object on the scale subresource is a *Scale.
|
||
captured = obj.(*autoscalingv1.Scale)
|
||
return true, obj, nil
|
||
})
|
||
|
||
err := ScaleDeployment(t.Context(), client, loggerfactory.GetLogger(), ns, name, 3)
|
||
require.NoError(t, err)
|
||
require.NotNil(t, captured)
|
||
assert.Equal(t, int32(3), captured.Spec.Replicas)
|
||
}
|
||
|
||
func TestWaitForDeployment(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
const ns = "default"
|
||
|
||
t.Run("returns once available replicas meet the target", func(t *testing.T) {
|
||
t.Parallel()
|
||
dep := &appsv1.Deployment{
|
||
Namespace: ns, Name: "ready",
|
||
Status: appsv1.DeploymentStatus{AvailableReplicas: 1},
|
||
}
|
||
client := fake.NewClientset(dep)
|
||
|
||
// AvailableReplicas already satisfies the target, so the poll loop
|
||
// returns on the first iteration without sleeping. The wall-clock
|
||
// timeout path (floored at fv1.DefaultSpecializationTimeOut) is left
|
||
// to integration coverage so the unit test stays fast.
|
||
got, err := WaitForDeployment(t.Context(), client, loggerfactory.GetLogger(), dep, 1, 1)
|
||
require.NoError(t, err)
|
||
require.NotNil(t, got)
|
||
assert.GreaterOrEqual(t, got.Status.AvailableReplicas, int32(1))
|
||
})
|
||
|
||
t.Run("ignores a status the controller has not caught up with", func(t *testing.T) {
|
||
t.Parallel()
|
||
// Scaled in from one replica to zero and back up: the spec is at
|
||
// generation 2, while the status still describes generation 1 and
|
||
// counts the pod the idle reaper is tearing down.
|
||
one := int32(1)
|
||
scaled := &appsv1.Deployment{
|
||
Namespace: ns, Name: "scaled-up", Generation: 2,
|
||
Spec: appsv1.DeploymentSpec{Replicas: &one},
|
||
Status: appsv1.DeploymentStatus{ObservedGeneration: 1, AvailableReplicas: 1},
|
||
}
|
||
client := fake.NewClientset(scaled)
|
||
var gets int
|
||
client.PrependReactor("get", "deployments", func(k8stesting.Action) (bool, runtime.Object, error) {
|
||
latest := scaled.DeepCopy()
|
||
gets++
|
||
if gets > 1 {
|
||
latest.Status.ObservedGeneration = latest.Generation
|
||
}
|
||
return true, latest, nil
|
||
})
|
||
|
||
got, err := WaitForDeployment(t.Context(), client, loggerfactory.GetLogger(), scaled, 1, 1)
|
||
require.NoError(t, err)
|
||
require.NotNil(t, got)
|
||
assert.Equal(t, got.Generation, got.Status.ObservedGeneration,
|
||
"returned on a status that predates the scale request")
|
||
})
|
||
|
||
t.Run("surfaces a get error", func(t *testing.T) {
|
||
t.Parallel()
|
||
client := fake.NewClientset()
|
||
missing := &appsv1.Deployment{Namespace: ns, Name: "ghost"}
|
||
|
||
got, err := WaitForDeployment(t.Context(), client, loggerfactory.GetLogger(), missing, 1, 1)
|
||
require.Error(t, err)
|
||
assert.Nil(t, got)
|
||
})
|
||
|
||
t.Run("returns promptly when the context is cancelled", func(t *testing.T) {
|
||
t.Parallel()
|
||
// Deployment exists but is not yet available, so the loop would
|
||
// otherwise sleep; a cancelled context must short-circuit instead of
|
||
// blocking for the (floored) timeout window.
|
||
dep := &appsv1.Deployment{
|
||
Namespace: ns, Name: "pending",
|
||
Status: appsv1.DeploymentStatus{AvailableReplicas: 0},
|
||
}
|
||
client := fake.NewClientset(dep)
|
||
ctx, cancel := context.WithCancel(t.Context())
|
||
cancel()
|
||
|
||
got, err := WaitForDeployment(ctx, client, loggerfactory.GetLogger(), dep, 1, 1)
|
||
require.Error(t, err)
|
||
assert.Nil(t, got)
|
||
})
|
||
}
|
||
|
||
func TestNextWaitPollDelay(t *testing.T) {
|
||
t.Parallel()
|
||
// ×1.5 growth from the initial delay, capped at the old fixed interval.
|
||
assert.Equal(t, 75*time.Millisecond, nextWaitPollDelay(waitPollInitialDelay))
|
||
assert.Equal(t, waitPollMaxDelay, nextWaitPollDelay(900*time.Millisecond))
|
||
assert.Equal(t, waitPollMaxDelay, nextWaitPollDelay(waitPollMaxDelay))
|
||
// The schedule reaches its cap quickly, bounding the extra apiserver GETs
|
||
// an adaptive poll adds over the old 1/s: ~9 in the first ~5s of a wait.
|
||
d, polls := waitPollInitialDelay, 0
|
||
for elapsed := time.Duration(0); d < waitPollMaxDelay; polls++ {
|
||
elapsed += d
|
||
d = nextWaitPollDelay(d)
|
||
require.Less(t, elapsed, 6*time.Second)
|
||
}
|
||
assert.LessOrEqual(t, polls, 10)
|
||
}
|