layer1: centralize namespace watcher handlers
This commit is contained in:
@@ -0,0 +1,43 @@
|
|||||||
|
# 2026-04-26 — NamespaceManager rewrite, step 39
|
||||||
|
|
||||||
|
## Цель шага
|
||||||
|
|
||||||
|
Свести три namespace watcher-а к одинаковому lifecycle поведению через общие handlers в `utils`.
|
||||||
|
|
||||||
|
## Что меняем
|
||||||
|
|
||||||
|
1. Добавляем helpers:
|
||||||
|
- `HandleWatcherNamespaceAdd()`
|
||||||
|
- `HandleWatcherNamespaceUpdate()`
|
||||||
|
- `HandleWatcherNamespaceDelete()`
|
||||||
|
2. Helpers централизуют:
|
||||||
|
- dispatch add/resync;
|
||||||
|
- remove bookkeeping;
|
||||||
|
- стандартное logging-сообщение.
|
||||||
|
3. `buildermgr`, `router`, `executor/multitenant` переходят на эти helpers.
|
||||||
|
|
||||||
|
## Что НЕ меняем
|
||||||
|
|
||||||
|
- не меняем runtime cleanup policy;
|
||||||
|
- не меняем manager state model.# 2026-04-26 — NamespaceManager rewrite, step 39
|
||||||
|
|
||||||
|
## Цель шага
|
||||||
|
|
||||||
|
Свести три namespace watcher-а к одинаковому lifecycle поведению через общие handlers в `utils`.
|
||||||
|
|
||||||
|
## Что меняем
|
||||||
|
|
||||||
|
1. Добавляем helpers:
|
||||||
|
- `HandleWatcherNamespaceAdd()`
|
||||||
|
- `HandleWatcherNamespaceUpdate()`
|
||||||
|
- `HandleWatcherNamespaceDelete()`
|
||||||
|
2. Helpers централизуют:
|
||||||
|
- dispatch add/resync;
|
||||||
|
- remove bookkeeping;
|
||||||
|
- стандартное logging-сообщение.
|
||||||
|
3. `buildermgr`, `router`, `executor/multitenant` переходят на эти helpers.
|
||||||
|
|
||||||
|
## Что НЕ меняем
|
||||||
|
|
||||||
|
- не меняем runtime cleanup policy;
|
||||||
|
- не меняем manager state model.
|
||||||
@@ -48,41 +48,19 @@ func StartNSWatcher(
|
|||||||
_, _ = nsInformer.AddEventHandler(k8sCache.ResourceEventHandlerFuncs{
|
_, _ = nsInformer.AddEventHandler(k8sCache.ResourceEventHandlerFuncs{
|
||||||
AddFunc: func(obj interface{}) {
|
AddFunc: func(obj interface{}) {
|
||||||
nsObj, ok := obj.(*corev1.Namespace)
|
nsObj, ok := obj.(*corev1.Namespace)
|
||||||
if !ok || nsObj.Name == "" {
|
if ok {
|
||||||
return
|
utils.HandleWatcherNamespaceAdd(ctx, logger, "buildermgr.NSWatcher", nsManager, nsObj)
|
||||||
}
|
|
||||||
if _, _, err := utils.DispatchNamespaceAdd(ctx, nsManager, nsObj, utils.NamespaceSourceWatcher, time.Now().UTC()); err != nil {
|
|
||||||
logger.Error("buildermgr.NSWatcher: DispatchAdd failed",
|
|
||||||
zap.String("namespace", nsObj.Name), zap.Error(err))
|
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
UpdateFunc: func(oldObj, newObj interface{}) {
|
UpdateFunc: func(oldObj, newObj interface{}) {
|
||||||
nsObj, ok := newObj.(*corev1.Namespace)
|
|
||||||
if !ok {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
oldNSObj, _ := oldObj.(*corev1.Namespace)
|
oldNSObj, _ := oldObj.(*corev1.Namespace)
|
||||||
if utils.NamespaceBecameUnmanaged(oldNSObj, nsObj) {
|
nsObj, ok := newObj.(*corev1.Namespace)
|
||||||
_, _ = utils.RecordNamespaceRemoval(nsManager, nsObj, utils.NamespaceSourceWatcher, time.Now().UTC())
|
if ok {
|
||||||
logger.Info("buildermgr.NSWatcher: namespace removed from manager state; runtime registrations kept",
|
utils.HandleWatcherNamespaceUpdate(ctx, logger, "buildermgr.NSWatcher", nsManager, oldNSObj, nsObj)
|
||||||
zap.String("namespace", nsObj.Name))
|
|
||||||
return
|
|
||||||
}
|
|
||||||
if !utils.IsManagedNamespace(nsObj.Labels) {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
if _, _, err := utils.DispatchNamespaceResync(ctx, nsManager, nsObj, utils.NamespaceSourceWatcher, time.Now().UTC()); err != nil {
|
|
||||||
logger.Error("buildermgr.NSWatcher: DispatchResync failed",
|
|
||||||
zap.String("namespace", nsObj.Name), zap.Error(err))
|
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
DeleteFunc: func(obj interface{}) {
|
DeleteFunc: func(obj interface{}) {
|
||||||
record, ok := utils.RecordNamespaceRemoval(nsManager, obj, utils.NamespaceSourceWatcher, time.Now().UTC())
|
utils.HandleWatcherNamespaceDelete(logger, "buildermgr.NSWatcher", nsManager, obj)
|
||||||
if !ok {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
logger.Info("buildermgr.NSWatcher: namespace deleted from manager state; runtime registrations kept",
|
|
||||||
zap.String("namespace", record.Name))
|
|
||||||
},
|
},
|
||||||
})
|
})
|
||||||
|
|
||||||
|
|||||||
@@ -114,43 +114,21 @@ func StartNSWatcher(
|
|||||||
// AddFunc fires when a new Namespace with the label appears.
|
// AddFunc fires when a new Namespace with the label appears.
|
||||||
AddFunc: func(obj interface{}) {
|
AddFunc: func(obj interface{}) {
|
||||||
nsObj, ok := obj.(*corev1.Namespace)
|
nsObj, ok := obj.(*corev1.Namespace)
|
||||||
if !ok || nsObj.Name == "" {
|
if ok {
|
||||||
return
|
utils.HandleWatcherNamespaceAdd(ctx, logger, "multitenant.NSWatcher", nsManager, nsObj)
|
||||||
}
|
|
||||||
if _, _, err := utils.DispatchNamespaceAdd(ctx, nsManager, nsObj, utils.NamespaceSourceWatcher, time.Now().UTC()); 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
|
// UpdateFunc fires when an existing Namespace is updated — covers the case
|
||||||
// where the label is added to a pre-existing Namespace.
|
// where the label is added to a pre-existing Namespace.
|
||||||
UpdateFunc: func(oldObj, newObj interface{}) {
|
UpdateFunc: func(oldObj, newObj interface{}) {
|
||||||
nsObj, ok := newObj.(*corev1.Namespace)
|
|
||||||
if !ok {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
oldNSObj, _ := oldObj.(*corev1.Namespace)
|
oldNSObj, _ := oldObj.(*corev1.Namespace)
|
||||||
if utils.NamespaceBecameUnmanaged(oldNSObj, nsObj) {
|
nsObj, ok := newObj.(*corev1.Namespace)
|
||||||
_, _ = utils.RecordNamespaceRemoval(nsManager, nsObj, utils.NamespaceSourceWatcher, time.Now().UTC())
|
if ok {
|
||||||
logger.Info("multitenant.NSWatcher: namespace removed from manager state; runtime registrations kept",
|
utils.HandleWatcherNamespaceUpdate(ctx, logger, "multitenant.NSWatcher", nsManager, oldNSObj, nsObj)
|
||||||
zap.String("namespace", nsObj.Name))
|
|
||||||
return
|
|
||||||
}
|
|
||||||
if !utils.IsManagedNamespace(nsObj.Labels) {
|
|
||||||
return // label was removed — nothing to do (executor keeps existing registrations)
|
|
||||||
}
|
|
||||||
if _, _, err := utils.DispatchNamespaceResync(ctx, nsManager, nsObj, utils.NamespaceSourceWatcher, time.Now().UTC()); err != nil {
|
|
||||||
logger.Error("multitenant.NSWatcher: DispatchResync failed",
|
|
||||||
zap.String("namespace", nsObj.Name), zap.Error(err))
|
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
DeleteFunc: func(obj interface{}) {
|
DeleteFunc: func(obj interface{}) {
|
||||||
record, ok := utils.RecordNamespaceRemoval(nsManager, obj, utils.NamespaceSourceWatcher, time.Now().UTC())
|
utils.HandleWatcherNamespaceDelete(logger, "multitenant.NSWatcher", nsManager, obj)
|
||||||
if !ok {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
logger.Info("multitenant.NSWatcher: namespace deleted from manager state; runtime registrations kept",
|
|
||||||
zap.String("namespace", record.Name))
|
|
||||||
},
|
},
|
||||||
})
|
})
|
||||||
|
|
||||||
|
|||||||
@@ -48,41 +48,19 @@ func StartNSWatcher(
|
|||||||
_, _ = nsInformer.AddEventHandler(k8sCache.ResourceEventHandlerFuncs{
|
_, _ = nsInformer.AddEventHandler(k8sCache.ResourceEventHandlerFuncs{
|
||||||
AddFunc: func(obj interface{}) {
|
AddFunc: func(obj interface{}) {
|
||||||
nsObj, ok := obj.(*corev1.Namespace)
|
nsObj, ok := obj.(*corev1.Namespace)
|
||||||
if !ok || nsObj.Name == "" {
|
if ok {
|
||||||
return
|
utils.HandleWatcherNamespaceAdd(ctx, logger, "router.NSWatcher", nsManager, nsObj)
|
||||||
}
|
|
||||||
if _, _, err := utils.DispatchNamespaceAdd(ctx, nsManager, nsObj, utils.NamespaceSourceWatcher, time.Now().UTC()); err != nil {
|
|
||||||
logger.Error("router.NSWatcher: DispatchAdd failed",
|
|
||||||
zap.String("namespace", nsObj.Name), zap.Error(err))
|
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
UpdateFunc: func(oldObj, newObj interface{}) {
|
UpdateFunc: func(oldObj, newObj interface{}) {
|
||||||
nsObj, ok := newObj.(*corev1.Namespace)
|
|
||||||
if !ok {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
oldNSObj, _ := oldObj.(*corev1.Namespace)
|
oldNSObj, _ := oldObj.(*corev1.Namespace)
|
||||||
if utils.NamespaceBecameUnmanaged(oldNSObj, nsObj) {
|
nsObj, ok := newObj.(*corev1.Namespace)
|
||||||
_, _ = utils.RecordNamespaceRemoval(nsManager, nsObj, utils.NamespaceSourceWatcher, time.Now().UTC())
|
if ok {
|
||||||
logger.Info("router.NSWatcher: namespace removed from manager state; runtime registrations kept",
|
utils.HandleWatcherNamespaceUpdate(ctx, logger, "router.NSWatcher", nsManager, oldNSObj, nsObj)
|
||||||
zap.String("namespace", nsObj.Name))
|
|
||||||
return
|
|
||||||
}
|
|
||||||
if !utils.IsManagedNamespace(nsObj.Labels) {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
if _, _, err := utils.DispatchNamespaceResync(ctx, nsManager, nsObj, utils.NamespaceSourceWatcher, time.Now().UTC()); err != nil {
|
|
||||||
logger.Error("router.NSWatcher: DispatchResync failed",
|
|
||||||
zap.String("namespace", nsObj.Name), zap.Error(err))
|
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
DeleteFunc: func(obj interface{}) {
|
DeleteFunc: func(obj interface{}) {
|
||||||
record, ok := utils.RecordNamespaceRemoval(nsManager, obj, utils.NamespaceSourceWatcher, time.Now().UTC())
|
utils.HandleWatcherNamespaceDelete(logger, "router.NSWatcher", nsManager, obj)
|
||||||
if !ok {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
logger.Info("router.NSWatcher: namespace deleted from manager state; runtime registrations kept",
|
|
||||||
zap.String("namespace", record.Name))
|
|
||||||
},
|
},
|
||||||
})
|
})
|
||||||
|
|
||||||
|
|||||||
@@ -7,6 +7,7 @@ import (
|
|||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"go.uber.org/zap"
|
||||||
corev1 "k8s.io/api/core/v1"
|
corev1 "k8s.io/api/core/v1"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -137,6 +138,42 @@ func RecordNamespaceRemoval(manager NamespaceManager, obj interface{}, source Na
|
|||||||
return record, true
|
return record, true
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func HandleWatcherNamespaceAdd(ctx context.Context, logger *zap.Logger, component string, manager NamespaceManager, namespace *corev1.Namespace) {
|
||||||
|
if namespace == nil || namespace.Name == "" {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if _, _, err := DispatchNamespaceAdd(ctx, manager, namespace, NamespaceSourceWatcher, time.Now().UTC()); err != nil {
|
||||||
|
logger.Error(component+": DispatchAdd failed", zap.String("namespace", namespace.Name), zap.Error(err))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func HandleWatcherNamespaceUpdate(ctx context.Context, logger *zap.Logger, component string, manager NamespaceManager, oldNamespace *corev1.Namespace, newNamespace *corev1.Namespace) {
|
||||||
|
if newNamespace == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if NamespaceBecameUnmanaged(oldNamespace, newNamespace) {
|
||||||
|
_, _ = RecordNamespaceRemoval(manager, newNamespace, NamespaceSourceWatcher, time.Now().UTC())
|
||||||
|
logger.Info(component+": namespace removed from manager state; runtime registrations kept",
|
||||||
|
zap.String("namespace", newNamespace.Name))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if !IsManagedNamespace(newNamespace.Labels) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if _, _, err := DispatchNamespaceResync(ctx, manager, newNamespace, NamespaceSourceWatcher, time.Now().UTC()); err != nil {
|
||||||
|
logger.Error(component+": DispatchResync failed", zap.String("namespace", newNamespace.Name), zap.Error(err))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func HandleWatcherNamespaceDelete(logger *zap.Logger, component string, manager NamespaceManager, obj interface{}) {
|
||||||
|
record, ok := RecordNamespaceRemoval(manager, obj, NamespaceSourceWatcher, time.Now().UTC())
|
||||||
|
if !ok {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
logger.Info(component+": namespace deleted from manager state; runtime registrations kept",
|
||||||
|
zap.String("namespace", record.Name))
|
||||||
|
}
|
||||||
|
|
||||||
func (m *inMemoryNamespaceManager) Subscribe(subscriber NamespaceSubscriber) {
|
func (m *inMemoryNamespaceManager) Subscribe(subscriber NamespaceSubscriber) {
|
||||||
m.mu.Lock()
|
m.mu.Lock()
|
||||||
defer m.mu.Unlock()
|
defer m.mu.Unlock()
|
||||||
|
|||||||
@@ -7,6 +7,7 @@ import (
|
|||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"go.uber.org/zap"
|
||||||
corev1 "k8s.io/api/core/v1"
|
corev1 "k8s.io/api/core/v1"
|
||||||
k8sCache "k8s.io/client-go/tools/cache"
|
k8sCache "k8s.io/client-go/tools/cache"
|
||||||
)
|
)
|
||||||
@@ -337,6 +338,32 @@ func TestRecordNamespaceRemoval(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestHandleWatcherNamespaceAddUpdateDelete(t *testing.T) {
|
||||||
|
manager := NewNamespaceManager()
|
||||||
|
router := &testNamespaceSubscriber{name: "router"}
|
||||||
|
manager.Subscribe(router)
|
||||||
|
logger := zap.NewNop()
|
||||||
|
namespace := &corev1.Namespace{}
|
||||||
|
namespace.Name = "tenant-a"
|
||||||
|
namespace.Labels = map[string]string{ManagedNamespaceLabelKey: ManagedNamespaceLabelValue}
|
||||||
|
|
||||||
|
HandleWatcherNamespaceAdd(context.Background(), logger, "router.NSWatcher", manager, namespace)
|
||||||
|
if router.addCalls != 1 {
|
||||||
|
t.Fatalf("expected add handler to dispatch add")
|
||||||
|
}
|
||||||
|
|
||||||
|
HandleWatcherNamespaceUpdate(context.Background(), logger, "router.NSWatcher", manager, namespace, namespace)
|
||||||
|
if router.resyncCalls != 1 {
|
||||||
|
t.Fatalf("expected update handler to dispatch resync")
|
||||||
|
}
|
||||||
|
|
||||||
|
HandleWatcherNamespaceDelete(logger, "router.NSWatcher", manager, k8sCache.DeletedFinalStateUnknown{Obj: namespace})
|
||||||
|
record, ok := manager.Get("tenant-a")
|
||||||
|
if !ok || record.Phase != NamespacePhaseRemoved {
|
||||||
|
t.Fatalf("expected delete handler to mark namespace removed")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestNamespaceManagerDispatchAdd(t *testing.T) {
|
func TestNamespaceManagerDispatchAdd(t *testing.T) {
|
||||||
manager := NewNamespaceManager()
|
manager := NewNamespaceManager()
|
||||||
manager.Upsert(NamespaceEvent{Type: NamespaceEventAdd, Name: "tenant-a", Source: NamespaceSourceWatcher})
|
manager.Upsert(NamespaceEvent{Type: NamespaceEventAdd, Name: "tenant-a", Source: NamespaceSourceWatcher})
|
||||||
|
|||||||
Reference in New Issue
Block a user