Nail/fission-src
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

layer1: share namespace watcher event handlers

Browse Source
This commit is contained in:
Naeel
2026-04-26 10:50:28 +03:00
parent c3b161da83
commit 49be1db3a0
6 changed files with 65 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/buildermgr/ns_watcher.go
+1 -18
View File
@@ -45,24 +45,7 @@ func StartNSWatcher(
nsInformer := factory.Core().V1().Namespaces().Informer()
_, _ = nsInformer.AddEventHandler(k8sCache.ResourceEventHandlerFuncs{
AddFunc: func(obj interface{}) {
nsObj, ok := obj.(*corev1.Namespace)
if ok {
utils.HandleWatcherNamespaceAdd(ctx, logger, "buildermgr.NSWatcher", nsManager, nsObj)
}
},
UpdateFunc: func(oldObj, newObj interface{}) {
oldNSObj, _ := oldObj.(*corev1.Namespace)
nsObj, ok := newObj.(*corev1.Namespace)
if ok {
utils.HandleWatcherNamespaceUpdate(ctx, logger, "buildermgr.NSWatcher", nsManager, oldNSObj, nsObj, utils.NamespaceRemovalStrategyTrackOnly)
}
},
DeleteFunc: func(obj interface{}) {
utils.HandleWatcherNamespaceDelete(ctx, logger, "buildermgr.NSWatcher", nsManager, obj, utils.NamespaceRemovalStrategyTrackOnly)
},
})
_, _ = nsInformer.AddEventHandler(utils.NewNamespaceWatcherEventHandlers(ctx, logger, "buildermgr.NSWatcher", nsManager, utils.NamespaceRemovalStrategyTrackOnly))
mgr.Add(ctx, func(ctx context.Context) {
logger.Info("buildermgr.NSWatcher: started",
pkg/executor/multitenant/ns_watcher.go
+1 -21
View File
@@ -110,27 +110,7 @@ func StartNSWatcher(
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 {
utils.HandleWatcherNamespaceAdd(ctx, logger, "multitenant.NSWatcher", nsManager, nsObj)
}
},
// 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{}) {
oldNSObj, _ := oldObj.(*corev1.Namespace)
nsObj, ok := newObj.(*corev1.Namespace)
if ok {
utils.HandleWatcherNamespaceUpdate(ctx, logger, "multitenant.NSWatcher", nsManager, oldNSObj, nsObj, utils.NamespaceRemovalStrategyTrackOnly)
}
},
DeleteFunc: func(obj interface{}) {
utils.HandleWatcherNamespaceDelete(ctx, logger, "multitenant.NSWatcher", nsManager, obj, utils.NamespaceRemovalStrategyTrackOnly)
},
})
_, _ = nsInformer.AddEventHandler(utils.NewNamespaceWatcherEventHandlers(ctx, logger, "multitenant.NSWatcher", nsManager, utils.NamespaceRemovalStrategyTrackOnly))
mgr.Add(ctx, func(ctx context.Context) {
logger.Info("multitenant.NSWatcher: started", zap.String("label", utils.ManagedNamespaceLabelSelector()))
pkg/router/ns_watcher.go
+1 -18
View File
@@ -45,24 +45,7 @@ func StartNSWatcher(
nsInformer := factory.Core().V1().Namespaces().Informer()
_, _ = nsInformer.AddEventHandler(k8sCache.ResourceEventHandlerFuncs{
AddFunc: func(obj interface{}) {
nsObj, ok := obj.(*corev1.Namespace)
if ok {
utils.HandleWatcherNamespaceAdd(ctx, logger, "router.NSWatcher", nsManager, nsObj)
}
},
UpdateFunc: func(oldObj, newObj interface{}) {
oldNSObj, _ := oldObj.(*corev1.Namespace)
nsObj, ok := newObj.(*corev1.Namespace)
if ok {
utils.HandleWatcherNamespaceUpdate(ctx, logger, "router.NSWatcher", nsManager, oldNSObj, nsObj, utils.NamespaceRemovalStrategyTrackOnly)
}
},
DeleteFunc: func(obj interface{}) {
utils.HandleWatcherNamespaceDelete(ctx, logger, "router.NSWatcher", nsManager, obj, utils.NamespaceRemovalStrategyTrackOnly)
},
})
_, _ = nsInformer.AddEventHandler(utils.NewNamespaceWatcherEventHandlers(ctx, logger, "router.NSWatcher", nsManager, utils.NamespaceRemovalStrategyTrackOnly))
mgr.Add(ctx, func(ctx context.Context) {
logger.Info("router.NSWatcher: started",
pkg/utils/namespace_manager.go
+22
View File
@@ -9,6 +9,7 @@ import (
"go.uber.org/zap"
corev1 "k8s.io/api/core/v1"
k8sCache "k8s.io/client-go/tools/cache"
)
type NamespaceSubscriber interface {
@@ -185,6 +186,27 @@ func HandleWatcherNamespaceRemoval(ctx context.Context, logger *zap.Logger, comp
zap.String("namespace", record.Name))
}
func NewNamespaceWatcherEventHandlers(ctx context.Context, logger *zap.Logger, component string, manager NamespaceManager, strategy NamespaceRemovalStrategy) k8sCache.ResourceEventHandlerFuncs {
return k8sCache.ResourceEventHandlerFuncs{
AddFunc: func(obj interface{}) {
namespace, ok := obj.(*corev1.Namespace)
if ok {
HandleWatcherNamespaceAdd(ctx, logger, component, manager, namespace)
}
},
UpdateFunc: func(oldObj, newObj interface{}) {
oldNamespace, _ := oldObj.(*corev1.Namespace)
newNamespace, ok := newObj.(*corev1.Namespace)
if ok {
HandleWatcherNamespaceUpdate(ctx, logger, component, manager, oldNamespace, newNamespace, strategy)
}
},
DeleteFunc: func(obj interface{}) {
HandleWatcherNamespaceDelete(ctx, logger, component, manager, obj, strategy)
},
}
}
func (m *inMemoryNamespaceManager) Subscribe(subscriber NamespaceSubscriber) {
m.mu.Lock()
defer m.mu.Unlock()
pkg/utils/namespace_manager_test.go
+23
View File
@@ -406,6 +406,29 @@ func TestHandleWatcherNamespaceUpdateDispatchRemoval(t *testing.T) {
}
}
func TestNewNamespaceWatcherEventHandlers(t *testing.T) {
manager := NewNamespaceManager()
router := &testNamespaceSubscriber{name: "router"}
manager.Subscribe(router)
logger := zap.NewNop()
handlers := NewNamespaceWatcherEventHandlers(context.Background(), logger, "router.NSWatcher", manager, NamespaceRemovalStrategyTrackOnly)
namespace := &corev1.Namespace{}
namespace.Name = "tenant-a"
namespace.Labels = map[string]string{ManagedNamespaceLabelKey: ManagedNamespaceLabelValue}
handlers.AddFunc(namespace)
handlers.UpdateFunc(namespace, namespace)
handlers.DeleteFunc(k8sCache.DeletedFinalStateUnknown{Obj: namespace})
record, ok := manager.Get("tenant-a")
if !ok || record.Phase != NamespacePhaseRemoved {
t.Fatalf("expected watcher event handlers to drive namespace lifecycle")
}
if router.addCalls != 1 || router.resyncCalls != 1 {
t.Fatalf("expected add and resync calls through event handlers")
}
}
func TestNamespaceManagerDispatchAdd(t *testing.T) {
manager := NewNamespaceManager()
manager.Upsert(NamespaceEvent{Type: NamespaceEventAdd, Name: "tenant-a", Source: NamespaceSourceWatcher})