214 lines
8.5 KiB
Go
214 lines
8.5 KiB
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"
|
|
"time"
|
|
|
|
"go.uber.org/zap"
|
|
corev1 "k8s.io/api/core/v1"
|
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
|
k8sInformers "k8s.io/client-go/informers"
|
|
"k8s.io/client-go/kubernetes"
|
|
k8sCache "k8s.io/client-go/tools/cache"
|
|
|
|
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,
|
|
) {
|
|
nsManager := utils.NewNamespaceManager()
|
|
nsManager.Subscribe(NewNamespaceSubscriber(logger, kubernetesClient, executorTypes, mgr))
|
|
if _, err := nsManager.BootstrapAndDispatch(ctx, utils.DefaultNSResolver().Snapshot(), utils.NamespaceSourceEnv, time.Now().UTC()); err != nil {
|
|
logger.Error("multitenant.NSWatcher: BootstrapAndDispatch failed", zap.Error(err))
|
|
}
|
|
|
|
// Use a label-filtered informer so only Namespaces with our label are delivered.
|
|
// The resync period of 30m is standard for Fission informers — it re-lists to recover
|
|
// from any missed events, but normal operation is purely event-driven (no ticking).
|
|
factory := k8sInformers.NewSharedInformerFactoryWithOptions(
|
|
kubernetesClient,
|
|
30*time.Minute,
|
|
k8sInformers.WithTweakListOptions(func(opts *metav1.ListOptions) {
|
|
opts.LabelSelector = utils.ManagedNamespaceLabelSelector()
|
|
}),
|
|
)
|
|
|
|
nsInformer := factory.Core().V1().Namespaces().Informer()
|
|
|
|
_, _ = nsInformer.AddEventHandler(k8sCache.ResourceEventHandlerFuncs{
|
|
// AddFunc fires when a new Namespace with the label appears.
|
|
AddFunc: func(obj interface{}) {
|
|
nsObj, ok := obj.(*corev1.Namespace)
|
|
if !ok || nsObj.Name == "" {
|
|
return
|
|
}
|
|
nsManager.Upsert(utils.NamespaceEventFromNamespace(utils.NamespaceEventAdd, nsObj, utils.NamespaceSourceWatcher, time.Now().UTC()))
|
|
if _, _, err := nsManager.DispatchAdd(ctx, nsObj.Name); err != nil {
|
|
logger.Error("multitenant.NSWatcher: DispatchAdd failed",
|
|
zap.String("namespace", nsObj.Name), zap.Error(err))
|
|
}
|
|
},
|
|
// UpdateFunc fires when an existing Namespace is updated — covers the case
|
|
// where the label is added to a pre-existing Namespace.
|
|
UpdateFunc: func(oldObj, newObj interface{}) {
|
|
nsObj, ok := newObj.(*corev1.Namespace)
|
|
if !ok {
|
|
return
|
|
}
|
|
oldNSObj, _ := oldObj.(*corev1.Namespace)
|
|
if oldNSObj != nil && utils.IsManagedNamespace(oldNSObj.Labels) && !utils.IsManagedNamespace(nsObj.Labels) {
|
|
event := utils.NamespaceEventFromNamespace(utils.NamespaceEventRemove, nsObj, utils.NamespaceSourceWatcher, time.Now().UTC())
|
|
nsManager.Upsert(event)
|
|
logger.Info("multitenant.NSWatcher: namespace removed from manager state; runtime registrations kept",
|
|
zap.String("namespace", nsObj.Name))
|
|
return
|
|
}
|
|
if !utils.IsManagedNamespace(nsObj.Labels) {
|
|
return // label was removed — nothing to do (executor keeps existing registrations)
|
|
}
|
|
nsManager.Upsert(utils.NamespaceEventFromNamespace(utils.NamespaceEventUpdate, nsObj, utils.NamespaceSourceWatcher, time.Now().UTC()))
|
|
if _, _, err := nsManager.DispatchResync(ctx, nsObj.Name); err != nil {
|
|
logger.Error("multitenant.NSWatcher: DispatchResync failed",
|
|
zap.String("namespace", nsObj.Name), zap.Error(err))
|
|
}
|
|
},
|
|
DeleteFunc: func(obj interface{}) {
|
|
event := utils.NamespaceEventFromObject(utils.NamespaceEventRemove, obj, utils.NamespaceSourceWatcher, time.Now().UTC())
|
|
if event.Name == "" {
|
|
return
|
|
}
|
|
nsManager.Upsert(event)
|
|
logger.Info("multitenant.NSWatcher: namespace deleted from manager state; runtime registrations kept",
|
|
zap.String("namespace", event.Name))
|
|
},
|
|
})
|
|
|
|
mgr.Add(ctx, func(ctx context.Context) {
|
|
logger.Info("multitenant.NSWatcher: started", zap.String("label", utils.ManagedNamespaceLabelSelector()))
|
|
factory.Start(ctx.Done())
|
|
factory.WaitForCacheSync(ctx.Done())
|
|
logger.Info("multitenant.NSWatcher: cache synced — watching for new namespaces")
|
|
<-ctx.Done()
|
|
logger.Info("multitenant.NSWatcher: stopped")
|
|
})
|
|
}
|
|
|
|
// 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.
|
|
func registerNamespace(
|
|
ctx context.Context,
|
|
logger *zap.Logger,
|
|
kubernetesClient kubernetes.Interface,
|
|
ns string,
|
|
executorTypes map[fv1.ExecutorType]executortype.ExecutorType,
|
|
mgr manager.Interface,
|
|
) {
|
|
// 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.
|
|
utils.EnsureNamespaceSA(ctx, kubernetesClient, logger, ns)
|
|
registerExecutorTypes(ctx, logger, ns, executorTypes, mgr)
|
|
logger.Info("multitenant.NSWatcher: registered namespace", zap.String("namespace", ns))
|
|
}
|
|
|
|
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
|
|
}
|