diff --git a/pkg/executor/executortype/container/containermgr.go b/pkg/executor/executortype/container/containermgr.go new file mode 100644 index 0000000..25d2761 --- /dev/null +++ b/pkg/executor/executortype/container/containermgr.go @@ -0,0 +1,827 @@ +// SPDX-FileCopyrightText: The Fission Authors +// +// SPDX-License-Identifier: Apache-2.0 + +package container + +import ( + "context" + "errors" + "fmt" + "os" + "reflect" + "strconv" + "strings" + "time" + + appsv1 "k8s.io/api/apps/v1" + apiv1 "k8s.io/api/core/v1" + apiequality "k8s.io/apimachinery/pkg/api/equality" + k8sErrs "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" + k8sTypes "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/kubernetes" + k8sCache "k8s.io/client-go/tools/cache" + "sigs.k8s.io/controller-runtime/pkg/client" + + "github.com/go-logr/logr" + "golang.org/x/sync/errgroup" + + fv1 "github.com/fission/fission/pkg/apis/core/v1" + "github.com/fission/fission/pkg/crd" + ferror "github.com/fission/fission/pkg/error" + "github.com/fission/fission/pkg/executor/executortype" + "github.com/fission/fission/pkg/executor/fscache" + "github.com/fission/fission/pkg/executor/metrics" + "github.com/fission/fission/pkg/executor/reaper" + "github.com/fission/fission/pkg/executor/reaper/idle" + executorUtils "github.com/fission/fission/pkg/executor/util" + hpautils "github.com/fission/fission/pkg/executor/util/hpa" + "github.com/fission/fission/pkg/generated/clientset/versioned" + "github.com/fission/fission/pkg/throttler" + "github.com/fission/fission/pkg/utils" + "github.com/fission/fission/pkg/utils/maps" + otelUtils "github.com/fission/fission/pkg/utils/otel" +) + +var ( + _ executortype.ExecutorType = &Container{} +) + +type ( + // Container represents an executor type + Container struct { + logger logr.Logger + + kubernetesClient kubernetes.Interface + fissionClient versioned.Interface + instanceID string + nsResolver *utils.NamespaceResolver + + runtimeImagePullPolicy apiv1.PullPolicy + useIstio bool + + fsCache *fscache.FunctionServiceCache // cache funcSvc's by function, address and pod name + + throttler *throttler.Throttler + + defaultIdlePodReapTime time.Duration + + // crClient is the executor Manager's cache-backed client, used by IsValid + // to read function Deployments/Services from the shared Manager cache + // (replacing this type's standalone SharedInformerFactory listers). Set in + // RegisterReconcilers once the Manager exists. + crClient client.Client + + hpaops *hpautils.HpaOperations + objectReaperIntervalSecond time.Duration + + enableOwnerReferences bool + } +) + +// MakeContainer initializes and returns an instance of CaaF +func MakeContainer( + ctx context.Context, + logger logr.Logger, + fissionClient versioned.Interface, + kubernetesClient kubernetes.Interface, + instanceID string, +) (executortype.ExecutorType, error) { + enableIstio := false + if len(os.Getenv("ENABLE_ISTIO")) > 0 { + istio, err := strconv.ParseBool(os.Getenv("ENABLE_ISTIO")) + if err != nil { + logger.Error(err, "failed to parse 'ENABLE_ISTIO', set to false") + } + enableIstio = istio + } + + caaf := &Container{ + logger: logger.WithName("CaaF"), + + fissionClient: fissionClient, + kubernetesClient: kubernetesClient, + instanceID: instanceID, + nsResolver: utils.DefaultNSResolver(), + + fsCache: fscache.MakeFunctionServiceCache(logger), + throttler: throttler.MakeThrottler(1 * time.Minute), + + runtimeImagePullPolicy: utils.GetImagePullPolicy(os.Getenv("RUNTIME_IMAGE_PULL_POLICY")), + useIstio: enableIstio, + // Time is set slightly higher than NewDeploy as cold starts are longer for CaaF + defaultIdlePodReapTime: 1 * time.Minute, + objectReaperIntervalSecond: time.Duration(executorUtils.GetObjectReaperInterval(logger, fv1.ExecutorTypeContainer, 5)) * time.Second, + hpaops: hpautils.NewHpaOperations(logger, kubernetesClient, instanceID), + + enableOwnerReferences: utils.IsOwnerReferencesEnabled(), + } + + // The Function watch is a controller-runtime reconciler registered via the + // shared funcreconciler on the executor Manager (see reconciler.go). + // Deployment/Service reads (IsValid) go through the Manager's cache-backed + // client (caaf.crClient), set in RegisterReconcilers — no per-type informer + // factory is needed. + return caaf, nil +} + +// Run is a no-op: the container manager no longer runs its own informer factory. +// Its Deployment/Service reads (IsValid) go through the executor Manager's cache, +// which controller-runtime syncs before any runnable (including this type's +// reapers) starts. +func (caaf *Container) Run(context.Context, *errgroup.Group) {} + +// GetTypeName returns the executor type name. +func (caaf *Container) GetTypeName(ctx context.Context) fv1.ExecutorType { + return fv1.ExecutorTypeContainer +} + +// UnTapService has not been implemented for CaaF. +func (caaf *Container) UnTapService(ctx context.Context, fnMeta *metav1.ObjectMeta, svcHost string) { + // Not Implemented for CaaF. +} + +// MarkSpecializationFailure has not been implemented for CaaF. +func (caaf *Container) MarkSpecializationFailure(ctx context.Context, fnMeta *metav1.ObjectMeta) { + // Not Implemented for CaaF. +} + +// GetFuncSvc returns a function service; error otherwise. +func (caaf *Container) GetFuncSvc(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) { + return caaf.createFunction(ctx, fn) +} + +// GetFuncSvcFromCache returns a function service from cache; error otherwise. +// The lookup is keyed (UID, Generation), never bare UID: fn may be an +// RFC-0025 versioned projection pinning an older Generation of a UID whose +// live entry is also cached, and a UID-only lookup would return whichever +// generation's fsvc was cached last — silently serving the wrong version's +// container (IsValid only checks the backing objects exist). +func (caaf *Container) GetFuncSvcFromCache(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) { + otelUtils.SpanTrackEvent(ctx, "GetFuncSvcFromCache", otelUtils.GetAttributesForFunction(fn)...) + return caaf.fsCache.GetByFunction(&fn.ObjectMeta) +} + +// DeleteFuncSvcFromCache deletes a function service from cache. +func (caaf *Container) DeleteFuncSvcFromCache(ctx context.Context, fsvc *fscache.FuncSvc) { + caaf.fsCache.DeleteEntry(fsvc) +} + +// TapService makes a TouchByAddress request to the cache. +func (caaf *Container) TapService(ctx context.Context, svcHost string) error { + err := caaf.fsCache.TouchByAddress(svcHost) + if err != nil { + return err + } + return nil +} + +// IsValid does a get on the service address to ensure it's a valid service, then +// scale deployment to 1 replica if there are no available replicas for function. +// Return true if no error occurs, return false otherwise. +func (caaf *Container) IsValid(ctx context.Context, fsvc *fscache.FuncSvc) bool { + logger := otelUtils.LoggerWithTraceID(ctx, caaf.logger) + otelUtils.SpanTrackEvent(ctx, "IsValid", fscache.GetAttributesForFuncSvc(fsvc)...) + if len(strings.Split(fsvc.Address, ".")) == 0 { + logger.Info("address not found in function service") + return false + } + if len(fsvc.KubernetesObjects) == 0 { + logger.Info("no kubernetes object related to function", "function", fsvc.Function.Name) + return false + } + for _, obj := range fsvc.KubernetesObjects { + objKey := client.ObjectKey{Namespace: obj.Namespace, Name: obj.Name} + if strings.ToLower(obj.Kind) == "service" { + if err := caaf.crClient.Get(ctx, objKey, &apiv1.Service{}); err != nil { + if !k8sErrs.IsNotFound(err) { + logger.Error(err, "error validating function service", "function", fsvc.Function.Name) + } + return false + } + } else if strings.ToLower(obj.Kind) == "deployment" { + currentDeploy := &appsv1.Deployment{} + if err := caaf.crClient.Get(ctx, objKey, currentDeploy); err != nil { + if !k8sErrs.IsNotFound(err) { + logger.Error(err, "error validating function deployment", "function", fsvc.Function.Name) + } + return false + } + if currentDeploy.Status.AvailableReplicas < 1 { + return false + } + } + } + return true +} + +// RefreshFuncPods deletes pods related to the function so that new pods are replenished +func (caaf *Container) RefreshFuncPods(ctx context.Context, logger logr.Logger, f fv1.Function) error { + + funcLabels := caaf.getDeployLabels(f.ObjectMeta) + + nsResolver := utils.DefaultNSResolver() + dep, err := caaf.kubernetesClient.AppsV1().Deployments(nsResolver.GetFunctionNS(f.ObjectMeta.Namespace)).List(ctx, metav1.ListOptions{ + LabelSelector: labels.Set(funcLabels).AsSelector().String(), + }) + if err != nil { + return err + } + + // Ideally there should be only one deployment but for now we rely on label/selector to ensure that condition + for _, deployment := range dep.Items { + rvCount, err := executorUtils.ReferencedResourcesRVSum(ctx, caaf.kubernetesClient, deployment.Namespace, f.Spec) + if err != nil { + return err + } + + patch := fmt.Sprintf(`{"spec" : {"template": {"spec":{"containers":[{"name": "%s", "env":[{"name": "%s", "value": "%d"}]}]}}}}`, + f.Name, fv1.ResourceVersionCount, rvCount) + + _, err = caaf.kubernetesClient.AppsV1().Deployments(deployment.ObjectMeta.Namespace).Patch(ctx, deployment.Name, + k8sTypes.StrategicMergePatchType, + []byte(patch), metav1.PatchOptions{}) + if err != nil { + return err + } + } + return nil +} + +// AdoptExistingResources re-claims the function objects this executor type left +// behind on a previous run. It routes through the throttled createFunction (not +// fnCreate directly), so the adopt pass and the Function reconciler — which also +// re-stamps existing objects on its initial sync — single-flight per function +// UID instead of racing on the in-place Update (which previously produced +// resourceVersion conflicts that tripped fnCreate's destroy-on-error path). +func (caaf *Container) AdoptExistingResources(ctx context.Context) { + executorUtils.AdoptFunctions(ctx, caaf.logger, caaf.fissionClient, fv1.ExecutorTypeContainer, + func(ctx context.Context, fn *fv1.Function) error { + _, err := caaf.createFunction(ctx, fn) + return err + }) +} + +// CleanupOldExecutorObjects cleans orphaned resources. +func (caaf *Container) CleanupOldExecutorObjects(ctx context.Context) { + reaper.CleanupExecutorObjects(ctx, caaf.logger, caaf.kubernetesClient, caaf.instanceID, fv1.ExecutorTypeContainer) +} + +// resourcesExist reports whether the function's backing Deployment and Service +// are present in the Manager cache. A missing object means it drifted away +// (deleted out-of-band) and the function should be recreated. Reads go through +// the cache-backed crClient (same path as IsValid). +func (caaf *Container) resourcesExist(ctx context.Context, fn *fv1.Function) (bool, error) { + ns := caaf.nsResolver.GetFunctionNS(fn.Namespace) + key := client.ObjectKey{Namespace: ns, Name: caaf.getObjName(fn)} + for _, obj := range []client.Object{&appsv1.Deployment{}, &apiv1.Service{}} { + if err := caaf.crClient.Get(ctx, key, obj); err != nil { + if k8sErrs.IsNotFound(err) { + return false, nil + } + return false, err + } + } + return true, nil +} + +func (caaf *Container) createFunction(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) { + if fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fv1.ExecutorTypeContainer { + return nil, nil + } + + // Single-flight per (UID, Generation), not per UID: RFC-0025 versioned + // projections share the live function's UID but create distinct + // per-version Deployments, so their creates must neither serialize + // behind each other nor have a loser handed another generation's fsvc + // from the cache fallback. + fsvc, err := caaf.throttler.RunOnce(crd.CacheKeyUGFromMeta(&fn.ObjectMeta).String(), func(ableToCreate bool) (*fscache.FuncSvc, error) { + if ableToCreate { + return caaf.fnCreate(ctx, fn) + } + return caaf.fsCache.GetByFunction(&fn.ObjectMeta) + }) + if err != nil { + e := "error creating k8s resources for function" + caaf.logger.Error(err, e, "function_name", fn.Name, + "function_namespace", fn.Namespace) + return nil, fmt.Errorf("error creating k8s resources for function %s/%s: %w", fn.Namespace, fn.Name, err) + } + + return fsvc, nil +} + +func (caaf *Container) deleteFunction(ctx context.Context, fn *fv1.Function) error { + if fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fv1.ExecutorTypeContainer { + return nil + } + err := caaf.fnDelete(ctx, fn) + if err != nil { + return fmt.Errorf("error deleting kubernetes objects of function %s: %w", k8sCache.MetaObjectToName(fn), err) + } + return err +} + +// destroyOnCreateError reports whether a fnCreate failure warrants tearing down +// the function's partial resources. A Conflict or AlreadyExists means the object +// exists and was concurrently modified (e.g. the adopt pass racing a reconcile, +// or two reconciles), so deleting it would turn a transient blip into a cold +// recreate — leave it for the next reconcile to converge instead. Genuine +// creation failures (quota, invalid spec, API errors) still trigger cleanup so a +// brand-new function doesn't leak half-created objects. +func destroyOnCreateError(err error) bool { + // An explicitly cancelled context means the executor is shutting down, lost + // leadership, or the caller gave up — not that creation genuinely failed — + // so leave any partially-created resources for the next leader/request to + // adopt instead of tearing them down. A context *deadline* is different: on + // the specialization path the context carries the per-function + // SpecializationTimeout (see pkg/executor/executor.go), so DeadlineExceeded + // is a genuine timeout and falls through to normal cleanup. + if errors.Is(err, context.Canceled) { + return false + } + return !k8sErrs.IsConflict(err) && !k8sErrs.IsAlreadyExists(err) +} + +func (caaf *Container) fnCreate(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) { + // Authoritative re-read: the Function object reaching this path originates + // from the router's request body and can be stale — its DeletionTimestamp + // may be absent even though the Function is being deleted. Without this + // check a create can race the delete teardown (funcreconciler) and + // re-create the Deployment/Service/HPA *after* teardown removed them, + // leaking objects whose owning Function CR is already gone. + live, err := caaf.fissionClient.CoreV1().Functions(fn.Namespace).Get(ctx, fn.Name, metav1.GetOptions{}) + if err != nil { + if k8sErrs.IsNotFound(err) { + return nil, ferror.MakeError(ferror.ErrorNotFound, + fmt.Sprintf("function %s is gone, not creating service", k8sCache.MetaObjectToName(fn))) + } + return nil, err + } + if live.UID != fn.UID || !live.DeletionTimestamp.IsZero() { + return nil, ferror.MakeError(ferror.ErrorNotFound, + fmt.Sprintf("function %s is being deleted, not creating service", k8sCache.MetaObjectToName(fn))) + } + + cleanupFunc := func(ns string, name string) { + err := caaf.cleanupContainer(ctx, ns, name) + if err != nil { + caaf.logger.Error(err, "received error while cleaning function resources", + "namespace", ns, "name", name) + } + } + objName := caaf.getObjName(fn) + deployLabels := caaf.getDeployLabels(fn.ObjectMeta) + deployAnnotations := caaf.getDeployAnnotations(fn.ObjectMeta) + + // to support backward compatibility, if the function was created in default ns, we fall back to creating the + // deployment of the function in fission-function ns + ns := caaf.nsResolver.GetFunctionNS(fn.Namespace) + + // Envoy(istio-proxy) returns 404 directly before istio pilot + // propagates latest Envoy-specific configuration. + // Since Container waits for pods of deployment to be ready, + // change the order of kubeObject creation (create service first, + // then deployment) to take advantage of waiting time. + // Transient executor errors are not written to Function.Status; see + // the analogous note in pkg/executor/executortype/poolmgr/gp.go. + svc, err := caaf.createOrGetSvc(ctx, fn, deployLabels, deployAnnotations, objName, ns) + if err != nil { + caaf.logger.Error(err, "error creating service", "service", objName) + if destroyOnCreateError(err) { + go cleanupFunc(ns, objName) + } + return nil, fmt.Errorf("error creating service %s: %w", objName, err) + } + // Absolute FQDN — same rationale as the newdeploy site: keep the + // router's per-request dials off the resolv.conf search path. + svcAddress := utils.ServiceFQDN(svc.Name, svc.Namespace) + + depl, err := caaf.createOrGetDeployment(ctx, fn, objName, deployLabels, deployAnnotations, ns) + if err != nil { + caaf.logger.Error(err, "error creating deployment", "deployment", objName) + if destroyOnCreateError(err) { + go cleanupFunc(ns, objName) + } + return nil, fmt.Errorf("error creating deployment %s: %w", objName, err) + } + + // For container-type functions the user image runs as a single container + // named after the function (see deployment.go); scope HPA metrics to it. + hpa, err := caaf.hpaops.CreateOrGetHpa(ctx, fn, objName, &fn.Spec.InvokeStrategy.ExecutionStrategy, fn.Name, depl, deployLabels, deployAnnotations) + if err != nil { + caaf.logger.Error(err, "error creating HPA", "hpa", objName) + if destroyOnCreateError(err) { + go cleanupFunc(ns, objName) + } + return nil, fmt.Errorf("error creating HPA %s: %w", objName, err) + } + + kubeObjRefs := []apiv1.ObjectReference{ + { + // obj.TypeMeta.Kind does not work hence this, needs investigation and a fix + Kind: "deployment", + Name: depl.Name, + APIVersion: depl.APIVersion, + Namespace: depl.Namespace, + ResourceVersion: depl.ResourceVersion, + UID: depl.UID, + }, + { + Kind: "service", + Name: svc.Name, + APIVersion: svc.APIVersion, + Namespace: svc.Namespace, + ResourceVersion: svc.ResourceVersion, + UID: svc.UID, + }, + { + Kind: "horizontalpodautoscaler", + Name: hpa.Name, + APIVersion: hpa.APIVersion, + Namespace: hpa.Namespace, + ResourceVersion: hpa.ResourceVersion, + UID: hpa.UID, + }, + } + + fsvc := &fscache.FuncSvc{ + Name: objName, + Function: &fn.ObjectMeta, + Address: svcAddress, + KubernetesObjects: kubeObjRefs, + Executor: fv1.ExecutorTypeContainer, + } + + _, err = caaf.fsCache.Add(*fsvc) + if err != nil { + caaf.logger.Error(nil, "error adding function to cache", "function", fsvc.Function) + metrics.RecordColdStartError(ctx, fn.Name, fn.Namespace) + return fsvc, err + } + + metrics.RecordColdStart(ctx, fn.Name, fn.Namespace) + executorUtils.SetFunctionReady(ctx, caaf.logger, caaf.fissionClient, fn, fv1.FunctionReasonReady, "container deployment is ready") + + return fsvc, nil +} + +func (caaf *Container) updateFunction(ctx context.Context, oldFn *fv1.Function, newFn *fv1.Function) error { + + if oldFn.ResourceVersion == newFn.ResourceVersion { + return nil + } + + // Ignoring updates to functions which are not of Container type + if newFn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fv1.ExecutorTypeContainer && + oldFn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fv1.ExecutorTypeContainer { + return nil + } + + // Executor type is no longer Container + if newFn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fv1.ExecutorTypeContainer && + oldFn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fv1.ExecutorTypeContainer { + caaf.logger.Info("function does not use new deployment executor anymore, deleting resources", + "function", newFn) + // IMP - pass the oldFn, as the new/modified function is not in cache + return caaf.deleteFunction(ctx, oldFn) + } + + // Executor type changed to Container from something else + if oldFn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fv1.ExecutorTypeContainer && + newFn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fv1.ExecutorTypeContainer { + caaf.logger.Info("function type changed to Container, creating resources", + "old_function", oldFn.ObjectMeta, + "new_function", newFn.ObjectMeta) + _, err := caaf.createFunction(ctx, newFn) + if err != nil { + caaf.updateStatus(oldFn, err, "error changing the function's type to Container") + } + return err + } + + if !reflect.DeepEqual(oldFn.Spec.InvokeStrategy, newFn.Spec.InvokeStrategy) { + // to support backward compatibility, if the function was created in default ns, we fall back to creating the + // deployment of the function in fission-function ns, so cleaning up resources there + ns := caaf.nsResolver.GetFunctionNS(newFn.Namespace) + + // Live entry only: updates arrive for the live Function CR, and its + // versioned projections' entries (which share the UID) are immutable + // snapshots whose HPA must not be touched here. + fsvc, err := caaf.fsCache.GetLiveByFunctionUID(newFn.UID) + if err != nil { + return fmt.Errorf("error updating function due to unable to find function service cache %s: %w", k8sCache.MetaObjectToName(oldFn), err) + } + + hpa, err := caaf.hpaops.GetHpa(ctx, ns, fsvc.Name) + if err != nil { + caaf.updateStatus(oldFn, err, "error getting HPA while updating function") + return err + } + + hpaChanged := false + + if newFn.Spec.InvokeStrategy.ExecutionStrategy.MinScale != oldFn.Spec.InvokeStrategy.ExecutionStrategy.MinScale { + replicas := int32(newFn.Spec.InvokeStrategy.ExecutionStrategy.MinScale) + hpa.Spec.MinReplicas = &replicas + hpaChanged = true + } + + if newFn.Spec.InvokeStrategy.ExecutionStrategy.MaxScale != oldFn.Spec.InvokeStrategy.ExecutionStrategy.MaxScale { + hpa.Spec.MaxReplicas = int32(newFn.Spec.InvokeStrategy.ExecutionStrategy.MaxScale) + hpaChanged = true + } + + if !reflect.DeepEqual(newFn.Spec.InvokeStrategy.ExecutionStrategy.Metrics, oldFn.Spec.InvokeStrategy.ExecutionStrategy.Metrics) { + // The CLI emits pod-wide Resource metrics; normalize them to + // ContainerResource metrics scoped to the function container, the + // same as createFunction does via CreateOrGetHpa. + hpa.Spec.Metrics = hpautils.RewriteResourceMetricsToContainer( + newFn.Spec.InvokeStrategy.ExecutionStrategy.Metrics, newFn.Name, caaf.logger) + hpaChanged = true + } + + if !reflect.DeepEqual(newFn.Spec.InvokeStrategy.ExecutionStrategy.Behavior, oldFn.Spec.InvokeStrategy.ExecutionStrategy.Behavior) { + hpa.Spec.Behavior = newFn.Spec.InvokeStrategy.ExecutionStrategy.Behavior + hpaChanged = true + } + + if hpaChanged { + err := caaf.hpaops.UpdateHpa(ctx, hpa) + if err != nil { + caaf.updateStatus(oldFn, err, "error updating HPA while updating function") + return err + } + } + } + + deployChanged := false + + // If length of slice has changed then no need to check individual elements + if len(oldFn.Spec.Secrets) != len(newFn.Spec.Secrets) { + deployChanged = true + } else { + for i, newSecret := range newFn.Spec.Secrets { + if newSecret != oldFn.Spec.Secrets[i] { + deployChanged = true + break + } + } + } + if len(oldFn.Spec.ConfigMaps) != len(newFn.Spec.ConfigMaps) { + deployChanged = true + } else { + for i, newConfig := range newFn.Spec.ConfigMaps { + if newConfig != oldFn.Spec.ConfigMaps[i] { + deployChanged = true + break + } + } + } + + if !reflect.DeepEqual(oldFn.Spec.PodSpec, newFn.Spec.PodSpec) { + deployChanged = true + } + + // RFC-0030 per-function env lands in the pod template — see the newdeploy + // counterpart: an env-only edit must roll the deployment, or the CR and + // the running container disagree indefinitely. + if !reflect.DeepEqual(oldFn.Spec.Env, newFn.Spec.Env) || + !reflect.DeepEqual(oldFn.Spec.EnvFrom, newFn.Spec.EnvFrom) { + deployChanged = true + } + + if deployChanged { + return caaf.updateFuncDeployment(ctx, newFn) + } + + return nil +} + +// reconcileDeploymentSpec brings an already-existing deployment up to the +// function's current spec when it lags — the container-type mirror of +// newdeploy's helper, and the level-trigger half of every create-routed +// reconcile (see createAndConverge in reconciler.go for why it must follow +// createFunction). The deployment carries the function's ResourceVersion as +// an annotation (getDeployAnnotations); compare it and push the current spec +// when stale. A no-op when already current. +func (caaf *Container) reconcileDeploymentSpec(ctx context.Context, fn *fv1.Function) error { + // Live entry only: versioned projections sharing the UID pin immutable + // snapshots that must not be rewritten to the live spec. + fsvc, err := caaf.fsCache.GetLiveByFunctionUID(fn.UID) + if err != nil { + // Not specialized yet — no deployment to reconcile; the on-demand path + // creates it from the current spec on first invocation. + return nil + } + ns := caaf.nsResolver.GetFunctionNS(fn.Namespace) + existingDepl, err := caaf.kubernetesClient.AppsV1().Deployments(ns).Get(ctx, fsvc.Name, metav1.GetOptions{}) + if err != nil { + if k8sErrs.IsNotFound(err) { + return nil + } + return err + } + deplRV := existingDepl.Annotations[fv1.FUNCTION_RESOURCE_VERSION] + if deplRV == fn.ResourceVersion { + return nil // deployment already reflects the current function spec + } + caaf.logger.Info("reconciling stale deployment to current function spec", + "function", fn.Name, "deployment", fsvc.Name, + "deployment_rv", deplRV, "function_rv", fn.ResourceVersion) + return caaf.updateFuncDeployment(ctx, fn) +} + +func (caaf *Container) updateFuncDeployment(ctx context.Context, fn *fv1.Function) error { + // Live entry only — see updateFunction: versioned projections' entries + // share the UID but must keep their pinned spec. + fsvc, err := caaf.fsCache.GetLiveByFunctionUID(fn.UID) + if err != nil { + return fmt.Errorf("error updating function due to unable to find function service cache %s: %w", k8sCache.MetaObjectToName(fn), err) + } + fnObjName := fsvc.Name + + deployLabels := caaf.getDeployLabels(fn.ObjectMeta) + caaf.logger.Info("updating deployment due to function update", + "deployment", fnObjName, "function", fn.Name) + + // to support backward compatibility, if the function was created in default ns, we fall back to creating the + // deployment of the function in fission-function ns + ns := caaf.nsResolver.GetFunctionNS(fn.Namespace) + + existingDepl, err := caaf.kubernetesClient.AppsV1().Deployments(ns).Get(ctx, fnObjName, metav1.GetOptions{}) + if err != nil { + return err + } + + // the resource version inside function packageRef is changed, + // so the content of fetchRequest in deployment cmd is different. + // Therefore, the deployment update will trigger a rolling update. + newDeployment, err := caaf.getDeploymentSpec(ctx, fn, existingDepl.Spec.Replicas, // use current replicas instead of minscale in the ExecutionStrategy. + fnObjName, ns, deployLabels, caaf.getDeployAnnotations(fn.ObjectMeta)) + if err != nil { + caaf.updateStatus(fn, err, "failed to get new deployment spec while updating function") + return err + } + + // A Deployment's selector is immutable, and for container functions it is + // derived from the Function CR's own labels (getDeployLabels copies them), + // which can change without bumping Generation. An in-place Update with a + // different selector is rejected by the API server; returning nil (not an + // error) avoids requeuing forever against a permanently immutable field — + // the same guard newdeploy's updateFuncDeployment carries. + if !apiequality.Semantic.DeepEqual(existingDepl.Spec.Selector, newDeployment.Spec.Selector) { + caaf.logger.Info("deployment selector changed (function labels changed); cannot update in place, leaving existing deployment", + "deployment", fnObjName, "function", fn.Name) + return nil + } + + err = caaf.updateDeployment(ctx, newDeployment, ns) + if err != nil { + caaf.updateStatus(fn, err, "failed to update deployment while updating function") + return err + } + + return nil +} + +func (caaf *Container) fnDelete(ctx context.Context, fn *fv1.Function) error { + var multierr error + + // GetByFunction keys on UID+Generation (see #3596), but the fn passed in + // on delete can carry a Generation the cache entries were never keyed + // under (a delete racing a spec update, or a stale informer snapshot) — + // so enumerate ALL cached entries for the UID instead. With RFC-0025 a + // UID can back several entries at once (the live function plus versioned + // projections, each with its own -v Deployment/Service/HPA), and + // deleting the live Function must tear down every one of them, not just + // whichever entry happened to be cached last. The computed name is always + // included: entries can be missing (never specialized, evicted, executor + // restarted) and bailing out would leak the live objects. + computedName := caaf.getObjName(fn) + objNames := map[string]struct{}{computedName: {}} + fsvcs := caaf.fsCache.ListByFunctionUID(fn.UID) + if len(fsvcs) == 0 { + caaf.logger.V(1).Info("fsvc not in cache, cleaning up by computed name", + "function", k8sCache.MetaObjectToName(fn), "obj_name", computedName) + } + for _, fsvc := range fsvcs { + objNames[fsvc.Name] = struct{}{} + if _, err := caaf.fsCache.DeleteOld(fsvc, time.Second*0); err != nil { + multierr = errors.Join(multierr, fmt.Errorf("error deleting function from cache: %w", err)) + } + } + + // to support backward compatibility, if the function was created in default ns, we fall back to creating the + // deployment of the function in fission-function ns, so cleaning up resources there + ns := caaf.nsResolver.GetFunctionNS(fn.Namespace) + + // Cache-INDEPENDENT enumeration: the cache walk above only finds + // generations with a live fscache entry, and the computed name only covers + // the Function object passed in (typically the live spec, no version + // label). Per-version -v objects specialized by a previous executor + // incarnation — or whose entry was evicted — would leak past both, so + // additionally list Deployments/Services/HPAs by the identity labels every + // created object carries (getDeployLabels) and tear down everything found. + selector := labels.Set{ + fv1.EXECUTOR_TYPE: string(fv1.ExecutorTypeContainer), + fv1.FUNCTION_UID: string(fn.UID), + }.AsSelector().String() + listed, listErr := executorUtils.FunctionObjectNames(ctx, caaf.kubernetesClient, ns, selector) + multierr = errors.Join(multierr, listErr) + for objName := range listed { + objNames[objName] = struct{}{} + } + + for objName := range objNames { + multierr = errors.Join(multierr, caaf.cleanupContainer(ctx, ns, objName)) + } + return multierr +} + +// CleanupFunctionVersion tears down the per-version Deployment/Service/HPA +// set (and the fscache entry) a deleted FunctionVersion's projection left +// behind, driven by the executor's versiongc reconciler. See the newdeploy +// counterpart (and pkg/executor/versiongc's package doc) for why teardown is +// label-based rather than an ownerReference on the objects: with the chart's +// functionNamespace value set the objects and the FunctionVersion CR live in +// different namespaces, which Kubernetes ownerRef GC cannot span. +func (caaf *Container) CleanupFunctionVersion(ctx context.Context, fnNamespace, versionName string) error { + var errs error + + // Cache first: evict the projection's entry so the idle reaper stops + // tracking a service whose backing objects are about to vanish, and so + // its object name is torn down even if a list below fails. + objNames := make(map[string]struct{}) + for _, fsvc := range caaf.fsCache.ListByFunctionVersion(fnNamespace, versionName) { + objNames[fsvc.Name] = struct{}{} + if _, err := caaf.fsCache.DeleteOld(fsvc, time.Second*0); err != nil { + errs = errors.Join(errs, fmt.Errorf("error deleting the function version entry from cache: %w", err)) + } + } + + ns := caaf.nsResolver.GetFunctionNS(fnNamespace) + selector := labels.Set{ + fv1.EXECUTOR_TYPE: string(fv1.ExecutorTypeContainer), + fv1.FUNCTION_NAMESPACE: fnNamespace, + fv1.FUNCTION_VERSION: versionName, + }.AsSelector().String() + listed, listErr := executorUtils.FunctionObjectNames(ctx, caaf.kubernetesClient, ns, selector) + errs = errors.Join(errs, listErr) + for objName := range listed { + objNames[objName] = struct{}{} + } + + for objName := range objNames { + errs = errors.Join(errs, caaf.cleanupContainer(ctx, ns, objName)) + } + + return errs +} + +// getObjName returns a unique name for kubernetes objects of function. A +// versioned Function (fv1.FUNCTION_VERSION label present, RFC-0025) gets a +// name suffixed with the version's bounded "-v" tail so each published +// version's Deployment/Service/HPA is distinct; see +// executorUtils.VersionedObjName for the truncation budget that keeps the +// whole name within the Kubernetes 63-char limit either way. The +// unversioned path is byte-identical to the pre-RFC-0025 behaviour. +func (caaf *Container) getObjName(fn *fv1.Function) string { + return executorUtils.VersionedObjName("container-", fn) +} + +func (caaf *Container) getDeployLabels(fnMeta metav1.ObjectMeta) map[string]string { + deployLabels := maps.CopyStringMap(fnMeta.Labels) + deployLabels[fv1.EXECUTOR_TYPE] = string(fv1.ExecutorTypeContainer) + deployLabels[fv1.FUNCTION_NAME] = fnMeta.Name + deployLabels[fv1.FUNCTION_NAMESPACE] = fnMeta.Namespace + deployLabels[fv1.FUNCTION_UID] = string(fnMeta.UID) + return deployLabels +} + +func (caaf *Container) getDeployAnnotations(fnMeta metav1.ObjectMeta) map[string]string { + deployAnnotations := maps.CopyStringMap(fnMeta.Annotations) + deployAnnotations[fv1.EXECUTOR_INSTANCEID_LABEL] = caaf.instanceID + deployAnnotations[fv1.FUNCTION_RESOURCE_VERSION] = fnMeta.ResourceVersion + return deployAnnotations +} + +// updateStatus is a function which updates status of update. +// Current implementation only logs messages, in future it will update function status +func (caaf *Container) updateStatus(fn *fv1.Function, err error, message string) { + caaf.logger.Error(err, "function status update", "function", fn, "message", message) +} + +// IdleStrategy returns the container idle-reaping strategy (scale the function +// deployment down to MinScale), run by the shared idle reaper. checkEnv is +// false: the container executor never inspected the environment list. +func (caaf *Container) IdleStrategy() idle.Strategy { + return idle.NewScaleDownStrategy(caaf.logger, fv1.ExecutorTypeContainer, caaf.fissionClient, + caaf.fsCache, caaf.kubernetesClient, caaf.defaultIdlePodReapTime, caaf.objectReaperIntervalSecond, false) +} + +func (caaf *Container) DumpDebugInfo(ctx context.Context) error { + return nil +} diff --git a/pkg/executor/executortype/container/isvalid_test.go b/pkg/executor/executortype/container/isvalid_test.go new file mode 100644 index 0000000..2e362cd --- /dev/null +++ b/pkg/executor/executortype/container/isvalid_test.go @@ -0,0 +1,76 @@ +// SPDX-FileCopyrightText: The Fission Authors +// +// SPDX-License-Identifier: Apache-2.0 + +package container + +import ( + "testing" + + "github.com/go-logr/logr" + "github.com/stretchr/testify/assert" + appsv1 "k8s.io/api/apps/v1" + apiv1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + clientgoscheme "k8s.io/client-go/kubernetes/scheme" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + "github.com/fission/fission/pkg/executor/fscache" +) + +func svcObj(ns, name string) *apiv1.Service { + return &apiv1.Service{Namespace: ns, Name: name} +} + +func deplObj(ns, name string, available int32) *appsv1.Deployment { + return &appsv1.Deployment{ + Namespace: ns, Name: name, + Status: appsv1.DeploymentStatus{AvailableReplicas: available}, + } +} + +func fsvcWith(objs ...apiv1.ObjectReference) *fscache.FuncSvc { + return &fscache.FuncSvc{ + Function: &metav1.ObjectMeta{Name: "fn", Namespace: "default"}, + Address: "10.0.0.1", + KubernetesObjects: objs, + } +} + +// TestContainerIsValid covers the IsValid path after it moved off the standalone +// SharedInformerFactory listers onto the executor Manager's cache-backed client. +func TestContainerIsValid(t *testing.T) { + svcRef := apiv1.ObjectReference{Kind: "service", Namespace: "default", Name: "fn-svc"} + deplRef := apiv1.ObjectReference{Kind: "deployment", Namespace: "default", Name: "fn-dep"} + + newContainer := func(objs ...client.Object) *Container { + c := fake.NewClientBuilder().WithScheme(clientgoscheme.Scheme).WithObjects(objs...).Build() + return &Container{logger: logr.Discard(), crClient: c} + } + + t.Run("valid when service and ready deployment both exist", func(t *testing.T) { + c := newContainer(svcObj("default", "fn-svc"), deplObj("default", "fn-dep", 1)) + assert.True(t, c.IsValid(t.Context(), fsvcWith(svcRef, deplRef))) + }) + + t.Run("invalid when the deployment has no available replicas", func(t *testing.T) { + c := newContainer(svcObj("default", "fn-svc"), deplObj("default", "fn-dep", 0)) + assert.False(t, c.IsValid(t.Context(), fsvcWith(svcRef, deplRef))) + }) + + t.Run("invalid when the service is missing from the cache", func(t *testing.T) { + c := newContainer(deplObj("default", "fn-dep", 1)) // no service + assert.False(t, c.IsValid(t.Context(), fsvcWith(svcRef, deplRef))) + }) + + t.Run("invalid when the deployment is missing from the cache", func(t *testing.T) { + c := newContainer(svcObj("default", "fn-svc")) // no deployment + assert.False(t, c.IsValid(t.Context(), fsvcWith(svcRef, deplRef))) + }) + + t.Run("invalid when there are no kubernetes objects", func(t *testing.T) { + c := newContainer() + assert.False(t, c.IsValid(t.Context(), fsvcWith())) + }) +} diff --git a/pkg/executor/executortype/newdeploy/isvalid_test.go b/pkg/executor/executortype/newdeploy/isvalid_test.go new file mode 100644 index 0000000..47c6cd8 --- /dev/null +++ b/pkg/executor/executortype/newdeploy/isvalid_test.go @@ -0,0 +1,116 @@ +// SPDX-FileCopyrightText: The Fission Authors +// +// SPDX-License-Identifier: Apache-2.0 + +package newdeploy + +import ( + "testing" + + "github.com/go-logr/logr" + "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" + clientgoscheme "k8s.io/client-go/kubernetes/scheme" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + fv1 "github.com/fission/fission/pkg/apis/core/v1" + "github.com/fission/fission/pkg/executor/fscache" + "github.com/fission/fission/pkg/utils" +) + +func svcObj(ns, name string) *apiv1.Service { + return &apiv1.Service{Namespace: ns, Name: name} +} + +func deplObj(ns, name string, available int32) *appsv1.Deployment { + return &appsv1.Deployment{ + Namespace: ns, Name: name, + Status: appsv1.DeploymentStatus{AvailableReplicas: available}, + } +} + +func fsvcWith(objs ...apiv1.ObjectReference) *fscache.FuncSvc { + return &fscache.FuncSvc{ + Function: &metav1.ObjectMeta{Name: "fn", Namespace: "default"}, + Address: "10.0.0.1", + KubernetesObjects: objs, + } +} + +// TestNewdeployIsValid covers the IsValid path after it moved off the standalone +// SharedInformerFactory listers onto the executor Manager's cache-backed client. +func TestNewdeployIsValid(t *testing.T) { + svcRef := apiv1.ObjectReference{Kind: "service", Namespace: "default", Name: "fn-svc"} + deplRef := apiv1.ObjectReference{Kind: "deployment", Namespace: "default", Name: "fn-dep"} + + newDeploy := func(objs ...client.Object) *NewDeploy { + c := fake.NewClientBuilder().WithScheme(clientgoscheme.Scheme).WithObjects(objs...).Build() + return &NewDeploy{logger: logr.Discard(), crClient: c} + } + + t.Run("valid when service and ready deployment both exist", func(t *testing.T) { + d := newDeploy(svcObj("default", "fn-svc"), deplObj("default", "fn-dep", 1)) + assert.True(t, d.IsValid(t.Context(), fsvcWith(svcRef, deplRef))) + }) + + t.Run("invalid when the deployment has no available replicas", func(t *testing.T) { + d := newDeploy(svcObj("default", "fn-svc"), deplObj("default", "fn-dep", 0)) + assert.False(t, d.IsValid(t.Context(), fsvcWith(svcRef, deplRef))) + }) + + t.Run("invalid when the service is missing from the cache", func(t *testing.T) { + d := newDeploy(deplObj("default", "fn-dep", 1)) // no service + assert.False(t, d.IsValid(t.Context(), fsvcWith(svcRef, deplRef))) + }) + + t.Run("invalid when the deployment is missing from the cache", func(t *testing.T) { + d := newDeploy(svcObj("default", "fn-svc")) // no deployment + assert.False(t, d.IsValid(t.Context(), fsvcWith(svcRef, deplRef))) + }) + + t.Run("invalid when there are no kubernetes objects", func(t *testing.T) { + d := newDeploy() + assert.False(t, d.IsValid(t.Context(), fsvcWith())) + }) +} + +// TestNewdeployResourcesExist covers the drift-detection read used by the +// level-triggered reconcile: present when both backing objects exist, absent when +// either has drifted away. +func TestNewdeployResourcesExist(t *testing.T) { + fn := fnOfType("fn", fv1.ExecutorTypeNewdeploy) + fn.UID = "83c82da2-81e9-4ebd-867e-f383e65e603f" + + build := func(objs ...client.Object) *NewDeploy { + c := fake.NewClientBuilder().WithScheme(clientgoscheme.Scheme).WithObjects(objs...).Build() + return &NewDeploy{logger: logr.Discard(), crClient: c, nsResolver: &utils.NamespaceResolver{}} + } + // Object name/namespace match what createOrGet* use. + name := build().getObjName(fn) + ns := (&utils.NamespaceResolver{}).GetFunctionNS(fn.Namespace) + + t.Run("present when both Deployment and Service exist", func(t *testing.T) { + d := build(deplObj(ns, name, 1), svcObj(ns, name)) + exist, err := d.resourcesExist(t.Context(), fn) + require.NoError(t, err) + assert.True(t, exist) + }) + + t.Run("absent when the Deployment has drifted away", func(t *testing.T) { + d := build(svcObj(ns, name)) // no deployment + exist, err := d.resourcesExist(t.Context(), fn) + require.NoError(t, err) + assert.False(t, exist) + }) + + t.Run("absent when the Service has drifted away", func(t *testing.T) { + d := build(deplObj(ns, name, 1)) // no service + exist, err := d.resourcesExist(t.Context(), fn) + require.NoError(t, err) + assert.False(t, exist) + }) +} diff --git a/pkg/executor/executortype/newdeploy/newdeploymgr.go b/pkg/executor/executortype/newdeploy/newdeploymgr.go new file mode 100644 index 0000000..15f4d90 --- /dev/null +++ b/pkg/executor/executortype/newdeploy/newdeploymgr.go @@ -0,0 +1,983 @@ +// SPDX-FileCopyrightText: The Fission Authors +// +// SPDX-License-Identifier: Apache-2.0 + +package newdeploy + +import ( + "context" + "errors" + "fmt" + "os" + "reflect" + "strconv" + "strings" + "time" + + appsv1 "k8s.io/api/apps/v1" + apiv1 "k8s.io/api/core/v1" + apiequality "k8s.io/apimachinery/pkg/api/equality" + k8sErrs "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" + k8sTypes "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/kubernetes" + k8sCache "k8s.io/client-go/tools/cache" + "sigs.k8s.io/controller-runtime/pkg/client" + + "github.com/go-logr/logr" + "golang.org/x/sync/errgroup" + + fv1 "github.com/fission/fission/pkg/apis/core/v1" + "github.com/fission/fission/pkg/crd" + ferror "github.com/fission/fission/pkg/error" + "github.com/fission/fission/pkg/executor/executortype" + "github.com/fission/fission/pkg/executor/fscache" + "github.com/fission/fission/pkg/executor/metrics" + "github.com/fission/fission/pkg/executor/reaper" + "github.com/fission/fission/pkg/executor/reaper/idle" + executorUtils "github.com/fission/fission/pkg/executor/util" + hpautils "github.com/fission/fission/pkg/executor/util/hpa" + fetcherConfig "github.com/fission/fission/pkg/fetcher/config" + "github.com/fission/fission/pkg/generated/clientset/versioned" + "github.com/fission/fission/pkg/throttler" + "github.com/fission/fission/pkg/utils" + "github.com/fission/fission/pkg/utils/maps" + otelUtils "github.com/fission/fission/pkg/utils/otel" +) + +var ( + _ executortype.ExecutorType = &NewDeploy{} +) + +type ( + // NewDeploy represents an ExecutorType + NewDeploy struct { + logger logr.Logger + + kubernetesClient kubernetes.Interface + fissionClient versioned.Interface + instanceID string + fetcherConfig *fetcherConfig.Config + nsResolver *utils.NamespaceResolver + + runtimeImagePullPolicy apiv1.PullPolicy + useIstio bool + + fsCache *fscache.FunctionServiceCache // cache funcSvc's by function, address and pod name + + throttler *throttler.Throttler + + defaultIdlePodReapTime time.Duration + + // crClient is the executor Manager's cache-backed client, used by IsValid + // to read function Deployments/Services from the shared Manager cache + // (replacing this type's standalone SharedInformerFactory listers). Set in + // RegisterReconcilers once the Manager exists. + crClient client.Client + + hpaops *hpautils.HpaOperations + + podSpecPatch *apiv1.PodSpec + objectReaperIntervalSecond time.Duration + + enableOwnerReferences bool + + // imageVolumeOK is the once-evaluated RFC-0001 Path B gate: + // ENABLE_OCI_IMAGE_VOLUME opted in AND the cluster supports + // KEP-4639 image volumes (>= 1.33). + imageVolumeOK bool + } +) + +// MakeNewDeploy initializes and returns an instance of NewDeploy. +func MakeNewDeploy( + ctx context.Context, + logger logr.Logger, + fissionClient versioned.Interface, + kubernetesClient kubernetes.Interface, + fetcherConfig *fetcherConfig.Config, + instanceID string, + podSpecPatch *apiv1.PodSpec, +) (executortype.ExecutorType, error) { + enableIstio := false + if len(os.Getenv("ENABLE_ISTIO")) > 0 { + istio, err := strconv.ParseBool(os.Getenv("ENABLE_ISTIO")) + if err != nil { + logger.Error(err, "failed to parse 'ENABLE_ISTIO', set to false") + } + enableIstio = istio + } + + // RFC-0001 Path B gate, evaluated once (shared helper, so poolmgr and + // newdeploy cannot drift). + imageVolumeOK := executorUtils.ImageVolumeGate(logger, kubernetesClient.Discovery()) + + nd := &NewDeploy{ + logger: logger.WithName("new_deploy"), + + imageVolumeOK: imageVolumeOK, + + fissionClient: fissionClient, + kubernetesClient: kubernetesClient, + instanceID: instanceID, + fsCache: fscache.MakeFunctionServiceCache(logger), + throttler: throttler.MakeThrottler(1 * time.Minute), + nsResolver: utils.DefaultNSResolver(), + + fetcherConfig: fetcherConfig, + runtimeImagePullPolicy: utils.GetImagePullPolicy(os.Getenv("RUNTIME_IMAGE_PULL_POLICY")), + useIstio: enableIstio, + + defaultIdlePodReapTime: 2 * time.Minute, + objectReaperIntervalSecond: time.Duration(executorUtils.GetObjectReaperInterval(logger, fv1.ExecutorTypeNewdeploy, 5)) * time.Second, + hpaops: hpautils.NewHpaOperations(logger, kubernetesClient, instanceID), + + podSpecPatch: podSpecPatch, + + enableOwnerReferences: utils.IsOwnerReferencesEnabled(), + } + + // The Function and Environment watches are controller-runtime reconcilers now + // (see reconciler.go / RegisterReconcilers), wired on the executor Manager. + // Deployment/Service reads (IsValid) go through the Manager's cache-backed + // client (nd.crClient), set in RegisterReconcilers — no per-type informer + // factory is needed. + return nd, nil +} + +// Run start the function and environment controller along with an object reaper. +// Run is a no-op: the newdeploy manager no longer runs its own informer +// factory. Its Deployment/Service reads (IsValid) go through the executor +// Manager's cache, which controller-runtime syncs before any runnable (including +// this type's reapers) starts. +func (deploy *NewDeploy) Run(context.Context, *errgroup.Group) {} + +// GetTypeName returns the executor type name. +func (deploy *NewDeploy) GetTypeName(ctx context.Context) fv1.ExecutorType { + return fv1.ExecutorTypeNewdeploy +} + +// GetFuncSvc returns a function service; error otherwise. +func (deploy *NewDeploy) GetFuncSvc(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) { + return deploy.createFunction(ctx, fn) +} + +// GetFuncSvcFromCache returns a function service from cache; error otherwise. +// The lookup is keyed (UID, Generation), never bare UID: fn may be an +// RFC-0025 versioned projection pinning an older Generation of a UID whose +// live entry is also cached, and a UID-only lookup would return whichever +// generation's fsvc was cached last — silently serving the wrong version's +// code (IsValid only checks the backing objects exist). +func (deploy *NewDeploy) GetFuncSvcFromCache(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) { + otelUtils.SpanTrackEvent(ctx, "GetFuncSvcFromCache") + return deploy.fsCache.GetByFunction(&fn.ObjectMeta) +} + +// DeleteFuncSvcFromCache deletes a function service from cache. +func (deploy *NewDeploy) DeleteFuncSvcFromCache(ctx context.Context, fsvc *fscache.FuncSvc) { + otelUtils.SpanTrackEvent(ctx, "DeleteFuncSvcFromCache") + deploy.fsCache.DeleteEntry(fsvc) +} + +// UnTapService has not been implemented for NewDeployment. +func (deploy *NewDeploy) UnTapService(ctx context.Context, fnMeta *metav1.ObjectMeta, svcHost string) { + // Not Implemented for NewDeployment. Will be used when support of concurrent specialization of same function is added. +} + +// MarkSpecializationFailure has not been implemented for NewDeployment. +func (deploy *NewDeploy) MarkSpecializationFailure(ctx context.Context, fnMeta *metav1.ObjectMeta) { + // Not Implemented for NewDeployment. Will be used when support of concurrent specialization of same function is added. +} + +// TapService makes a TouchByAddress request to the cache. +func (deploy *NewDeploy) TapService(ctx context.Context, svcHost string) error { + otelUtils.SpanTrackEvent(ctx, "TapService") + err := deploy.fsCache.TouchByAddress(svcHost) + if err != nil { + return err + } + return nil +} + +// IsValid does a get on the service address to ensure it's a valid service, then +// scale deployment to 1 replica if there are no available replicas for function. +// Return true if no error occurs, return false otherwise. +func (deploy *NewDeploy) IsValid(ctx context.Context, fsvc *fscache.FuncSvc) bool { + logger := otelUtils.LoggerWithTraceID(ctx, deploy.logger) + otelUtils.SpanTrackEvent(ctx, "IsValid", fscache.GetAttributesForFuncSvc(fsvc)...) + if len(strings.Split(fsvc.Address, ".")) == 0 { + logger.Error(nil, "address not found in function service") + return false + } + if len(fsvc.KubernetesObjects) == 0 { + logger.Info("no kubernetes object related to function", "function", fsvc.Function.Name) + return false + } + for _, obj := range fsvc.KubernetesObjects { + objKey := client.ObjectKey{Namespace: obj.Namespace, Name: obj.Name} + if strings.ToLower(obj.Kind) == "service" { + if err := deploy.crClient.Get(ctx, objKey, &apiv1.Service{}); err != nil { + if !k8sErrs.IsNotFound(err) { + logger.Error(err, "error validating function service", "function", fsvc.Function.Name) + } + return false + } + + } else if strings.ToLower(obj.Kind) == "deployment" { + currentDeploy := &appsv1.Deployment{} + if err := deploy.crClient.Get(ctx, objKey, currentDeploy); err != nil { + if !k8sErrs.IsNotFound(err) { + logger.Error(err, "error validating function deployment", "function", fsvc.Function.Name) + } + return false + } + if currentDeploy.Status.AvailableReplicas < 1 { + return false + } + } + } + return true +} + +// RefreshFuncPods deletes pods related to the function so that new pods are replenished +func (deploy *NewDeploy) RefreshFuncPods(ctx context.Context, logger logr.Logger, f fv1.Function) error { + // Defence in depth for GHSA-cvw6-gfvv-953q — see fnCreate for context. + if envNs := f.Spec.Environment.Namespace; envNs != "" && envNs != f.Namespace { + return fmt.Errorf("cross-namespace environment reference is not allowed: fn.namespace=%s env.namespace=%s", + f.Namespace, envNs) + } + + env, err := deploy.fissionClient.CoreV1().Environments(f.Spec.Environment.Namespace).Get(ctx, f.Spec.Environment.Name, metav1.GetOptions{}) + if err != nil { + return err + } + + funcLabels := deploy.getDeployLabels(f.ObjectMeta, metav1.ObjectMeta{ + Name: f.Spec.Environment.Name, + Namespace: f.Spec.Environment.Namespace, + UID: env.UID, + }) + + dep, err := deploy.kubernetesClient.AppsV1().Deployments(deploy.nsResolver.GetFunctionNS(f.ObjectMeta.Namespace)).List(ctx, metav1.ListOptions{ + LabelSelector: labels.Set(funcLabels).AsSelector().String(), + }) + + if err != nil { + return err + } + + // Ideally there should be only one deployment but for now we rely on label/selector to ensure that condition + for _, deployment := range dep.Items { + rvCount, err := executorUtils.ReferencedResourcesRVSum(ctx, deploy.kubernetesClient, f.Namespace, f.Spec) + if err != nil { + return err + } + + patch := fmt.Sprintf(`{"spec" : {"template": {"spec":{"containers":[{"name": "%s", "image": "%s", "env":[{"name": "%s", "value": "%d"}]}]}}}}`, + env.Name, env.Spec.Runtime.Image, fv1.ResourceVersionCount, rvCount) + + _, err = deploy.kubernetesClient.AppsV1().Deployments(deployment.ObjectMeta.Namespace).Patch(ctx, deployment.Name, + k8sTypes.StrategicMergePatchType, + []byte(patch), metav1.PatchOptions{}) + if err != nil { + return err + } + } + return nil +} + +// AdoptExistingResources re-claims the function objects this executor type left +// behind on a previous run. It routes through the throttled createFunction (not +// fnCreate directly), so the adopt pass and the Function reconciler — which also +// re-stamps existing objects on its initial sync — single-flight per function +// UID instead of racing on the in-place Update (which previously produced +// resourceVersion conflicts that tripped fnCreate's destroy-on-error path). +func (deploy *NewDeploy) AdoptExistingResources(ctx context.Context) { + executorUtils.AdoptFunctions(ctx, deploy.logger, deploy.fissionClient, fv1.ExecutorTypeNewdeploy, + func(ctx context.Context, fn *fv1.Function) error { + _, err := deploy.createFunction(ctx, fn) + return err + }) +} + +// CleanupOldExecutorObjects cleans orphaned resources. +func (deploy *NewDeploy) CleanupOldExecutorObjects(ctx context.Context) { + reaper.CleanupExecutorObjects(ctx, deploy.logger, deploy.kubernetesClient, deploy.instanceID, fv1.ExecutorTypeNewdeploy) +} + +func (deploy *NewDeploy) getEnvFunctions(ctx context.Context, m *metav1.ObjectMeta) []fv1.Function { + funcList, err := deploy.fissionClient.CoreV1().Functions(m.Namespace).List(ctx, metav1.ListOptions{}) + if err != nil { + deploy.logger.Error(err, "Error getting functions for env", "environment", m) + } + relatedFunctions := make([]fv1.Function, 0) + for _, f := range funcList.Items { + if (f.Spec.Environment.Name == m.Name) && (f.Spec.Environment.Namespace == m.Namespace) && f.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fv1.ExecutorTypeNewdeploy { + relatedFunctions = append(relatedFunctions, f) + } + } + return relatedFunctions +} + +// updateEnvFunctions rebuilds the deployments of all newdeploy functions of the +// given environment after its runtime image changed. It drives the Environment +// reconciler (replacing the EnvEventHandlers UpdateFunc body). A per-function +// failure is logged and skipped rather than failing the whole environment sweep: +// requeuing the environment would re-roll every (including already-updated) +// function, which is the amplification the informer handler avoided. +func (deploy *NewDeploy) updateEnvFunctions(ctx context.Context, env *fv1.Environment) error { + deploy.logger.V(1).Info("updating functions of environment with changed image", "environment", env.ObjectMeta) + for _, f := range deploy.getEnvFunctions(ctx, &env.ObjectMeta) { + function, err := deploy.fissionClient.CoreV1().Functions(f.Namespace).Get(ctx, f.Name, metav1.GetOptions{}) + if err != nil { + deploy.logger.Error(err, "error getting function while updating environment functions", "function", f.ObjectMeta) + continue + } + if err := deploy.updateFuncDeployment(ctx, function, env); err != nil { + deploy.logger.Error(err, "error updating function deployment after environment image change", "function", function.ObjectMeta) + continue + } + } + return nil +} + +// resourcesExist reports whether the function's backing Deployment and Service +// are present in the Manager cache. A missing object means it drifted away +// (deleted out-of-band) and the function should be recreated. Reads go through +// the cache-backed crClient (same path as IsValid). +func (deploy *NewDeploy) resourcesExist(ctx context.Context, fn *fv1.Function) (bool, error) { + ns := deploy.nsResolver.GetFunctionNS(fn.Namespace) + key := client.ObjectKey{Namespace: ns, Name: deploy.getObjName(fn)} + for _, obj := range []client.Object{&appsv1.Deployment{}, &apiv1.Service{}} { + if err := deploy.crClient.Get(ctx, key, obj); err != nil { + if k8sErrs.IsNotFound(err) { + return false, nil + } + return false, err + } + } + return true, nil +} + +func (deploy *NewDeploy) createFunction(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) { + if fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fv1.ExecutorTypeNewdeploy { + return nil, nil + } + + logger := otelUtils.LoggerWithTraceID(ctx, deploy.logger) + + // Single-flight per (UID, Generation), not per UID: RFC-0025 versioned + // projections share the live function's UID but create distinct + // per-version Deployments, so their creates must neither serialize + // behind each other nor have a loser handed another generation's fsvc + // from the cache fallback. + fsvc, err := deploy.throttler.RunOnce(crd.CacheKeyUGFromMeta(&fn.ObjectMeta).String(), func(ableToCreate bool) (*fscache.FuncSvc, error) { + if ableToCreate { + return deploy.fnCreate(ctx, fn) + } + return deploy.fsCache.GetByFunction(&fn.ObjectMeta) + }) + + if err != nil { + e := "error creating k8s resources for function" + logger.Error(err, e, "function_name", fn.Name, + "function_namespace", fn.Namespace) + return nil, fmt.Errorf("error creating k8s resources for function %s: %w", k8sCache.MetaObjectToName(fn), err) + } + otelUtils.SpanTrackEvent(ctx, "fnSvcResponse", fscache.GetAttributesForFuncSvc(fsvc)...) + + return fsvc, nil +} + +func (deploy *NewDeploy) deleteFunction(ctx context.Context, fn *fv1.Function) error { + if fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fv1.ExecutorTypeNewdeploy { + return nil + } + err := deploy.fnDelete(ctx, fn) + if err != nil { + return fmt.Errorf("error deleting kubernetes objects of function %s: %w", k8sCache.MetaObjectToName(fn), err) + } + return nil +} + +// destroyOnCreateError reports whether a fnCreate failure warrants tearing down +// the function's partial resources. A Conflict or AlreadyExists means the object +// exists and was concurrently modified (e.g. the adopt pass racing a reconcile, +// or two reconciles), so deleting it would turn a transient blip into a cold +// recreate — leave it for the next reconcile to converge instead. Genuine +// creation failures (quota, invalid spec, API errors) still trigger cleanup so a +// brand-new function doesn't leak half-created objects. +func destroyOnCreateError(err error) bool { + // An explicitly cancelled context means the executor is shutting down, lost + // leadership, or the caller gave up — not that creation genuinely failed — + // so leave any partially-created resources for the next leader/request to + // adopt instead of tearing them down. A context *deadline* is different: on + // the specialization path the context carries the per-function + // SpecializationTimeout (see pkg/executor/executor.go), so DeadlineExceeded + // is a genuine timeout and falls through to normal cleanup. + if errors.Is(err, context.Canceled) { + return false + } + return !k8sErrs.IsConflict(err) && !k8sErrs.IsAlreadyExists(err) +} + +func (deploy *NewDeploy) fnCreate(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) { + // Defence in depth for GHSA-cvw6-gfvv-953q — primary defence is the + // admission webhook in pkg/webhook/function.go, but a stale Function + // from a pre-webhook upgrade window (or failurePolicy=ignore) could + // still reach this path. + if envNs := fn.Spec.Environment.Namespace; envNs != "" && envNs != fn.Namespace { + return nil, fmt.Errorf("cross-namespace environment reference is not allowed: fn.namespace=%s env.namespace=%s", + fn.Namespace, envNs) + } + + // Authoritative re-read: the Function object reaching this path originates + // from the router's request body and can be stale — its DeletionTimestamp + // may be absent even though the Function is being deleted. Without this + // check a create can race the delete teardown (funcreconciler) and + // re-create the Deployment/Service/HPA *after* teardown removed them, + // leaking objects whose owning Function CR is already gone. + live, err := deploy.fissionClient.CoreV1().Functions(fn.Namespace).Get(ctx, fn.Name, metav1.GetOptions{}) + if err != nil { + if k8sErrs.IsNotFound(err) { + return nil, ferror.MakeError(ferror.ErrorNotFound, + fmt.Sprintf("function %s is gone, not creating service", k8sCache.MetaObjectToName(fn))) + } + return nil, err + } + if live.UID != fn.UID || !live.DeletionTimestamp.IsZero() { + return nil, ferror.MakeError(ferror.ErrorNotFound, + fmt.Sprintf("function %s is being deleted, not creating service", k8sCache.MetaObjectToName(fn))) + } + + cleanupFunc := func(ctx context.Context, ns string, name string) { + err := deploy.cleanupNewdeploy(ctx, ns, name) + if err != nil { + deploy.logger.Error(err, "received error while cleaning function resources", + "namespace", ns, "name", name) + } + } + env, err := deploy.fissionClient.CoreV1(). + Environments(fn.Spec.Environment.Namespace). + Get(ctx, fn.Spec.Environment.Name, metav1.GetOptions{}) + if err != nil { + return nil, err + } + + objName := deploy.getObjName(fn) + deployLabels := deploy.getDeployLabels(fn.ObjectMeta, env.ObjectMeta) + deployAnnotations := deploy.getDeployAnnotations(fn.ObjectMeta, env.ObjectMeta) + + // to support backward compatibility, if the function was created in default ns, we fall back to creating the + // deployment of the function in fission-function ns + ns := deploy.nsResolver.GetFunctionNS(fn.Namespace) + + // Envoy(istio-proxy) returns 404 directly before istio pilot + // propagates latest Envoy-specific configuration. + // Since newdeploy waits for pods of deployment to be ready, + // change the order of kubeObject creation (create service first, + // then deployment) to take advantage of waiting time. + // Transient executor errors are not written to Function.Status; see + // the analogous note in pkg/executor/executortype/poolmgr/gp.go. + svc, err := deploy.createOrGetSvc(ctx, fn, deployLabels, deployAnnotations, objName, ns) + if err != nil { + deploy.logger.Error(err, "error creating service", "service", objName) + if destroyOnCreateError(err) { + go cleanupFunc(context.Background(), ns, objName) + } + return nil, fmt.Errorf("error creating service %s: %w", objName, err) + } + // Absolute FQDN, deliberately: the router dials this on every proxied + // request whose connection isn't already warm, and the short name's + // search-path walk is what melts down in-cluster DNS under load. + svcAddress := utils.ServiceFQDN(svc.Name, svc.Namespace) + + depl, err := deploy.createOrGetDeployment(ctx, fn, env, objName, deployLabels, deployAnnotations, ns) + if err != nil { + deploy.logger.Error(err, "error creating deployment", "deployment", objName) + if destroyOnCreateError(err) { + go cleanupFunc(context.Background(), ns, objName) + } + return nil, fmt.Errorf("error creating deployment %s: %w", objName, err) + } + + // env was already validated by createOrGetDeployment above, so the + // container name resolves without error here. + mainContainerName, err := deploy.mainContainerName(env) + if err != nil { + if destroyOnCreateError(err) { + go cleanupFunc(context.Background(), ns, objName) + } + return nil, fmt.Errorf("error resolving main container for HPA %s: %w", objName, err) + } + + hpa, err := deploy.hpaops.CreateOrGetHpa(ctx, fn, objName, &fn.Spec.InvokeStrategy.ExecutionStrategy, mainContainerName, depl, deployLabels, deployAnnotations) + if err != nil { + deploy.logger.Error(err, "error creating HPA", "hpa", objName) + if destroyOnCreateError(err) { + go cleanupFunc(context.Background(), ns, objName) + } + return nil, fmt.Errorf("error creating HPA %s: %w", objName, err) + } + + kubeObjRefs := []apiv1.ObjectReference{ + { + // obj.TypeMeta.Kind does not work hence this, needs investigation and a fix + Kind: "deployment", + Name: depl.Name, + APIVersion: depl.APIVersion, + Namespace: depl.Namespace, + ResourceVersion: depl.ResourceVersion, + UID: depl.UID, + }, + { + Kind: "service", + Name: svc.Name, + APIVersion: svc.APIVersion, + Namespace: svc.Namespace, + ResourceVersion: svc.ResourceVersion, + UID: svc.UID, + }, + { + Kind: "horizontalpodautoscaler", + Name: hpa.Name, + APIVersion: hpa.APIVersion, + Namespace: hpa.Namespace, + ResourceVersion: hpa.ResourceVersion, + UID: hpa.UID, + }, + } + + fsvc := &fscache.FuncSvc{ + Name: objName, + Function: &fn.ObjectMeta, + Environment: env, + Address: svcAddress, + KubernetesObjects: kubeObjRefs, + Executor: fv1.ExecutorTypeNewdeploy, + } + + _, err = deploy.fsCache.Add(*fsvc) + if err != nil { + deploy.logger.Error(err, "error adding function to cache", "function", fsvc.Function) + metrics.RecordColdStartError(ctx, fn.Name, fn.Namespace) + return fsvc, err + } + + metrics.RecordColdStart(ctx, fn.Name, fn.Namespace) + executorUtils.SetFunctionReady(ctx, deploy.logger, deploy.fissionClient, fn, fv1.FunctionReasonReady, "newdeploy deployment is ready") + + return fsvc, nil +} + +func (deploy *NewDeploy) updateFunction(ctx context.Context, oldFn *fv1.Function, newFn *fv1.Function) error { + + if oldFn.ResourceVersion == newFn.ResourceVersion { + return nil + } + + // Ignoring updates to functions which are not of NewDeployment type + if newFn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fv1.ExecutorTypeNewdeploy && + oldFn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fv1.ExecutorTypeNewdeploy { + return nil + } + + // Executor type is no longer New Deployment + if newFn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fv1.ExecutorTypeNewdeploy && + oldFn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fv1.ExecutorTypeNewdeploy { + deploy.logger.Info("function does not use new deployment executor anymore, deleting resources", + "function", newFn) + // IMP - pass the oldFn, as the new/modified function is not in cache + return deploy.deleteFunction(ctx, oldFn) + } + + // Executor type changed to New Deployment from something else + if oldFn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fv1.ExecutorTypeNewdeploy && + newFn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fv1.ExecutorTypeNewdeploy { + deploy.logger.Info("function type changed to new deployment, creating resources", + "old_function", oldFn.ObjectMeta, + "new_function", newFn.ObjectMeta) + _, err := deploy.createFunction(ctx, newFn) + if err != nil { + deploy.updateStatus(oldFn, err, "error changing the function's type to newdeploy") + } + return err + } + + deployChanged := false + + if !reflect.DeepEqual(oldFn.Spec.InvokeStrategy, newFn.Spec.InvokeStrategy) { + + // to support backward compatibility, if the function was created in default ns, we fall back to creating the + // deployment of the function in fission-function ns, so cleaning up resources there + ns := deploy.nsResolver.GetFunctionNS(newFn.Namespace) + + // Live entry only: updates arrive for the live Function CR, and its + // versioned projections' entries (which share the UID) are immutable + // snapshots whose HPA must not be touched here. + fsvc, err := deploy.fsCache.GetLiveByFunctionUID(newFn.UID) + if err != nil { + return fmt.Errorf("error updating function due to unable to find function service cache %s: %w", k8sCache.MetaObjectToName(oldFn), err) + } + + hpa, err := deploy.hpaops.GetHpa(ctx, ns, fsvc.Name) + if err != nil { + deploy.updateStatus(oldFn, err, "error getting HPA while updating function") + return err + } + + hpaChanged := false + + if newFn.Spec.InvokeStrategy.ExecutionStrategy.MinScale != oldFn.Spec.InvokeStrategy.ExecutionStrategy.MinScale { + replicas := int32(newFn.Spec.InvokeStrategy.ExecutionStrategy.MinScale) + hpa.Spec.MinReplicas = &replicas + hpaChanged = true + } + + if newFn.Spec.InvokeStrategy.ExecutionStrategy.MaxScale != oldFn.Spec.InvokeStrategy.ExecutionStrategy.MaxScale { + hpa.Spec.MaxReplicas = int32(newFn.Spec.InvokeStrategy.ExecutionStrategy.MaxScale) + hpaChanged = true + } + + if !reflect.DeepEqual(newFn.Spec.InvokeStrategy.ExecutionStrategy.Metrics, oldFn.Spec.InvokeStrategy.ExecutionStrategy.Metrics) { + // The CLI emits pod-wide Resource metrics; normalize them to + // ContainerResource metrics scoped to the function container, the + // same as createFunction does via CreateOrGetHpa. + env, err := deploy.fissionClient.CoreV1().Environments(newFn.Spec.Environment.Namespace). + Get(ctx, newFn.Spec.Environment.Name, metav1.GetOptions{}) + if err != nil { + deploy.updateStatus(oldFn, err, "failed to get environment while updating HPA metrics") + return err + } + mainContainerName, err := deploy.mainContainerName(env) + if err != nil { + deploy.updateStatus(oldFn, err, "failed to resolve main container while updating HPA metrics") + return err + } + hpa.Spec.Metrics = hpautils.RewriteResourceMetricsToContainer( + newFn.Spec.InvokeStrategy.ExecutionStrategy.Metrics, mainContainerName, deploy.logger) + hpaChanged = true + } + + if !reflect.DeepEqual(newFn.Spec.InvokeStrategy.ExecutionStrategy.Behavior, oldFn.Spec.InvokeStrategy.ExecutionStrategy.Behavior) { + hpa.Spec.Behavior = newFn.Spec.InvokeStrategy.ExecutionStrategy.Behavior + hpaChanged = true + } + + if hpaChanged { + err := deploy.hpaops.UpdateHpa(ctx, hpa) + if err != nil { + deploy.updateStatus(oldFn, err, "error updating HPA while updating function") + return err + } + } + } + + if podTemplateChanged(oldFn, newFn) { + deploy.logger.V(1).Info("deployment changed", "msg", "deployment changed") + deployChanged = true + } + + if deployChanged { + env, err := deploy.fissionClient.CoreV1().Environments(newFn.Spec.Environment.Namespace). + Get(ctx, newFn.Spec.Environment.Name, metav1.GetOptions{}) + if err != nil { + deploy.updateStatus(oldFn, err, "failed to get environment while updating function") + return err + } + return deploy.updateFuncDeployment(ctx, newFn, env) + } + + return nil +} + +// reconcileDeploymentSpec brings an already-existing deployment up to the +// function's current spec when it lags — the level-trigger half of every +// create-routed reconcile (see createAndConverge in reconciler.go for why it +// must follow createFunction). The deployment carries the function's +// ResourceVersion as a metadata annotation (getDeployAnnotations), so compare +// it: if stale, push the current spec. A no-op when already current. +func (deploy *NewDeploy) reconcileDeploymentSpec(ctx context.Context, fn *fv1.Function) error { + // Live entry only: fn is the live Function CR here, and versioned + // projections sharing its UID pin immutable snapshots that must not be + // rewritten to the live spec. + fsvc, err := deploy.fsCache.GetLiveByFunctionUID(fn.UID) + if err != nil { + // Not specialized yet — no deployment to reconcile; the on-demand path will + // create it from the current spec when the function is first invoked. + return nil + } + ns := deploy.nsResolver.GetFunctionNS(fn.Namespace) + existingDepl, err := deploy.kubernetesClient.AppsV1().Deployments(ns).Get(ctx, fsvc.Name, metav1.GetOptions{}) + if err != nil { + if k8sErrs.IsNotFound(err) { + return nil + } + return err + } + deplRV := existingDepl.Annotations[fv1.FUNCTION_RESOURCE_VERSION] + if deplRV == fn.ResourceVersion { + return nil // deployment already reflects the current function spec + } + env, err := deploy.fissionClient.CoreV1().Environments(fn.Spec.Environment.Namespace). + Get(ctx, fn.Spec.Environment.Name, metav1.GetOptions{}) + if err != nil { + return err + } + deploy.logger.Info("reconciling stale deployment to current function spec", + "function", fn.Name, "deployment", fsvc.Name, + "deployment_rv", deplRV, "function_rv", fn.ResourceVersion) + return deploy.updateFuncDeployment(ctx, fn, env) +} + +func (deploy *NewDeploy) updateFuncDeployment(ctx context.Context, fn *fv1.Function, env *fv1.Environment) error { + // Live entry only — see updateFunction: versioned projections' entries + // share the UID but must keep their pinned spec. + fsvc, err := deploy.fsCache.GetLiveByFunctionUID(fn.UID) + if err != nil { + return fmt.Errorf("error updating function due to unable to find function service cache: %s: %w", k8sCache.MetaObjectToName(fn), err) + } + fnObjName := fsvc.Name + + deployLabels := deploy.getDeployLabels(fn.ObjectMeta, env.ObjectMeta) + deploy.logger.Info("updating deployment due to function/environment update", + "deployment", fnObjName, "function", fn.Name) + + // to support backward compatibility, if the function was created in default ns, we fall back to creating the + // deployment of the function in fission-function ns + ns := deploy.nsResolver.GetFunctionNS(fn.Namespace) + + deployAnnotations := deploy.getDeployAnnotations(fn.ObjectMeta, env.ObjectMeta) + + existingDepl, err := deploy.kubernetesClient.AppsV1().Deployments(ns).Get(ctx, fnObjName, metav1.GetOptions{}) + if err != nil { + if k8sErrs.IsNotFound(err) { + // The deployment is gone (e.g. raced with a delete). There is nothing to + // update in place; the next on-demand specialization recreates it. Return + // nil rather than an error, which would requeue forever against a missing + // object. This matches the old informer handler, which logged the Get + // error and returned. + deploy.logger.Info("deployment not found while updating function; skipping in-place update", + "deployment", fnObjName, "function", fn.Name) + return nil + } + return err + } + + // the resource version inside function packageRef is changed, + // so the content of fetchRequest in deployment cmd is different. + // Therefore, the deployment update will trigger a rolling update. + newDeployment, err := deploy.getDeploymentSpec(ctx, fn, env, + existingDepl.Spec.Replicas, // use current replicas instead of minscale in the ExecutionStrategy. + fnObjName, ns, deployLabels, deployAnnotations) + if err != nil { + deploy.updateStatus(fn, err, "failed to get new deployment spec while updating function") + return err + } + + // A Deployment's selector is immutable. It is stable across a function's + // code/secret/HPA updates, but changes when the function's environment + // reference changes (the environment UID is part of the selector labels). An + // in-place Update with a different selector is rejected by the API server, so + // skip the rebuild and leave the existing pods serving — matching the old + // informer handler, which logged the rejected Update and moved on. Returning + // nil (not an error) avoids requeuing forever against a permanently immutable + // field. + if !apiequality.Semantic.DeepEqual(existingDepl.Spec.Selector, newDeployment.Spec.Selector) { + deploy.logger.Info("deployment selector changed (e.g. environment reference changed); cannot update in place, leaving existing deployment", + "deployment", fnObjName, "function", fn.Name) + return nil + } + + err = deploy.updateDeployment(ctx, newDeployment, ns) + if err != nil { + deploy.updateStatus(fn, err, "failed to update deployment while updating function") + return err + } + + return nil +} + +func (deploy *NewDeploy) fnDelete(ctx context.Context, fn *fv1.Function) error { + var errs error + + // GetByFunction keys on UID+Generation (see #3596), but the fn passed in + // on delete can carry a Generation the cache entries were never keyed + // under (a delete racing a spec update, or a stale informer snapshot) — + // so enumerate ALL cached entries for the UID instead. With RFC-0025 a + // UID can back several entries at once (the live function plus versioned + // projections, each with its own -v Deployment/Service/HPA), and + // deleting the live Function must tear down every one of them, not just + // whichever entry happened to be cached last. The computed name is always + // included: entries can be missing (never specialized, evicted, executor + // restarted) and bailing out would leak the live objects. + computedName := deploy.getObjName(fn) + objNames := map[string]struct{}{computedName: {}} + fsvcs := deploy.fsCache.ListByFunctionUID(fn.UID) + if len(fsvcs) == 0 { + deploy.logger.V(1).Info("fsvc not in cache, cleaning up by computed name", + "function", k8sCache.MetaObjectToName(fn), "obj_name", computedName) + } + for _, fsvc := range fsvcs { + objNames[fsvc.Name] = struct{}{} + if _, err := deploy.fsCache.DeleteOld(fsvc, time.Second*0); err != nil { + errs = errors.Join(errs, fmt.Errorf("error deleting the function from cache")) + } + } + + // to support backward compatibility, if the function was created in default ns, we fall back to creating the + // deployment of the function in fission-function ns, so cleaning up resources there + ns := deploy.nsResolver.GetFunctionNS(fn.Namespace) + + // Cache-INDEPENDENT enumeration: the cache walk above only finds + // generations with a live fscache entry, and the computed name only covers + // the Function object passed in (typically the live spec, no version + // label). Per-version -v objects specialized by a previous executor + // incarnation — or whose entry was evicted — would leak past both, so + // additionally list Deployments/Services/HPAs by the identity labels every + // created object carries (getDeployLabels) and tear down everything found. + selector := labels.Set{ + fv1.EXECUTOR_TYPE: string(fv1.ExecutorTypeNewdeploy), + fv1.FUNCTION_UID: string(fn.UID), + }.AsSelector().String() + listed, listErr := executorUtils.FunctionObjectNames(ctx, deploy.kubernetesClient, ns, selector) + errs = errors.Join(errs, listErr) + for objName := range listed { + objNames[objName] = struct{}{} + } + + for objName := range objNames { + errs = errors.Join(errs, deploy.cleanupNewdeploy(ctx, ns, objName)) + } + + return errs +} + +// CleanupFunctionVersion tears down the per-version Deployment/Service/HPA +// set (and the fscache entry) a deleted FunctionVersion's projection left +// behind, driven by the executor's versiongc reconciler. Without it those +// objects leak until the whole Function is deleted: retention GC (or a manual +// delete passing the webhook guard) removes only the FunctionVersion CR. +// +// Label-based teardown is used INSTEAD of an ownerReference on the objects +// pointing at the FunctionVersion, for two reasons (see also +// pkg/executor/versiongc's package doc): +// - Kubernetes ownerRefs require owner and dependent in the same namespace, +// but with the chart's functionNamespace value set, a default-namespace +// function's objects deploy to that namespace (nsResolver.GetFunctionNS) +// while its FunctionVersion stays in the function's namespace — the GC +// would treat the cross-namespace owner as absent and delete the +// per-version Deployment right after creation. +// - These objects already carry a controller ownerRef to the Function when +// enableOwnerReferences is on; a FunctionVersion ownerRef ADDED to it +// would never cascade (the GC keeps a dependent while ANY owner is +// alive), and REPLACING it changes teardown semantics for the flag-off / +// cross-namespace installs the flag exists for. +// +// fnNamespace is the function CR's namespace — also the FunctionVersion CR's +// namespace and the FUNCTION_NAMESPACE label value on the objects. +func (deploy *NewDeploy) CleanupFunctionVersion(ctx context.Context, fnNamespace, versionName string) error { + var errs error + + // Cache first: evict the projection's entry so the idle reaper stops + // tracking a service whose backing objects are about to vanish, and so + // its object name is torn down even if a list below fails. + objNames := make(map[string]struct{}) + for _, fsvc := range deploy.fsCache.ListByFunctionVersion(fnNamespace, versionName) { + objNames[fsvc.Name] = struct{}{} + if _, err := deploy.fsCache.DeleteOld(fsvc, time.Second*0); err != nil { + errs = errors.Join(errs, fmt.Errorf("error deleting the function version entry from cache: %w", err)) + } + } + + ns := deploy.nsResolver.GetFunctionNS(fnNamespace) + selector := labels.Set{ + fv1.EXECUTOR_TYPE: string(fv1.ExecutorTypeNewdeploy), + fv1.FUNCTION_NAMESPACE: fnNamespace, + fv1.FUNCTION_VERSION: versionName, + }.AsSelector().String() + listed, listErr := executorUtils.FunctionObjectNames(ctx, deploy.kubernetesClient, ns, selector) + errs = errors.Join(errs, listErr) + for objName := range listed { + objNames[objName] = struct{}{} + } + + for objName := range objNames { + errs = errors.Join(errs, deploy.cleanupNewdeploy(ctx, ns, objName)) + } + + return errs +} + +// getObjName returns a unique name for kubernetes objects of function. A +// versioned Function (fv1.FUNCTION_VERSION label present, RFC-0025) gets a +// name suffixed with the version's bounded "-v" tail so each published +// version's Deployment/Service/HPA is distinct; see +// executorUtils.VersionedObjName for the truncation budget that keeps the +// whole name within the Kubernetes 63-char limit either way. The +// unversioned path is byte-identical to the pre-RFC-0025 behaviour. +func (deploy *NewDeploy) getObjName(fn *fv1.Function) string { + return executorUtils.VersionedObjName("newdeploy-", fn) +} + +func (deploy *NewDeploy) getDeployLabels(fnMeta metav1.ObjectMeta, envMeta metav1.ObjectMeta) map[string]string { + deployLabels := map[string]string{ + fv1.EXECUTOR_TYPE: string(fv1.ExecutorTypeNewdeploy), + fv1.ENVIRONMENT_NAME: envMeta.Name, + fv1.ENVIRONMENT_NAMESPACE: envMeta.Namespace, + fv1.ENVIRONMENT_UID: string(envMeta.UID), + fv1.FUNCTION_NAME: fnMeta.Name, + fv1.FUNCTION_NAMESPACE: fnMeta.Namespace, + fv1.FUNCTION_UID: string(fnMeta.UID), + } + maps.MergeStringMap(deployLabels, envMeta.Labels) + maps.MergeStringMap(deployLabels, fnMeta.Labels) + return deployLabels +} + +func (deploy *NewDeploy) getDeployAnnotations(fnMeta metav1.ObjectMeta, envMeta metav1.ObjectMeta) map[string]string { + deployAnnotations := maps.CopyStringMap(envMeta.Annotations) + maps.MergeStringMap(deployAnnotations, fnMeta.Annotations) + deployAnnotations[fv1.EXECUTOR_INSTANCEID_LABEL] = deploy.instanceID + deployAnnotations[fv1.FUNCTION_RESOURCE_VERSION] = fnMeta.ResourceVersion + return deployAnnotations +} + +// updateStatus is a function which updates status of update. +// Current implementation only logs messages, in future it will update function status +func (deploy *NewDeploy) updateStatus(fn *fv1.Function, err error, message string) { + deploy.logger.Error(nil, "function status update", "function", fn, "message", message) +} + +// IdleStrategy returns the newdeploy idle-reaping strategy (scale the function +// deployment down to MinScale), run by the shared idle reaper. checkEnv is true +// so it mirrors the previous behaviour of logging function services whose +// environment was deleted and skipping the pass on an environment-list error. +func (deploy *NewDeploy) IdleStrategy() idle.Strategy { + return idle.NewScaleDownStrategy(deploy.logger, fv1.ExecutorTypeNewdeploy, deploy.fissionClient, + deploy.fsCache, deploy.kubernetesClient, deploy.defaultIdlePodReapTime, deploy.objectReaperIntervalSecond, true) +} + +func (deploy *NewDeploy) DumpDebugInfo(ctx context.Context) error { + return nil +} + +// podTemplateChanged reports whether a function update changes anything that +// is baked into the deployment's pod template, so updateFunction knows to roll +// it. Every field the deployment builder reads must be listed here: a field +// that lands in the pod template but is missing from this diff leaves the CR +// and the running container disagreeing indefinitely, because updateFunction +// is the only steady-state path that rewrites the template (the RV-annotation +// catch-all in reconcileDeploymentSpec runs only on create-routed reconciles — +// first sight and drift-recreate — never on the plain update path). +func podTemplateChanged(oldFn, newFn *fv1.Function) bool { + return oldFn.Spec.Environment != newFn.Spec.Environment || + oldFn.Spec.Package.PackageRef != newFn.Spec.Package.PackageRef || + oldFn.Spec.Package.FunctionName != newFn.Spec.Package.FunctionName || + !apiequality.Semantic.DeepEqual(oldFn.Spec.Secrets, newFn.Spec.Secrets) || + !apiequality.Semantic.DeepEqual(oldFn.Spec.ConfigMaps, newFn.Spec.ConfigMaps) || + // RFC-0030 per-function env lands in the pod template too. + !apiequality.Semantic.DeepEqual(oldFn.Spec.Env, newFn.Spec.Env) || + !apiequality.Semantic.DeepEqual(oldFn.Spec.EnvFrom, newFn.Spec.EnvFrom) +}