Upstream fission/fission state of the touched files before 56824b6df3

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<seq> 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<seq> 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<seq>" 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
+}

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()))
+	})
+}

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

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<seq> 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<seq> 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<seq>" 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)
+}