fission-src/pkg/executor/multitenant/ns_watcher.go

// Package multitenant provides utilities for running Fission in a multi-tenant
// Kubernetes environment where user namespaces are created dynamically at runtime.
//
// # The Problem
//
// In a standard Fission installation, all resource namespaces must be enumerated in the
// FISSION_RESOURCE_NAMESPACES environment variable before the process starts. Adding a
// new user namespace requires patching that env var and triggering a rolling restart of
// every Fission component (executor, router, buildermgr, etc.) — causing ~30 seconds of
// downtime per new tenant.
//
// At scale this becomes a severe operational problem: with hundreds of new tenants being
// created continuously, the executor is in a permanent rolling restart loop, causing
// cascading failures for all existing users.
//
// # The Solution
//
// This package implements hot namespace registration without pod restarts.
// The mechanism is intentionally simple and decoupled from any specific platform:
//
//  1. A platform (a cloud console, a CI/CD system, an operator) creates a Kubernetes
//     Namespace and sets the label:
//     fission.io/managed=true
//
//  2. StartNSWatcher registers a Kubernetes Namespace Informer that receives an event
//     the moment a labeled Namespace is created or updated — no polling, no delay.
//
//  3. On the AddFunc / UpdateFunc callback NSWatcher calls AddNamespace on every
//     registered executor type (poolmgr, newdeploy, container). Each type creates
//     per-NS informer factories, pod listers, and event handlers — live, without restart.
//
//  4. NamespaceResolver.AddNamespace deduplicates — calling AddNamespace on an already-
//     registered namespace is always a safe no-op.
//
// # Backward Compatibility
//
// The FISSION_RESOURCE_NAMESPACES environment variable continues to work as before.
// Namespaces listed there are registered at startup and do not require the label.
// This package adds on top of the existing mechanism — it does not replace it.
//
// # Required RBAC
//
// The fission-executor ServiceAccount must be granted permission to list and watch
// Namespaces at the cluster scope. Apply the manifest at:
//
//	deploy/multitenant/rbac.yaml
//
// # Integration Contract
//
// The entire integration contract for external platforms is a single label on a Namespace:
//
//	apiVersion: v1
//	kind: Namespace
//	metadata:
//	  name: tenant-abc123
//	  labels:
//	    fission.io/managed: "true"
//
// No other coupling to Fission internals is required or expected.
package multitenant

import (
	"context"
	"errors"
	"fmt"

	"go.uber.org/zap"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/kubernetes"

	fv1 "github.com/fission/fission/pkg/apis/core/v1"
	"github.com/fission/fission/pkg/executor/executortype"
	"github.com/fission/fission/pkg/utils"
	"github.com/fission/fission/pkg/utils/manager"
)

// StartNSWatcher registers a Kubernetes Namespace Informer that reacts immediately
// when a Namespace with label fission.io/managed=true is created or relabeled.
//
// Unlike a polling approach, this uses the standard k8s Watch mechanism — the executor
// receives the event within milliseconds of the Namespace appearing, with zero wasted
// API calls between events.
//
// The informer is managed via mgr and shuts down cleanly when ctx is cancelled.
func StartNSWatcher(
	ctx context.Context,
	logger *zap.Logger,
	kubernetesClient kubernetes.Interface,
	executorTypes map[fv1.ExecutorType]executortype.ExecutorType,
	mgr manager.Interface,
) {
	config := utils.NewDefaultManagedNamespaceWatcherConfig("multitenant.NSWatcher", NewNamespaceSubscriber(logger, kubernetesClient, executorTypes, mgr))
	config.RemovalStrategy = utils.NamespaceRemovalStrategyDispatchRemove
	_, err := utils.RunManagedNamespaceWatcher(ctx, logger, kubernetesClient, mgr, config)
	if err != nil {
		logger.Error("multitenant.NSWatcher: BootstrapAndDispatch failed", zap.Error(err))
	}
}

