Upstream fission/fission state of the touched files before dae21a228f

pkg/executor/util/deployment.go

@@ -0,0 +1,198 @@
+// SPDX-FileCopyrightText: The Fission Authors
+//
+// SPDX-License-Identifier: Apache-2.0
+
+package util
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+	"time"
+
+	"github.com/go-logr/logr"
+	appsv1 "k8s.io/api/apps/v1"
+	autoscalingv1 "k8s.io/api/autoscaling/v1"
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+
+	fv1 "github.com/fission/fission/pkg/apis/core/v1"
+	otelUtils "github.com/fission/fission/pkg/utils/otel"
+)
+
+// ScaleDeployment scales the named deployment to replicas via the scale
+// subresource. Shared by the newdeploy and container managers, which both
+// scale a per-function deployment up to MinScale on (re)specialization.
+func ScaleDeployment(ctx context.Context, kubeClient kubernetes.Interface, logger logr.Logger, ns, name string, replicas int32) error {
+	otelUtils.SpanTrackEvent(ctx, "scaleDeployment", otelUtils.MapToAttributes(map[string]string{
+		"deployment-name":      name,
+		"deployment-namespace": ns,
+		"replicas":             fmt.Sprintf("%d", replicas),
+	})...)
+	logger = otelUtils.LoggerWithTraceID(ctx, logger)
+	logger.Info("scaling deployment",
+		"deployment", name,
+		"namespace", ns,
+		"replicas", replicas)
+	_, err := kubeClient.AppsV1().Deployments(ns).UpdateScale(ctx, name, &autoscalingv1.Scale{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: ns,
+		},
+		Spec: autoscalingv1.ScaleSpec{
+			Replicas: replicas,
+		},
+	}, metav1.UpdateOptions{})
+	return err
+}
+
+// waitPollInitialDelay/waitPollMaxDelay shape WaitForDeployment's adaptive
+// poll: scale-from-zero deployments usually become available well under a
+// second, so polling starts fine-grained and backs off ×1.5 to the old 1s
+// interval — a fixed 1s tick quantized every newdeploy/container cold start
+// up to a full second.
+const (
+	waitPollInitialDelay = 50 * time.Millisecond
+	waitPollMaxDelay     = time.Second
+)
+
+// nextWaitPollDelay returns the poll interval following cur.
+func nextWaitPollDelay(cur time.Duration) time.Duration {
+	return min(cur*3/2, waitPollMaxDelay)
+}
+
+// WaitForDeployment polls the deployment until it has at least replicas
+// AvailableReplicas, or until specializationTimeout seconds elapse. Shared by
+// the newdeploy and container managers. specializationTimeout is floored at
+// fv1.DefaultSpecializationTimeOut. AvailableReplicas is used in preference to
+// ReadyReplicas since the pods may not be able to serve network traffic yet.
+func WaitForDeployment(ctx context.Context, kubeClient kubernetes.Interface, logger logr.Logger, depl *appsv1.Deployment, replicas int32, specializationTimeout int) (latestDepl *appsv1.Deployment, err error) {
+	oldStatus := depl.Status
+	otelUtils.SpanTrackEvent(ctx, "waitForDeployment", otelUtils.GetAttributesForDeployment(depl)...)
+	// if no specializationTimeout is set, use default value
+	if specializationTimeout < fv1.DefaultSpecializationTimeOut {
+		specializationTimeout = fv1.DefaultSpecializationTimeOut
+	}
+
+	logger = otelUtils.LoggerWithTraceID(ctx, logger)
+
+	// The timeout is a wall-clock deadline (the poll interval varies, so an
+	// iteration count would silently shrink or stretch it).
+	deadline := time.Now().Add(time.Duration(specializationTimeout) * time.Second)
+	delay := waitPollInitialDelay
+	// One reusable timer across polls (Reset is safe on Go ≥1.23 timers)
+	// instead of a time.After allocation per iteration.
+	timer := time.NewTimer(time.Hour)
+	defer timer.Stop()
+
+	for {
+		latestDepl, err = kubeClient.AppsV1().Deployments(depl.ObjectMeta.Namespace).Get(ctx, depl.Name, metav1.GetOptions{})
+		if err != nil {
+			return nil, err
+		}
+		// TODO check for imagePullerror
+		if latestDepl.Status.AvailableReplicas >= replicas {
+			otelUtils.SpanTrackEvent(ctx, "deploymentAvailable", otelUtils.GetAttributesForDeployment(latestDepl)...)
+			return latestDepl, err
+		}
+		remaining := time.Until(deadline)
+		if remaining <= 0 {
+			break
+		}
+		// Wait before the next poll, but stay responsive to cancellation
+		// (executor shutdown / loss of leadership) instead of blocking for the
+		// full interval.
+		timer.Reset(min(delay, remaining))
+		select {
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		case <-timer.C:
+		}
+		delay = nextWaitPollDelay(delay)
+	}
+
+	logger.Error(nil, "Deployment provision failed within timeout window", "name", latestDepl.Name, "old_status", oldStatus,
+		"current_status", latestDepl.Status, "timeout", specializationTimeout)
+
+	// this error appears in the executor pod logs
+	timeoutError := fmt.Errorf("failed to create deployment within the timeout window of %d seconds", specializationTimeout)
+	return nil, timeoutError
+}
+
+// ReferencedResourcesRVSum returns the sum of the resource versions of the
+// secrets and configmaps the function references. Shared by the newdeploy and
+// container managers.
+//
+// We used to update a timestamp in the deployment environment field in order to
+// trigger a rolling update when the function's referenced resources changed.
+// But a timestamp also changes when the executor merely adopts an orphaned
+// deployment, triggering an unwanted rolling update. The sum of the referenced
+// resources' resource versions reflects content changes without depending on
+// time, so adoption alone does not perturb it.
+//
+// Each reference is fetched by name (rather than listing every secret/configmap
+// in the namespace) to minimise API traffic. Behaviour is preserved: only
+// references resolvable within the deployment namespace contribute, and a
+// missing referenced object contributes nothing rather than erroring.
+//
+// The sum is returned as int64: resource versions are int64 etcd revisions and
+// can exceed 2^31-1 on long-running clusters, so narrowing to int would risk
+// truncation (and collisions that mask a genuine change).
+// The signature takes the whole FunctionSpec because the reference surface is
+// wider than Spec.Secrets/ConfigMaps: objects referenced only via
+// Env[].valueFrom / EnvFrom (RFC-0030) must contribute too — Kubernetes never
+// refreshes env in a running container, so a rotation the sum misses would
+// propagate to nothing (silently stale credentials).
+func ReferencedResourcesRVSum(ctx context.Context, client kubernetes.Interface, namespace string, spec fv1.FunctionSpec) (int64, error) {
+	var rvCount int64
+
+	// Deduplicate: the same object referenced via Spec.Secrets AND an env
+	// field must contribute once, or removing one of two references would
+	// change the sum and roll pods for a content-identical state.
+	secretNames := make(map[string]struct{})
+	for _, ref := range spec.Secrets {
+		if ref.Namespace == namespace {
+			secretNames[ref.Name] = struct{}{}
+		}
+	}
+	// Env references are same-namespace by construction (LocalObjectReference).
+	for _, name := range spec.EnvSecretNames() {
+		secretNames[name] = struct{}{}
+	}
+	cfgmapNames := make(map[string]struct{})
+	for _, ref := range spec.ConfigMaps {
+		if ref.Namespace == namespace {
+			cfgmapNames[ref.Name] = struct{}{}
+		}
+	}
+	for _, name := range spec.EnvConfigMapNames() {
+		cfgmapNames[name] = struct{}{}
+	}
+
+	for name := range secretNames {
+		s, err := client.CoreV1().Secrets(namespace).Get(ctx, name, metav1.GetOptions{})
+		if err != nil {
+			if k8serrors.IsNotFound(err) {
+				continue
+			}
+			return 0, err
+		}
+		rv, _ := strconv.ParseInt(s.ResourceVersion, 10, 64)
+		rvCount += rv
+	}
+
+	for name := range cfgmapNames {
+		cfg, err := client.CoreV1().ConfigMaps(namespace).Get(ctx, name, metav1.GetOptions{})
+		if err != nil {
+			if k8serrors.IsNotFound(err) {
+				continue
+			}
+			return 0, err
+		}
+		rv, _ := strconv.ParseInt(cfg.ResourceVersion, 10, 64)
+		rvCount += rv
+	}
+
+	return rvCount, nil
+}

pkg/executor/util/deployment_test.go

@@ -0,0 +1,241 @@
+// 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("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)
+}