commit 4bd00b3eef47098d1dd9c19791cbaf3391002597 Author: upstream Date: Fri Sep 11 23:07:10 2026 +0200 Upstream fission/fission state of the touched files before d5703081e0 diff --git a/pkg/executor/executortype/container/deployment.go b/pkg/executor/executortype/container/deployment.go new file mode 100644 index 0000000..a4307b0 --- /dev/null +++ b/pkg/executor/executortype/container/deployment.go @@ -0,0 +1,270 @@ +// SPDX-FileCopyrightText: The Fission Authors +// +// SPDX-License-Identifier: Apache-2.0 + +package container + +import ( + "context" + "fmt" + "maps" + + appsv1 "k8s.io/api/apps/v1" + apiv1 "k8s.io/api/core/v1" + k8s_err "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/intstr" + + fv1 "github.com/fission/fission/pkg/apis/core/v1" + "github.com/fission/fission/pkg/executor/util" + "github.com/fission/fission/pkg/utils" + otelUtils "github.com/fission/fission/pkg/utils/otel" +) + +func (cn *Container) createOrGetDeployment(ctx context.Context, fn *fv1.Function, deployName string, deployLabels map[string]string, deployAnnotations map[string]string, deployNamespace string) (*appsv1.Deployment, error) { + logger := otelUtils.LoggerWithTraceID(ctx, cn.logger) + + // The specializationTimeout here refers to the creation of the pod and not the loading of function + // as in other executors. + specializationTimeout := fn.Spec.InvokeStrategy.ExecutionStrategy.SpecializationTimeout + minScale := int32(fn.Spec.InvokeStrategy.ExecutionStrategy.MinScale) + + // Always scale to at least one pod when createOrGetDeployment + // is called. The idleObjectReaper will scale-in the deployment + // later if no requests to the function. + if minScale <= 0 { + minScale = 1 + } + + deployment, err := cn.getDeploymentSpec(ctx, fn, &minScale, deployName, deployNamespace, deployLabels, deployAnnotations) + if err != nil { + return nil, err + } + + existingDepl, err := cn.kubernetesClient.AppsV1().Deployments(deployNamespace).Get(ctx, deployName, metav1.GetOptions{}) + if err != nil && !k8s_err.IsNotFound(err) { + return nil, err + } + + // Create new deployment if one does not previously exist + if k8s_err.IsNotFound(err) { + depl, err := cn.kubernetesClient.AppsV1().Deployments(deployNamespace).Create(ctx, deployment, metav1.CreateOptions{}) + if err != nil { + if k8s_err.IsAlreadyExists(err) { + depl, err = cn.kubernetesClient.AppsV1().Deployments(deployNamespace).Get(ctx, deployName, metav1.GetOptions{}) + } + if err != nil { + logger.Error(err, "error while creating function deployment", "function", fn.Name, + "deployment_name", deployName, + "deployment_namespace", deployNamespace) + return nil, err + } + } + otelUtils.SpanTrackEvent(ctx, "deploymentCreated", otelUtils.GetAttributesForDeployment(depl)...) + if minScale > 0 { + depl, err = util.WaitForDeployment(ctx, cn.kubernetesClient, cn.logger, depl, minScale, specializationTimeout) + } + return depl, err + } + + // Try to adopt orphan deployment created by the old executor. + if existingDepl.Annotations[fv1.EXECUTOR_INSTANCEID_LABEL] != cn.instanceID { + existingDepl.Annotations = deployment.Annotations + existingDepl.Labels = deployment.Labels + existingDepl.OwnerReferences = deployment.OwnerReferences + existingDepl.Spec.Template.Spec.Containers = deployment.Spec.Template.Spec.Containers + existingDepl.Spec.Template.Spec.ServiceAccountName = deployment.Spec.Template.Spec.ServiceAccountName + existingDepl.Spec.Template.Spec.TerminationGracePeriodSeconds = deployment.Spec.Template.Spec.TerminationGracePeriodSeconds + + // Update with the latest deployment spec. Kubernetes will trigger + // rolling update if spec is different from the one in the cluster. + existingDepl, err = cn.kubernetesClient.AppsV1().Deployments(deployNamespace).Update(ctx, existingDepl, metav1.UpdateOptions{}) + if err != nil { + logger.Error(err, "error adopting cn", "cn", deployName, "ns", deployNamespace) + return nil, err + } + // In this case, we just return without waiting for it for fast bootstraping. + return existingDepl, nil + } + + if *existingDepl.Spec.Replicas < minScale { + err = util.ScaleDeployment(ctx, cn.kubernetesClient, cn.logger, existingDepl.Namespace, existingDepl.Name, minScale) + if err != nil { + logger.Error(err, "error scaling up function deployment", "function", fn.Name) + return nil, err + } + } + if existingDepl.Status.AvailableReplicas < minScale { + existingDepl, err = util.WaitForDeployment(ctx, cn.kubernetesClient, cn.logger, existingDepl, minScale, specializationTimeout) + } + + return existingDepl, err +} + +func (cn *Container) updateDeployment(ctx context.Context, deployment *appsv1.Deployment, ns string) error { + _, err := cn.kubernetesClient.AppsV1().Deployments(ns).Update(ctx, deployment, metav1.UpdateOptions{}) + return err +} + +func (cn *Container) deleteDeployment(ctx context.Context, ns string, name string) error { + // DeletePropagationBackground deletes the object immediately and dependent are deleted later + // DeletePropagationForeground not advisable; it marks for deletion and API can still serve those objects + deletePropagation := metav1.DeletePropagationBackground + return cn.kubernetesClient.AppsV1().Deployments(ns).Delete(ctx, name, metav1.DeleteOptions{ + PropagationPolicy: &deletePropagation, + }) +} + +func (cn *Container) getDeploymentSpec(ctx context.Context, fn *fv1.Function, targetReplicas *int32, + deployName string, deployNamespace string, deployLabels map[string]string, deployAnnotations map[string]string) (*appsv1.Deployment, error) { + + replicas := int32(fn.Spec.InvokeStrategy.ExecutionStrategy.MinScale) + if targetReplicas != nil { + replicas = *targetReplicas + } + + if fn.Spec.PodSpec == nil { + return nil, fmt.Errorf("podSpec is not set for function %s", fn.Name) + } + + gracePeriodSeconds := int64(6 * 60) + if fn.Spec.PodSpec.TerminationGracePeriodSeconds != nil && *fn.Spec.PodSpec.TerminationGracePeriodSeconds >= 0 { + gracePeriodSeconds = *fn.Spec.PodSpec.TerminationGracePeriodSeconds + } + + podAnnotations := make(map[string]string) + + if cn.useIstio { + podAnnotations["sidecar.istio.io/inject"] = "false" + } + + podLabels := make(map[string]string) + + maps.Copy(podLabels, deployLabels) + + // Set maxUnavailable and maxSurge to 20% is because we want + // fission to rollout newer function version gradually without + // affecting any online service. For example, if you set maxSurge + // to 100%, the new ReplicaSet scales up immediately and may + // consume all remaining compute resources which might be an + // issue if a cluster's resource is on a budget. + // TODO: add to ExecutionStrategy so that the user + // can do more fine control over different functions. + maxUnavailable := intstr.FromString("20%") + maxSurge := intstr.FromString("20%") + + // Container updates the environment variable "LastUpdateTimestamp" of deployment + // whenever a configmap/secret gets an update, but it also leaves multiple ReplicaSets for + // rollback purpose. Since fission always update a deployment instead of performing a + // rollback, set RevisionHistoryLimit to 0 to disable this feature. + revisionHistoryLimit := int32(0) + + resources := cn.getResources(fn) + + // Other executor types rely on Environments to add configmaps and secrets. + // When the function declares RFC-0030 EnvFrom, that field is authoritative + // and the legacy whole-object synthesis from the name-only references is + // skipped — one source of truth, no double-injection (RFC-0030 §2). The + // legacy synthesis is preserved byte-for-byte when the new field is absent. + var envFromSources []apiv1.EnvFromSource + if len(fn.Spec.EnvFrom) == 0 { + var err error + envFromSources, err = util.ConvertConfigSecrets(ctx, fn, cn.kubernetesClient) + if err != nil { + return nil, err + } + } + + rvCount, err := util.ReferencedResourcesRVSum(ctx, cn.kubernetesClient, fn.Namespace, fn.Spec) + if err != nil { + return nil, err + } + + // Platform-owned env for the user container, computed once: it seeds the + // container AND is re-appended last by ApplyFunctionEnv below, whose + // kubelet-last-wins ordering is what keeps these names authoritative. + platformEnv := []apiv1.EnvVar{ + { + Name: fv1.ResourceVersionCount, + Value: fmt.Sprintf("%d", rvCount), + }, + } + + container := &apiv1.Container{ + Name: fn.Name, + ImagePullPolicy: cn.runtimeImagePullPolicy, + TerminationMessagePath: "/dev/termination-log", + // Connection-draining preStop hook; see utils.DrainLifecycle. + Lifecycle: utils.DrainLifecycle(gracePeriodSeconds), + Env: platformEnv, + EnvFrom: envFromSources, + // https://istio.io/docs/setup/kubernetes/additional-setup/requirements/ + Resources: resources, + } + // RFC-0030 §4 files mode. Poolmgr and newdeploy get this from the fetcher; + // the container executor has none, so the same layout is produced with + // native projected volumes. Built from the validated reference rather than + // taken from pod-spec input, and applied BEFORE MergePodSpec so a + // user-supplied podSpec cannot redirect a platform-constructed mount. + mpVolumes, mpMounts, err := util.MountPathVolumes(fn) + if err != nil { + return nil, err + } + container.VolumeMounts = append(container.VolumeMounts, mpMounts...) + + podSpec, err := util.MergePodSpec(&apiv1.PodSpec{ + Containers: []apiv1.Container{*container}, + Volumes: mpVolumes, + TerminationGracePeriodSeconds: &gracePeriodSeconds, + }, fn.Spec.PodSpec) + if err != nil { + return nil, err + } + pod := apiv1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{ + Labels: podLabels, + Annotations: podAnnotations, + }, + Spec: *podSpec, + } + + pod.Spec = *(util.ApplyImagePullSecret("", pod.Spec)) + + // After MergePodSpec, so the user's own podspec cannot reorder the + // platform-last guarantee (env podspec < EnvFrom < Env, platform above all). + if err := util.ApplyFunctionEnv(&pod.Spec, fn.Name, deployNamespace, fn, platformEnv); err != nil { + return nil, err + } + + var ownerReferences []metav1.OwnerReference + if cn.enableOwnerReferences { + ownerReferences = []metav1.OwnerReference{ + *metav1.NewControllerRef(fn, fv1.SchemeGroupVersion.WithKind("Function")), + } + } + deployment := &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{ + Name: deployName, + Labels: deployLabels, + Annotations: deployAnnotations, + OwnerReferences: ownerReferences, + }, + Spec: appsv1.DeploymentSpec{ + Replicas: &replicas, + Selector: &metav1.LabelSelector{ + MatchLabels: deployLabels, + }, + Template: pod, + Strategy: appsv1.DeploymentStrategy{ + Type: appsv1.RollingUpdateDeploymentStrategyType, + RollingUpdate: &appsv1.RollingUpdateDeployment{ + MaxUnavailable: &maxUnavailable, + MaxSurge: &maxSurge, + }, + }, + RevisionHistoryLimit: &revisionHistoryLimit, + }, + } + + return deployment, nil +} diff --git a/pkg/executor/executortype/container/deployment_test.go b/pkg/executor/executortype/container/deployment_test.go new file mode 100644 index 0000000..d298c95 --- /dev/null +++ b/pkg/executor/executortype/container/deployment_test.go @@ -0,0 +1,308 @@ +// SPDX-FileCopyrightText: The Fission Authors +// +// SPDX-License-Identifier: Apache-2.0 + +package container + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + appsv1 "k8s.io/api/apps/v1" + apiv1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes/fake" + + fv1 "github.com/fission/fission/pkg/apis/core/v1" + ferror "github.com/fission/fission/pkg/error" + "github.com/fission/fission/pkg/executor/fscache" + hpautils "github.com/fission/fission/pkg/executor/util/hpa" + fissionfake "github.com/fission/fission/pkg/generated/clientset/versioned/fake" + "github.com/fission/fission/pkg/utils" + "github.com/fission/fission/pkg/utils/loggerfactory" +) + +func newTestContainer() *Container { + return &Container{ + logger: loggerfactory.GetLogger(), + kubernetesClient: fake.NewClientset(), + runtimeImagePullPolicy: apiv1.PullIfNotPresent, + } +} + +func newTestContainerFunction() *fv1.Function { + grace := int64(30) + return &fv1.Function{ + Name: "ctr-fn", Namespace: "default", + Spec: fv1.FunctionSpec{ + InvokeStrategy: fv1.InvokeStrategy{ + ExecutionStrategy: fv1.ExecutionStrategy{ExecutorType: fv1.ExecutorTypeContainer, MinScale: 1, MaxScale: 3}, + }, + PodSpec: &apiv1.PodSpec{ + TerminationGracePeriodSeconds: &grace, + Containers: []apiv1.Container{ + {Name: "ctr-fn", Image: "example/app:latest"}, + }, + }, + }, + } +} + +func findContainer(pod apiv1.PodSpec, name string) *apiv1.Container { + for i := range pod.Containers { + if pod.Containers[i].Name == name { + return &pod.Containers[i] + } + } + return nil +} + +func TestContainerGetDeploymentSpec(t *testing.T) { + t.Run("builds a deployment from the function pod spec", func(t *testing.T) { + cn := newTestContainer() + fn := newTestContainerFunction() + replicas := int32(1) + + deployment, err := cn.getDeploymentSpec(t.Context(), fn, &replicas, "ctr-fn", "default", + map[string]string{"app": "ctr"}, map[string]string{"note": "x"}) + require.NoError(t, err) + + assert.Equal(t, "ctr-fn", deployment.Name) + assert.Equal(t, map[string]string{"app": "ctr"}, deployment.Labels) + assert.Equal(t, map[string]string{"note": "x"}, deployment.Annotations) + require.NotNil(t, deployment.Spec.Replicas) + assert.Equal(t, int32(1), *deployment.Spec.Replicas) + assert.Equal(t, map[string]string{"app": "ctr"}, deployment.Spec.Selector.MatchLabels) + require.NotNil(t, deployment.Spec.RevisionHistoryLimit) + assert.Equal(t, int32(0), *deployment.Spec.RevisionHistoryLimit) + assert.Equal(t, appsv1.RollingUpdateDeploymentStrategyType, deployment.Spec.Strategy.Type) + + pod := deployment.Spec.Template.Spec + ctr := findContainer(pod, "ctr-fn") + require.NotNil(t, ctr, "user container must be present") + assert.Equal(t, "example/app:latest", ctr.Image) + assert.Equal(t, apiv1.PullIfNotPresent, ctr.ImagePullPolicy) + require.NotNil(t, pod.TerminationGracePeriodSeconds) + assert.Equal(t, int64(30), *pod.TerminationGracePeriodSeconds) + }) + + t.Run("nil targetReplicas falls back to MinScale", func(t *testing.T) { + cn := newTestContainer() + fn := newTestContainerFunction() + fn.Spec.InvokeStrategy.ExecutionStrategy.MinScale = 2 + + deployment, err := cn.getDeploymentSpec(t.Context(), fn, nil, "ctr-fn", "default", nil, nil) + require.NoError(t, err) + require.NotNil(t, deployment.Spec.Replicas) + assert.Equal(t, int32(2), *deployment.Spec.Replicas) + }) + + t.Run("nil TerminationGracePeriodSeconds falls back to the default", func(t *testing.T) { + cn := newTestContainer() + fn := newTestContainerFunction() + fn.Spec.PodSpec.TerminationGracePeriodSeconds = nil + replicas := int32(1) + + deployment, err := cn.getDeploymentSpec(t.Context(), fn, &replicas, "ctr-fn", "default", nil, nil) + require.NoError(t, err) + + pod := deployment.Spec.Template.Spec + require.NotNil(t, pod.TerminationGracePeriodSeconds) + assert.Equal(t, int64(6*60), *pod.TerminationGracePeriodSeconds) + }) + + t.Run("nil PodSpec returns an error instead of panicking", func(t *testing.T) { + cn := newTestContainer() + fn := newTestContainerFunction() + fn.Spec.PodSpec = nil + replicas := int32(1) + + _, err := cn.getDeploymentSpec(t.Context(), fn, &replicas, "ctr-fn", "default", nil, nil) + require.Error(t, err) + }) + + t.Run("istio disables sidecar injection", func(t *testing.T) { + cn := newTestContainer() + cn.useIstio = true + fn := newTestContainerFunction() + replicas := int32(1) + + deployment, err := cn.getDeploymentSpec(t.Context(), fn, &replicas, "ctr-fn", "default", nil, nil) + require.NoError(t, err) + assert.Equal(t, "false", deployment.Spec.Template.Annotations["sidecar.istio.io/inject"]) + }) + + t.Run("owner references are added when enabled", func(t *testing.T) { + cn := newTestContainer() + cn.enableOwnerReferences = true + fn := newTestContainerFunction() + replicas := int32(1) + + deployment, err := cn.getDeploymentSpec(t.Context(), fn, &replicas, "ctr-fn", "default", nil, nil) + require.NoError(t, err) + require.Len(t, deployment.OwnerReferences, 1) + assert.Equal(t, "Function", deployment.OwnerReferences[0].Kind) + assert.Equal(t, "ctr-fn", deployment.OwnerReferences[0].Name) + }) +} + +// TestContainerFnDeleteCacheMiss verifies that deleting a function whose fsvc +// is absent from the in-memory cache (never specialized, evicted, or executor +// restarted) does not error out — it must fall back to the deterministic +// computed object name and clean up the backing objects instead of leaking +// them. +func TestContainerFnDeleteCacheMiss(t *testing.T) { + logger := loggerfactory.GetLogger() + cn := &Container{ + logger: logger, + kubernetesClient: fake.NewClientset(), + fsCache: fscache.MakeFunctionServiceCache(logger), + nsResolver: utils.DefaultNSResolver(), + hpaops: hpautils.NewHpaOperations(logger, fake.NewClientset(), "test-instance"), + } + + fn := newTestContainerFunction() + fn.UID = "abcdef01-2345-6789-abcd-ef0123456789" + + // fsCache is intentionally left empty so ListByFunctionUID finds nothing. + require.NoError(t, cn.fnDelete(t.Context(), fn), + "fnDelete must tolerate a cache miss and clean up by computed name") +} + +// TestContainerFnCreateDeletionGuard verifies the authoritative re-read at the +// top of fnCreate refuses to create backing objects for a Function that the +// router presented but which is gone or being deleted in the cluster. Without +// this guard an in-flight create can race the delete teardown and re-create the +// Deployment/Service/HPA after teardown removed them, leaking objects whose +// owning Function CR is already gone. +func TestContainerFnCreateDeletionGuard(t *testing.T) { + t.Parallel() + + const liveUID = "abcdef01-2345-6789-abcd-ef0123456789" + + tests := []struct { + name string + // stored is the Function in the authoritative store (fake + // fissionClient); nil means the function is absent. + stored *fv1.Function + }{ + { + name: "function absent from fissionClient", + stored: nil, + }, + { + name: "function present but being deleted", + stored: func() *fv1.Function { + fn := newTestContainerFunction() + fn.UID = liveUID + fn.DeletionTimestamp = &metav1.Time{Time: time.Now()} + fn.Finalizers = []string{"fission.io/test"} + return fn + }(), + }, + { + name: "function present but with a different UID", + stored: func() *fv1.Function { + fn := newTestContainerFunction() + fn.UID = "00000000-0000-0000-0000-000000000000" + return fn + }(), + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + logger := loggerfactory.GetLogger() + kubeClient := fake.NewClientset() + fissionClient := fissionfake.NewSimpleClientset() + if tc.stored != nil { + fissionClient = fissionfake.NewSimpleClientset(tc.stored) + } + caaf := &Container{ + logger: logger, + kubernetesClient: kubeClient, + fissionClient: fissionClient, + fsCache: fscache.MakeFunctionServiceCache(logger), + nsResolver: utils.DefaultNSResolver(), + hpaops: hpautils.NewHpaOperations(logger, kubeClient, "test-instance"), + } + + // The Function the router presents looks alive (no DeletionTimestamp). + fn := newTestContainerFunction() + fn.UID = liveUID + + _, err := caaf.fnCreate(t.Context(), fn) + require.Error(t, err, "fnCreate must refuse a gone/deleting function") + assert.True(t, ferror.IsNotFound(err), + "guard must surface a ferror NotFound, got: %v", err) + + deploys, listErr := kubeClient.AppsV1().Deployments(metav1.NamespaceAll).List(t.Context(), metav1.ListOptions{}) + require.NoError(t, listErr) + assert.Empty(t, deploys.Items, + "no Deployment must be created when the function is gone/deleting") + }) + } +} + +// TestContainerFnCreateGuardPass verifies the guard lets a live, matching +// Function proceed past the re-read. We do not stand up a full happy-path +// (running pods); proving the guard passed is enough: fnCreate proceeds to +// create the backing objects and then fails waiting for the deployment to +// become available once the bounded context expires — a DeadlineExceeded +// error, which is not a ferror NotFound. +func TestContainerFnCreateGuardPass(t *testing.T) { + t.Parallel() + + // A container function needs exactly one container port for the Service to + // be created, so the create path can proceed past the guard to the + // deployment wait. + withPort := func(fn *fv1.Function) *fv1.Function { + fn.Spec.PodSpec.Containers[0].Ports = []apiv1.ContainerPort{{ContainerPort: 8080}} + return fn + } + + live := withPort(newTestContainerFunction()) + live.UID = "abcdef01-2345-6789-abcd-ef0123456789" + + logger := loggerfactory.GetLogger() + kubeClient := fake.NewClientset() + caaf := &Container{ + logger: logger, + kubernetesClient: kubeClient, + fissionClient: fissionfake.NewSimpleClientset(live), + fsCache: fscache.MakeFunctionServiceCache(logger), + nsResolver: utils.DefaultNSResolver(), + hpaops: hpautils.NewHpaOperations(logger, kubeClient, "test-instance"), + } + + fn := withPort(newTestContainerFunction()) + fn.UID = live.UID + + // Bound the context so the (never-ready, no real pods) deployment wait + // returns promptly instead of blocking the default specialization timeout. + ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second) + defer cancel() + + _, err := caaf.fnCreate(ctx, fn) + require.Error(t, err, "expected fnCreate to fail past the guard waiting for the deployment") + assert.False(t, ferror.IsNotFound(err), + "guard must have passed; the error should come from the deployment wait, not the guard. got: %v", err) + // The non-NotFound error proves the guard passed: fnCreate proceeded past + // the re-read into createOrGetDeployment and only failed at the bounded + // deployment-readiness wait ("context deadline exceeded"). + assert.ErrorIs(t, err, context.DeadlineExceeded, + "guard-pass should fail at the deployment wait, got: %v", err) +} + +func TestContainerGetResources(t *testing.T) { + cn := newTestContainer() + fn := newTestContainerFunction() + // no resource requests/limits set -> maps initialised, not nil + res := cn.getResources(fn) + assert.NotNil(t, res.Requests) + assert.NotNil(t, res.Limits) +}