From 7355ea9c65baac65a00d6e0e820e36f50ec499e7 Mon Sep 17 00:00:00 2001 From: upstream Date: Thu, 17 Sep 2026 15:59:49 +0200 Subject: [PATCH] Upstream fission/fission state of the touched files before dae21a228f --- pkg/executor/util/deployment.go | 198 ++++++++++++++++++++++ pkg/executor/util/deployment_test.go | 241 +++++++++++++++++++++++++++ 2 files changed, 439 insertions(+) create mode 100644 pkg/executor/util/deployment.go create mode 100644 pkg/executor/util/deployment_test.go diff --git a/pkg/executor/util/deployment.go b/pkg/executor/util/deployment.go new file mode 100644 index 0000000..49cb4f3 --- /dev/null +++ b/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 +} diff --git a/pkg/executor/util/deployment_test.go b/pkg/executor/util/deployment_test.go new file mode 100644 index 0000000..ccd4f8c --- /dev/null +++ b/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) +}