// registerNamespace calls AddNamespace on every executor type for the given namespace.
// The global NamespaceResolver is updated here — once, before any executor type is called.
// Each executor type uses its own internal state for deduplication instead of the
// global resolver, so all executor types receive the AddNamespace call regardless of
// iteration order.
//
// Returns an error if SA provisioning or any executor-type initialization fails.
// The error is propagated to the NamespaceSubscriber so the NamespaceManager can
// mark the namespace as NamespacePhaseFailed and the reconciler will retry automatically.
func registerNamespace(
	ctx context.Context,
	logger *zap.Logger,
	kubernetesClient kubernetes.Interface,
	ns string,
	executorTypes map[fv1.ExecutorType]executortype.ExecutorType,
	mgr manager.Interface,
) error {
	// Update the global resolver once here. Each executor type must NOT call
	// DefaultNSResolver().AddNamespace() for dedup — they have their own checks.
	utils.DefaultNSResolver().AddNamespace(ns)
	// Ensure fission-fetcher SA exists in the new namespace so pool pods can start.
	// A failure here means function pods will crash (no SA to pull fetcher image) —
	// propagate so the reconciler retries until the API is available again.
	if err := utils.EnsureNamespaceSA(ctx, kubernetesClient, logger, ns); err != nil {
		return fmt.Errorf("EnsureNamespaceSA: %w", err)
	}
	if err := registerExecutorTypes(ctx, logger, ns, executorTypes, mgr); err != nil {
		return fmt.Errorf("registerExecutorTypes: %w", err)
	}
	logger.Info("multitenant.NSWatcher: registered namespace", zap.String("namespace", ns))
	return nil
}

func registerExecutorTypes(
	ctx context.Context,
	logger *zap.Logger,
	ns string,
	executorTypes map[fv1.ExecutorType]executortype.ExecutorType,
	mgr manager.Interface,
) error {
	var joinErr error

	for _, et := range executorTypes {
		if err := et.AddNamespace(ctx, ns, mgr); err != nil {
			logger.Error("multitenant.NSWatcher: AddNamespace failed",
				zap.String("namespace", ns),
				zap.Error(err),
			)
			joinErr = errors.Join(joinErr, err)
		}
	}
	return joinErr
}

// deregisterNamespace calls RemoveNamespace on every executor type for the given namespace.
// Called when a Namespace with label fission.io/managed=true is removed.
func deregisterNamespace(
	ctx context.Context,
	logger *zap.Logger,
	ns string,
	executorTypes map[fv1.ExecutorType]executortype.ExecutorType,
) error {
	var joinErr error

	for _, et := range executorTypes {
		if err := et.RemoveNamespace(ctx, ns); err != nil {
			logger.Error("multitenant.NSWatcher: RemoveNamespace failed",
				zap.String("namespace", ns),
				zap.Error(err),
			)
			joinErr = errors.Join(joinErr, err)
		}
	}
	logger.Info("multitenant.NSWatcher: deregistered namespace", zap.String("namespace", ns))
	return joinErr
}

// PreRegisterManagedNamespaces does a one-time synchronous List of all Namespaces
// labeled fission.io/managed=true and adds them to the global DefaultNSResolver.
//
// Called at executor startup BEFORE AdoptExistingResources and CleanupOldExecutorObjects
// so that adopt and cleanup cover managed (dynamic) namespaces, not just static ones
// from FISSION_RESOURCE_NAMESPACES. This closes the P2 race where old executor pods
// in managed NS were never adopted (causing unnecessary cold starts) and never cleaned
// (causing orphaned pod accumulation).
//
// Failure is non-fatal: a warning is logged and the executor proceeds with static NS only.
func PreRegisterManagedNamespaces(ctx context.Context, logger *zap.Logger, client kubernetes.Interface) {
	nsList, err := client.CoreV1().Namespaces().List(ctx, metav1.ListOptions{
		LabelSelector: utils.ManagedNamespaceLabelSelector(),
	})
	if err != nil {
		logger.Warn("PreRegisterManagedNamespaces: failed to list managed namespaces; adopt/cleanup will use static NS only",
			zap.Error(err))
		return
	}
	for i := range nsList.Items {
		utils.DefaultNSResolver().AddNamespace(nsList.Items[i].Name)
	}
	logger.Info("PreRegisterManagedNamespaces: pre-registered managed namespaces",
		zap.Int("count", len(nsList.Items)))
}