“Naeel”
3b93c5dc8b
- DefaultNSResolver.RemoveNamespace(): removes NS from global map on label removal so Snapshot() and idleObjectReaper stop iterating deleted namespaces. Fixes class of dirty-state bugs when NS name is reused by new tenant. - HandleWatcherNamespaceRemoval: call RemoveNamespace on both TrackOnly and DispatchRemove strategies — global resolver cleanup is always required. - dispatch(): parallel subscriber execution via goroutine per subscriber + sync.WaitGroup. Reduces onboarding latency from O(N_subscribers × API_latency) to O(max(API_latency)). Safe: MarkPart* are internally mutex-protected. - inMemoryNamespaceManager.RunReconciler(): 30s ticker scans for NamespacePhaseFailed records and retries via DispatchResync. Started automatically by RunManagedNamespaceWatcher. Fixes permanent stuck-failed state caused by transient k8s API errors. Analysis source: FORENSIC_ARCHITECTURE_AUDIT.md §Deep Risk Analysis
668 lines
22 KiB
Go
668 lines
22 KiB
Go
package utils
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"sort"
|
|
"sync"
|
|
"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"
|
|
|
|
managerPkg "github.com/fission/fission/pkg/utils/manager"
|
|
)
|
|
|
|
type NamespaceSubscriber interface {
|
|
Name() string
|
|
OnNamespaceAdd(ctx context.Context, record NamespaceRecord) error
|
|
OnNamespaceRemove(ctx context.Context, record NamespaceRecord) error
|
|
OnNamespaceResync(ctx context.Context, record NamespaceRecord) error
|
|
}
|
|
|
|
type NamespaceSubscriberFuncs struct {
|
|
SubscriberName string
|
|
AddFunc func(ctx context.Context, record NamespaceRecord) error
|
|
RemoveFunc func(ctx context.Context, record NamespaceRecord) error
|
|
ResyncFunc func(ctx context.Context, record NamespaceRecord) error
|
|
}
|
|
|
|
func (s NamespaceSubscriberFuncs) Name() string {
|
|
return s.SubscriberName
|
|
}
|
|
|
|
func (s NamespaceSubscriberFuncs) OnNamespaceAdd(ctx context.Context, record NamespaceRecord) error {
|
|
if s.AddFunc == nil {
|
|
return nil
|
|
}
|
|
return s.AddFunc(ctx, record)
|
|
}
|
|
|
|
func (s NamespaceSubscriberFuncs) OnNamespaceRemove(ctx context.Context, record NamespaceRecord) error {
|
|
if s.RemoveFunc == nil {
|
|
return nil
|
|
}
|
|
return s.RemoveFunc(ctx, record)
|
|
}
|
|
|
|
func (s NamespaceSubscriberFuncs) OnNamespaceResync(ctx context.Context, record NamespaceRecord) error {
|
|
if s.ResyncFunc == nil {
|
|
return nil
|
|
}
|
|
return s.ResyncFunc(ctx, record)
|
|
}
|
|
|
|
type NamespaceManager interface {
|
|
Snapshot() []string
|
|
SnapshotRecords() []NamespaceRecord
|
|
Summary() NamespaceManagerSummary
|
|
Get(name string) (NamespaceRecord, bool)
|
|
Bootstrap(namespaces []string, source NamespaceSource, observedAt time.Time) []NamespaceRecord
|
|
BootstrapAndDispatch(ctx context.Context, namespaces []string, source NamespaceSource, observedAt time.Time) ([]NamespaceRecord, error)
|
|
Subscribe(subscriber NamespaceSubscriber)
|
|
SnapshotSubscribers() []string
|
|
Upsert(event NamespaceEvent) NamespaceRecord
|
|
DispatchAdd(ctx context.Context, namespace string) (NamespaceRecord, bool, error)
|
|
DispatchRemove(ctx context.Context, namespace string) (NamespaceRecord, bool, error)
|
|
DispatchResync(ctx context.Context, namespace string) (NamespaceRecord, bool, error)
|
|
MarkPartState(namespace string, part string, state NamespacePartState) (NamespaceRecord, bool)
|
|
MarkPartRegistering(namespace string, part string) (NamespaceRecord, bool)
|
|
MarkPartActive(namespace string, part string) (NamespaceRecord, bool)
|
|
MarkPartFailed(namespace string, part string, err error) (NamespaceRecord, bool)
|
|
Remove(name string) bool
|
|
// RunReconciler periodically retries namespaces stuck in NamespacePhaseFailed.
|
|
// Must be started as a goroutine; exits when ctx is cancelled.
|
|
RunReconciler(ctx context.Context)
|
|
}
|
|
|
|
type inMemoryNamespaceManager struct {
|
|
mu sync.RWMutex
|
|
records map[string]NamespaceRecord
|
|
subs map[string]NamespaceSubscriber
|
|
}
|
|
|
|
func NewNamespaceManager() NamespaceManager {
|
|
return &inMemoryNamespaceManager{
|
|
records: make(map[string]NamespaceRecord),
|
|
subs: make(map[string]NamespaceSubscriber),
|
|
}
|
|
}
|
|
|
|
func NewBootstrappedNamespaceManager(resolver *NamespaceResolver, source NamespaceSource, observedAt time.Time) NamespaceManager {
|
|
manager := NewNamespaceManager()
|
|
if resolver == nil {
|
|
return manager
|
|
}
|
|
manager.Bootstrap(resolver.Snapshot(), source, observedAt)
|
|
return manager
|
|
}
|
|
|
|
func NewWatcherNamespaceManager(ctx context.Context, namespaces []string, source NamespaceSource, observedAt time.Time, subscribers ...NamespaceSubscriber) (NamespaceManager, error) {
|
|
manager := NewNamespaceManager()
|
|
for _, subscriber := range subscribers {
|
|
if subscriber == nil {
|
|
continue
|
|
}
|
|
manager.Subscribe(subscriber)
|
|
}
|
|
_, err := manager.BootstrapAndDispatch(ctx, namespaces, source, observedAt)
|
|
return manager, err
|
|
}
|
|
|
|
func NewDefaultManagedNamespaceWatcherConfig(component string, subscriber NamespaceSubscriber) ManagedNamespaceWatcherConfig {
|
|
return ManagedNamespaceWatcherConfig{
|
|
Component: component,
|
|
Namespaces: DefaultNSResolver().Snapshot(),
|
|
RemovalStrategy: NamespaceRemovalStrategyTrackOnly,
|
|
Subscriber: subscriber,
|
|
}
|
|
}
|
|
|
|
func namespaceManagerLogger(logger *zap.Logger) *zap.Logger {
|
|
if logger != nil {
|
|
return logger
|
|
}
|
|
return zap.NewNop()
|
|
}
|
|
|
|
func PrepareManagedNamespaceWatcher(ctx context.Context, logger *zap.Logger, config ManagedNamespaceWatcherConfig) (NamespaceManager, k8sCache.ResourceEventHandlerFuncs, error) {
|
|
logger = namespaceManagerLogger(logger)
|
|
strategy := config.RemovalStrategy
|
|
if strategy == "" {
|
|
strategy = NamespaceRemovalStrategyTrackOnly
|
|
}
|
|
manager, err := NewWatcherNamespaceManager(ctx, config.Namespaces, NamespaceSourceEnv, time.Now().UTC(), config.Subscriber)
|
|
handlers := NewNamespaceWatcherEventHandlers(ctx, logger, config.Component, manager, strategy)
|
|
LogNamespaceManagerSummary(logger, config.Component+": prepared namespace manager", manager.Summary())
|
|
return manager, handlers, err
|
|
}
|
|
|
|
func RunManagedNamespaceWatcher(ctx context.Context, logger *zap.Logger, kubeClient kubernetes.Interface, mgr managerPkg.Interface, config ManagedNamespaceWatcherConfig) (NamespaceManager, error) {
|
|
logger = namespaceManagerLogger(logger)
|
|
manager, handlers, err := PrepareManagedNamespaceWatcher(ctx, logger, config)
|
|
StartManagedNamespaceWatcher(ctx, logger, config.Component, kubeClient, mgr, handlers)
|
|
// Start reconciler: retries namespaces stuck in NamespacePhaseFailed every 30s.
|
|
mgr.Add(ctx, func(ctx context.Context) {
|
|
logger.Info(config.Component + ": namespace reconciler started")
|
|
manager.RunReconciler(ctx)
|
|
logger.Info(config.Component + ": namespace reconciler stopped")
|
|
})
|
|
LogNamespaceManagerSummary(logger, config.Component+": started namespace watcher", manager.Summary())
|
|
return manager, err
|
|
}
|
|
|
|
func LogNamespaceManagerSummary(logger *zap.Logger, message string, summary NamespaceManagerSummary) {
|
|
if logger == nil {
|
|
return
|
|
}
|
|
logger.Info(message,
|
|
zap.Int("total_namespaces", summary.TotalNamespaces),
|
|
zap.Int("live_namespaces", summary.LiveNamespaces),
|
|
zap.Bool("has_active_namespaces", summary.HasActiveNamespaces()),
|
|
zap.Any("phase_counts", summary.PhaseCounts),
|
|
zap.Any("source_counts", summary.SourceCounts),
|
|
zap.Strings("subscribers", summary.Subscribers),
|
|
)
|
|
}
|
|
|
|
func NamespaceBecameUnmanaged(oldNamespace *corev1.Namespace, newNamespace *corev1.Namespace) bool {
|
|
if oldNamespace == nil || newNamespace == nil {
|
|
return false
|
|
}
|
|
return IsManagedNamespace(oldNamespace.Labels) && !IsManagedNamespace(newNamespace.Labels)
|
|
}
|
|
|
|
func DispatchNamespaceAdd(ctx context.Context, manager NamespaceManager, namespace *corev1.Namespace, source NamespaceSource, observedAt time.Time) (NamespaceRecord, bool, error) {
|
|
if manager == nil || namespace == nil || namespace.Name == "" {
|
|
return NamespaceRecord{}, false, nil
|
|
}
|
|
manager.Upsert(NamespaceEventFromNamespace(NamespaceEventAdd, namespace, source, observedAt))
|
|
return manager.DispatchAdd(ctx, namespace.Name)
|
|
}
|
|
|
|
func DispatchNamespaceResync(ctx context.Context, manager NamespaceManager, namespace *corev1.Namespace, source NamespaceSource, observedAt time.Time) (NamespaceRecord, bool, error) {
|
|
if manager == nil || namespace == nil || namespace.Name == "" {
|
|
return NamespaceRecord{}, false, nil
|
|
}
|
|
manager.Upsert(NamespaceEventFromNamespace(NamespaceEventUpdate, namespace, source, observedAt))
|
|
return manager.DispatchResync(ctx, namespace.Name)
|
|
}
|
|
|
|
func RecordNamespaceRemoval(manager NamespaceManager, obj interface{}, source NamespaceSource, observedAt time.Time) (NamespaceRecord, bool) {
|
|
if manager == nil {
|
|
return NamespaceRecord{}, false
|
|
}
|
|
event := NamespaceEventFromObject(NamespaceEventRemove, obj, source, observedAt)
|
|
if event.Name == "" {
|
|
return NamespaceRecord{}, false
|
|
}
|
|
record := manager.Upsert(event)
|
|
return record, true
|
|
}
|
|
|
|
func HandleWatcherNamespaceAdd(ctx context.Context, logger *zap.Logger, component string, manager NamespaceManager, namespace *corev1.Namespace) {
|
|
logger = namespaceManagerLogger(logger)
|
|
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))
|
|
return
|
|
}
|
|
LogNamespaceManagerSummary(logger, component+": namespace manager summary", manager.Summary())
|
|
}
|
|
|
|
func HandleWatcherNamespaceUpdate(ctx context.Context, logger *zap.Logger, component string, manager NamespaceManager, oldNamespace *corev1.Namespace, newNamespace *corev1.Namespace, strategy NamespaceRemovalStrategy) {
|
|
logger = namespaceManagerLogger(logger)
|
|
if newNamespace == nil {
|
|
return
|
|
}
|
|
if NamespaceBecameUnmanaged(oldNamespace, newNamespace) {
|
|
HandleWatcherNamespaceRemoval(ctx, logger, component, manager, newNamespace, strategy)
|
|
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))
|
|
return
|
|
}
|
|
LogNamespaceManagerSummary(logger, component+": namespace manager summary", manager.Summary())
|
|
}
|
|
|
|
func HandleWatcherNamespaceDelete(ctx context.Context, logger *zap.Logger, component string, manager NamespaceManager, obj interface{}, strategy NamespaceRemovalStrategy) {
|
|
HandleWatcherNamespaceRemoval(ctx, logger, component, manager, obj, strategy)
|
|
}
|
|
|
|
func HandleWatcherNamespaceRemoval(ctx context.Context, logger *zap.Logger, component string, manager NamespaceManager, obj interface{}, strategy NamespaceRemovalStrategy) {
|
|
logger = namespaceManagerLogger(logger)
|
|
record, ok := RecordNamespaceRemoval(manager, obj, NamespaceSourceWatcher, time.Now().UTC())
|
|
if !ok {
|
|
return
|
|
}
|
|
// Always remove from global resolver so Snapshot() and idleObjectReaper
|
|
// stop iterating this namespace. This is safe: if the same name is re-added
|
|
// later, AddNamespace will return true and all subscribers will re-register.
|
|
DefaultNSResolver().RemoveNamespace(record.Name)
|
|
|
|
if strategy == NamespaceRemovalStrategyDispatchRemove {
|
|
if _, _, err := manager.DispatchRemove(ctx, record.Name); err != nil {
|
|
logger.Error(component+": DispatchRemove failed", zap.String("namespace", record.Name), zap.Error(err))
|
|
return
|
|
}
|
|
logger.Info(component+": namespace removal dispatched through subscribers",
|
|
zap.String("namespace", record.Name))
|
|
LogNamespaceManagerSummary(logger, component+": namespace manager summary", manager.Summary())
|
|
return
|
|
}
|
|
logger.Info(component+": namespace removed from manager state; runtime registrations kept",
|
|
zap.String("namespace", record.Name))
|
|
LogNamespaceManagerSummary(logger, component+": namespace manager summary", manager.Summary())
|
|
}
|
|
|
|
func NewNamespaceWatcherEventHandlers(ctx context.Context, logger *zap.Logger, component string, manager NamespaceManager, strategy NamespaceRemovalStrategy) k8sCache.ResourceEventHandlerFuncs {
|
|
logger = namespaceManagerLogger(logger)
|
|
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 StartManagedNamespaceWatcher(ctx context.Context, logger *zap.Logger, component string, kubeClient kubernetes.Interface, mgr managerPkg.Interface, handlers k8sCache.ResourceEventHandlerFuncs) {
|
|
logger = namespaceManagerLogger(logger)
|
|
factory := k8sInformers.NewSharedInformerFactoryWithOptions(
|
|
kubeClient,
|
|
30*time.Minute,
|
|
k8sInformers.WithTweakListOptions(func(opts *metav1.ListOptions) {
|
|
opts.LabelSelector = ManagedNamespaceLabelSelector()
|
|
}),
|
|
)
|
|
|
|
nsInformer := factory.Core().V1().Namespaces().Informer()
|
|
_, _ = nsInformer.AddEventHandler(handlers)
|
|
|
|
mgr.Add(ctx, func(ctx context.Context) {
|
|
logger.Info(component+": started", zap.String("label", ManagedNamespaceLabelSelector()))
|
|
factory.Start(ctx.Done())
|
|
factory.WaitForCacheSync(ctx.Done())
|
|
logger.Info(component + ": cache synced — watching for new namespaces")
|
|
<-ctx.Done()
|
|
logger.Info(component + ": stopped")
|
|
})
|
|
}
|
|
|
|
func (m *inMemoryNamespaceManager) Summary() NamespaceManagerSummary {
|
|
records := m.SnapshotRecords()
|
|
summary := NamespaceManagerSummary{
|
|
TotalNamespaces: len(records),
|
|
LiveNamespaces: len(m.Snapshot()),
|
|
PhaseCounts: make(map[NamespacePhase]int),
|
|
SourceCounts: make(map[NamespaceSource]int),
|
|
Subscribers: m.SnapshotSubscribers(),
|
|
}
|
|
if summary.Subscribers == nil {
|
|
summary.Subscribers = []string{}
|
|
}
|
|
for _, record := range records {
|
|
summary.PhaseCounts[record.Phase]++
|
|
summary.SourceCounts[record.Source]++
|
|
}
|
|
return summary
|
|
}
|
|
|
|
func (m *inMemoryNamespaceManager) Subscribe(subscriber NamespaceSubscriber) {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
m.subs[subscriber.Name()] = subscriber
|
|
}
|
|
|
|
func (m *inMemoryNamespaceManager) Bootstrap(namespaces []string, source NamespaceSource, observedAt time.Time) []NamespaceRecord {
|
|
records := make([]NamespaceRecord, 0, len(namespaces))
|
|
for _, namespace := range namespaces {
|
|
records = append(records, m.Upsert(NamespaceEvent{
|
|
Type: NamespaceEventAdd,
|
|
Name: namespace,
|
|
Source: source,
|
|
ObservedAt: observedAt,
|
|
}))
|
|
}
|
|
sort.Slice(records, func(i, j int) bool {
|
|
return records[i].Name < records[j].Name
|
|
})
|
|
return records
|
|
}
|
|
|
|
func (m *inMemoryNamespaceManager) BootstrapAndDispatch(ctx context.Context, namespaces []string, source NamespaceSource, observedAt time.Time) ([]NamespaceRecord, error) {
|
|
records := m.Bootstrap(namespaces, source, observedAt)
|
|
var joinErr error
|
|
for _, record := range records {
|
|
updatedRecord, ok, err := m.DispatchAdd(ctx, record.Name)
|
|
if ok {
|
|
record = updatedRecord
|
|
}
|
|
if err != nil {
|
|
joinErr = errors.Join(joinErr, err)
|
|
}
|
|
for index := range records {
|
|
if records[index].Name == record.Name {
|
|
records[index] = record
|
|
break
|
|
}
|
|
}
|
|
}
|
|
return records, joinErr
|
|
}
|
|
|
|
func (m *inMemoryNamespaceManager) SnapshotSubscribers() []string {
|
|
m.mu.RLock()
|
|
defer m.mu.RUnlock()
|
|
|
|
names := make([]string, 0, len(m.subs))
|
|
for name := range m.subs {
|
|
names = append(names, name)
|
|
}
|
|
sort.Strings(names)
|
|
return names
|
|
}
|
|
|
|
func (m *inMemoryNamespaceManager) snapshotSubscriberObjects() []NamespaceSubscriber {
|
|
m.mu.RLock()
|
|
defer m.mu.RUnlock()
|
|
|
|
names := make([]string, 0, len(m.subs))
|
|
for name := range m.subs {
|
|
names = append(names, name)
|
|
}
|
|
sort.Strings(names)
|
|
|
|
subscribers := make([]NamespaceSubscriber, 0, len(names))
|
|
for _, name := range names {
|
|
subscribers = append(subscribers, m.subs[name])
|
|
}
|
|
return subscribers
|
|
}
|
|
|
|
func (m *inMemoryNamespaceManager) Snapshot() []string {
|
|
m.mu.RLock()
|
|
defer m.mu.RUnlock()
|
|
|
|
namespaces := make([]string, 0, len(m.records))
|
|
for name, record := range m.records {
|
|
if record.Phase == NamespacePhaseRemoved {
|
|
continue
|
|
}
|
|
namespaces = append(namespaces, name)
|
|
}
|
|
sort.Strings(namespaces)
|
|
return namespaces
|
|
}
|
|
|
|
func (m *inMemoryNamespaceManager) SnapshotRecords() []NamespaceRecord {
|
|
m.mu.RLock()
|
|
defer m.mu.RUnlock()
|
|
|
|
records := make([]NamespaceRecord, 0, len(m.records))
|
|
for _, record := range m.records {
|
|
records = append(records, record.Clone())
|
|
}
|
|
sort.Slice(records, func(i, j int) bool {
|
|
return records[i].Name < records[j].Name
|
|
})
|
|
return records
|
|
}
|
|
|
|
func (m *inMemoryNamespaceManager) Get(name string) (NamespaceRecord, bool) {
|
|
m.mu.RLock()
|
|
defer m.mu.RUnlock()
|
|
|
|
record, ok := m.records[name]
|
|
if !ok {
|
|
return NamespaceRecord{}, false
|
|
}
|
|
return record.Clone(), true
|
|
}
|
|
|
|
func (m *inMemoryNamespaceManager) Upsert(event NamespaceEvent) NamespaceRecord {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
|
|
now := event.ObservedAt
|
|
if now.IsZero() {
|
|
now = time.Now().UTC()
|
|
}
|
|
|
|
record, exists := m.records[event.Name]
|
|
if !exists {
|
|
record = NamespaceRecord{
|
|
Name: event.Name,
|
|
RegisteredParts: make(map[string]NamespacePartState),
|
|
}
|
|
if event.Type == NamespaceEventRemove {
|
|
record.Generation = 0
|
|
} else {
|
|
record.Generation = 1
|
|
}
|
|
} else if event.Type != NamespaceEventRemove {
|
|
record.Generation++
|
|
}
|
|
|
|
record.Name = event.Name
|
|
record.Source = event.Source
|
|
record.Labels = cloneStringMap(event.Labels)
|
|
record.UpdatedAt = now
|
|
|
|
switch event.Type {
|
|
case NamespaceEventAdd, NamespaceEventUpdate, NamespaceEventResync:
|
|
record.Phase = NamespacePhaseDiscovered
|
|
record.LastError = ""
|
|
case NamespaceEventRemove:
|
|
record.Phase = NamespacePhaseRemoved
|
|
}
|
|
|
|
if record.RegisteredParts == nil {
|
|
record.RegisteredParts = make(map[string]NamespacePartState)
|
|
}
|
|
|
|
m.records[event.Name] = record
|
|
return record.Clone()
|
|
}
|
|
|
|
func (m *inMemoryNamespaceManager) DispatchAdd(ctx context.Context, namespace string) (NamespaceRecord, bool, error) {
|
|
return m.dispatch(ctx, namespace, func(subscriber NamespaceSubscriber, record NamespaceRecord) error {
|
|
return subscriber.OnNamespaceAdd(ctx, record)
|
|
})
|
|
}
|
|
|
|
func (m *inMemoryNamespaceManager) DispatchRemove(ctx context.Context, namespace string) (NamespaceRecord, bool, error) {
|
|
record, ok := m.Get(namespace)
|
|
if !ok {
|
|
return NamespaceRecord{}, false, nil
|
|
}
|
|
|
|
var firstErr error
|
|
for _, subscriber := range m.snapshotSubscriberObjects() {
|
|
err := subscriber.OnNamespaceRemove(ctx, record)
|
|
if err != nil && firstErr == nil {
|
|
firstErr = err
|
|
}
|
|
}
|
|
|
|
removedRecord := m.Upsert(NamespaceEvent{
|
|
Type: NamespaceEventRemove,
|
|
Name: record.Name,
|
|
Labels: record.Labels,
|
|
Source: record.Source,
|
|
ObservedAt: time.Now().UTC(),
|
|
})
|
|
return removedRecord, true, firstErr
|
|
}
|
|
|
|
func (m *inMemoryNamespaceManager) DispatchResync(ctx context.Context, namespace string) (NamespaceRecord, bool, error) {
|
|
return m.dispatch(ctx, namespace, func(subscriber NamespaceSubscriber, record NamespaceRecord) error {
|
|
return subscriber.OnNamespaceResync(ctx, record)
|
|
})
|
|
}
|
|
|
|
func (m *inMemoryNamespaceManager) dispatch(ctx context.Context, namespace string, handler func(NamespaceSubscriber, NamespaceRecord) error) (NamespaceRecord, bool, error) {
|
|
_, ok := m.Get(namespace)
|
|
if !ok {
|
|
return NamespaceRecord{}, false, nil
|
|
}
|
|
|
|
subscribers := m.snapshotSubscriberObjects()
|
|
for _, sub := range subscribers {
|
|
_, _ = m.MarkPartRegistering(namespace, sub.Name())
|
|
}
|
|
|
|
// Run all subscriber handlers in parallel — each handler makes independent k8s API calls.
|
|
// MarkPart* methods are internally mutex-protected and safe for concurrent calls.
|
|
var (
|
|
wg sync.WaitGroup
|
|
mu sync.Mutex
|
|
errs error
|
|
)
|
|
for _, sub := range subscribers {
|
|
sub := sub
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
currentRecord, _ := m.Get(namespace)
|
|
err := handler(sub, currentRecord)
|
|
if err != nil {
|
|
_, _ = m.MarkPartFailed(namespace, sub.Name(), err)
|
|
mu.Lock()
|
|
errs = errors.Join(errs, err)
|
|
mu.Unlock()
|
|
return
|
|
}
|
|
_, _ = m.MarkPartActive(namespace, sub.Name())
|
|
}()
|
|
}
|
|
wg.Wait()
|
|
|
|
record, _ := m.Get(namespace)
|
|
return record, true, errs
|
|
}
|
|
|
|
func (m *inMemoryNamespaceManager) MarkPartState(namespace string, part string, state NamespacePartState) (NamespaceRecord, bool) {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
|
|
record, ok := m.records[namespace]
|
|
if !ok {
|
|
return NamespaceRecord{}, false
|
|
}
|
|
if record.RegisteredParts == nil {
|
|
record.RegisteredParts = make(map[string]NamespacePartState)
|
|
}
|
|
if state.UpdatedAt.IsZero() {
|
|
state.UpdatedAt = time.Now().UTC()
|
|
}
|
|
record.RegisteredParts[part] = state
|
|
record.Phase = deriveNamespacePhase(record)
|
|
record.UpdatedAt = state.UpdatedAt
|
|
m.records[namespace] = record
|
|
return record.Clone(), true
|
|
}
|
|
|
|
func (m *inMemoryNamespaceManager) MarkPartRegistering(namespace string, part string) (NamespaceRecord, bool) {
|
|
return m.MarkPartState(namespace, part, NamespacePartState{State: NamespacePartStateRegistering})
|
|
}
|
|
|
|
func (m *inMemoryNamespaceManager) MarkPartActive(namespace string, part string) (NamespaceRecord, bool) {
|
|
return m.MarkPartState(namespace, part, NamespacePartState{State: NamespacePartStateActive})
|
|
}
|
|
|
|
func (m *inMemoryNamespaceManager) MarkPartFailed(namespace string, part string, err error) (NamespaceRecord, bool) {
|
|
lastError := ""
|
|
if err != nil {
|
|
lastError = err.Error()
|
|
}
|
|
return m.MarkPartState(namespace, part, NamespacePartState{State: NamespacePartStateFailed, LastError: lastError})
|
|
}
|
|
|
|
func (m *inMemoryNamespaceManager) Remove(name string) bool {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
|
|
if _, ok := m.records[name]; !ok {
|
|
return false
|
|
}
|
|
delete(m.records, name)
|
|
return true
|
|
}
|
|
|
|
// RunReconciler periodically finds namespaces in NamespacePhaseFailed and retries them
|
|
// via DispatchResync. This ensures transient k8s API errors (e.g. temporary 503) do not
|
|
// permanently strand a namespace. Exits when ctx is cancelled.
|
|
func (m *inMemoryNamespaceManager) RunReconciler(ctx context.Context) {
|
|
ticker := time.NewTicker(30 * time.Second)
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-ticker.C:
|
|
m.mu.RLock()
|
|
var failedNS []string
|
|
for ns, rec := range m.records {
|
|
if rec.Phase == NamespacePhaseFailed {
|
|
failedNS = append(failedNS, ns)
|
|
}
|
|
}
|
|
m.mu.RUnlock()
|
|
for _, ns := range failedNS {
|
|
if _, _, err := m.DispatchResync(ctx, ns); err != nil {
|
|
// Still failing — will retry on next tick.
|
|
_ = err
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func deriveNamespacePhase(record NamespaceRecord) NamespacePhase {
|
|
if len(record.RegisteredParts) == 0 {
|
|
return record.Phase
|
|
}
|
|
|
|
hasRegistering := false
|
|
for _, part := range record.RegisteredParts {
|
|
switch part.State {
|
|
case NamespacePartStateFailed:
|
|
return NamespacePhaseFailed
|
|
case NamespacePartStateRegistering:
|
|
hasRegistering = true
|
|
case NamespacePartStateActive:
|
|
continue
|
|
default:
|
|
hasRegistering = true
|
|
}
|
|
}
|
|
|
|
if hasRegistering {
|
|
return NamespacePhaseRegistering
|
|
}
|
|
|
|
return NamespacePhaseActive
|
|
}
|