Upstream fission/fission state of the touched files before d5703081e0

This commit is contained in:
upstream committed 2026-09-11 23:07:10 +02:00
commit 4bd00b3eef
2 files changed
+578

No files matched your search

@@ -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
}
@@ -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)
}