From c9223bb9f628727b0ea36647c261daf9f3c45a3a Mon Sep 17 00:00:00 2001 From: Christian Kruse Date: Mon, 22 Jun 2026 11:47:27 -0700 Subject: [PATCH] Recover Full Site Before Reconfiguring Rotuer Adds a new recovery mode to supress partial router ConfigMap writes with missing bridge configuration. Adds additional Controler init processes for recovering all resources that can affect site bindings prior to finalization, including selector connector pods and network status for exposeByPods listeners. Signed-off-by: Christian Kruse --- internal/kube/controller/controller.go | 147 +++++++++++++++----- internal/kube/controller/controller_test.go | 143 +++++++++++++++++++ internal/kube/site/bindings.go | 9 ++ internal/kube/site/bindings_test.go | 33 +++++ internal/kube/site/extended_bindings.go | 18 +++ internal/kube/site/site.go | 42 +++++- 6 files changed, 358 insertions(+), 34 deletions(-) diff --git a/internal/kube/controller/controller.go b/internal/kube/controller/controller.go index f484fa40d..1f2833290 100644 --- a/internal/kube/controller/controller.go +++ b/internal/kube/controller/controller.go @@ -8,6 +8,7 @@ import ( "log/slog" "os" "regexp" + "time" appsv1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" @@ -30,32 +31,35 @@ import ( ) type Controller struct { - self skupperv2alpha1.Controller - deploymentName string - deploymentUid string - eventProcessor *watchers.EventProcessor - stopCh <-chan struct{} - siteWatcher *watchers.SiteWatcher - listenerWatcher *watchers.ListenerWatcher - connectorWatcher *watchers.ConnectorWatcher - multiKeyListenerWatcher *watchers.MultiKeyListenerWatcher - linkAccessWatcher *watchers.RouterAccessWatcher - grantWatcher *watchers.AccessGrantWatcher - serviceWatcher *watchers.ServiceWatcher - sites map[string]*site.Site - startGrantServer func() - accessMgr *securedaccess.SecuredAccessManager - accessRecovery *securedaccess.SecuredAccessResourceWatcher - certMgr *certificates.CertificateManagerImpl - siteSizing *sizing.Registry - siteSizingWatcher *watchers.ConfigMapWatcher - labelling *labels.LabelsAndAnnotations - labellingWatcher *watchers.ConfigMapWatcher - attachableConnectors map[string]*skupperv2alpha1.AttachedConnector - disableSecContext bool - log *slog.Logger - namespaces *NamespaceConfig - observedServices map[string]string + self skupperv2alpha1.Controller + deploymentName string + deploymentUid string + eventProcessor *watchers.EventProcessor + stopCh <-chan struct{} + siteWatcher *watchers.SiteWatcher + listenerWatcher *watchers.ListenerWatcher + connectorWatcher *watchers.ConnectorWatcher + multiKeyListenerWatcher *watchers.MultiKeyListenerWatcher + linkAccessWatcher *watchers.RouterAccessWatcher + attachedConnectorWatcher *watchers.AttachedConnectorWatcher + attachedConnectorBindingWatcher *watchers.AttachedConnectorBindingWatcher + grantWatcher *watchers.AccessGrantWatcher + serviceWatcher *watchers.ServiceWatcher + networkStatusWatcher *watchers.ConfigMapWatcher + sites map[string]*site.Site + startGrantServer func() + accessMgr *securedaccess.SecuredAccessManager + accessRecovery *securedaccess.SecuredAccessResourceWatcher + certMgr *certificates.CertificateManagerImpl + siteSizing *sizing.Registry + siteSizingWatcher *watchers.ConfigMapWatcher + labelling *labels.LabelsAndAnnotations + labellingWatcher *watchers.ConfigMapWatcher + attachableConnectors map[string]*skupperv2alpha1.AttachedConnector + disableSecContext bool + log *slog.Logger + namespaces *NamespaceConfig + observedServices map[string]string } func skupperRouterConfig() internalinterfaces.TweakListOptionsFunc { @@ -136,10 +140,10 @@ func NewController(cli internalclient.Clients, config *Config, options ...watche controller.serviceWatcher = controller.eventProcessor.WatchServices(sansSkupperListenerServices(), config.WatchNamespace, filter(controller, controller.checkObservedService)) controller.connectorWatcher = controller.eventProcessor.WatchConnectors(config.WatchNamespace, filter(controller, controller.checkConnector)) controller.linkAccessWatcher = controller.eventProcessor.WatchRouterAccesses(config.WatchNamespace, filter(controller, controller.checkRouterAccess)) - controller.eventProcessor.WatchAttachedConnectors(config.WatchNamespace, filter(controller, controller.checkAttachedConnector)) - controller.eventProcessor.WatchAttachedConnectorBindings(config.WatchNamespace, filter(controller, controller.checkAttachedConnectorBinding)) + controller.attachedConnectorWatcher = controller.eventProcessor.WatchAttachedConnectors(config.WatchNamespace, filter(controller, controller.checkAttachedConnector)) + controller.attachedConnectorBindingWatcher = controller.eventProcessor.WatchAttachedConnectorBindings(config.WatchNamespace, filter(controller, controller.checkAttachedConnectorBinding)) controller.eventProcessor.WatchLinks(config.WatchNamespace, filter(controller, controller.checkLink)) - controller.eventProcessor.WatchConfigMaps(skupperNetworkStatus(), config.WatchNamespace, filter(controller, controller.networkStatusUpdate)) + controller.networkStatusWatcher = controller.eventProcessor.WatchConfigMaps(skupperNetworkStatus(), config.WatchNamespace, filter(controller, controller.networkStatusUpdate)) controller.eventProcessor.WatchConfigMaps(skupperRouterConfig(), config.WatchNamespace, filter(controller, controller.routerConfigUpdate)) controller.eventProcessor.WatchAccessTokens(config.WatchNamespace, filter(controller, controller.checkAccessToken)) controller.eventProcessor.WatchPods("skupper.io/component=router,skupper.io/type=site", config.WatchNamespace, filter(controller, controller.routerPodEvent)) @@ -238,12 +242,15 @@ func (c *Controller) init(stopCh <-chan struct{}) error { ) c.labelling.Update(config.Namespace+"/"+config.Name, config) } + c.certMgr.Recover() + c.accessRecovery.Recover() // get observed services prior to restoring listeners for _, svc := range c.serviceWatcher.List() { c.observedServices[svc.Namespace+"/"+svc.ObjectMeta.Name] = svc.ObjectMeta.Name } //recover existing sites & bindings siteRecovery := site.NewSiteRecovery(c.eventProcessor.GetKubeClient()) + recoveringSites := map[string]string{} for _, site := range c.siteWatcher.List() { if !c.namespaces.isControlled(site.Namespace) { continue @@ -266,6 +273,8 @@ func (c *Controller) init(stopCh <-chan struct{}) error { slog.String("name", site.Name), slog.Any("error", err), ) + } else { + recoveringSites[site.Namespace] = site.Name } } for _, connector := range c.connectorWatcher.List() { @@ -322,12 +331,88 @@ func (c *Controller) init(stopCh <-chan struct{}) error { ) site.CheckRouterAccess(la.ObjectMeta.Name, la) } + for _, binding := range c.attachedConnectorBindingWatcher.List() { + if !c.namespaces.isControlled(binding.Namespace) { + continue + } + c.log.Info("Recovering attached connector binding", + slog.String("namespace", binding.Namespace), + slog.String("name", binding.Name), + ) + if err := c.checkAttachedConnectorBinding(binding.Namespace+"/"+binding.Name, binding); err != nil { + c.log.Error("Error recovering attached connector binding", + slog.String("namespace", binding.Namespace), + slog.String("name", binding.Name), + slog.Any("error", err), + ) + } + } + for _, connector := range c.attachedConnectorWatcher.List() { + if !c.namespaces.isControlled(connector.Namespace) { + continue + } + c.log.Info("Recovering attached connector", + slog.String("namespace", connector.Namespace), + slog.String("name", connector.Name), + ) + if err := c.checkAttachedConnector(connector.Namespace+"/"+connector.Name, connector); err != nil { + c.log.Error("Error recovering attached connector", + slog.String("namespace", connector.Namespace), + slog.String("name", connector.Name), + slog.Any("error", err), + ) + } + } + for namespace, name := range recoveringSites { + syncCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + go func() { + select { + case <-stopCh: + cancel() + case <-syncCtx.Done(): + } + }() + siteController := c.getSite(namespace) + if !siteController.SyncConnectorPods(syncCtx.Done()) { + c.log.Error("Failed to sync connector pod watchers during recovery; continuing startup", + slog.String("namespace", namespace), + slog.String("name", name), + slog.Any("reason", syncCtx.Err()), + ) + } + if !siteController.SyncAttachedConnectorPods(syncCtx.Done()) { + c.log.Error("Failed to sync attached connector pod watchers during recovery; continuing startup", + slog.String("namespace", namespace), + slog.String("name", name), + slog.Any("reason", syncCtx.Err()), + ) + } + cancel() + } + for _, cm := range c.networkStatusWatcher.List() { + if !c.namespaces.isControlled(cm.Namespace) { + continue + } + if err := c.networkStatusUpdate(cm.Namespace+"/"+cm.Name, cm); err != nil { + c.log.Error("Error recovering network status", + slog.String("namespace", cm.Namespace), + slog.String("name", cm.Name), + slog.Any("error", err), + ) + } + } for _, site := range c.siteWatcher.List() { if !c.namespaces.isControlled(site.Namespace) { continue } site.Status.Controller = &c.self - err := c.getSite(site.ObjectMeta.Namespace).Reconcile(site) + siteController := c.getSite(site.ObjectMeta.Namespace) + var err error + if recoveringSites[site.Namespace] == site.Name { + err = siteController.FinishRecovery(site) + } else { + err = siteController.Reconcile(site) + } if err != nil { c.log.Error("Error recovering site", slog.String("namespace", site.Namespace), @@ -341,8 +426,6 @@ func (c *Controller) init(stopCh <-chan struct{}) error { ) } } - c.certMgr.Recover() - c.accessRecovery.Recover() if c.startGrantServer != nil { c.startGrantServer() } diff --git a/internal/kube/controller/controller_test.go b/internal/kube/controller/controller_test.go index fcc029fe2..d8801f79a 100644 --- a/internal/kube/controller/controller_test.go +++ b/internal/kube/controller/controller_test.go @@ -28,6 +28,7 @@ import ( discoveryfake "k8s.io/client-go/discovery/fake" "k8s.io/client-go/dynamic" fakedynamic "k8s.io/client-go/dynamic/fake" + k8sfake "k8s.io/client-go/kubernetes/fake" k8stesting "k8s.io/client-go/testing" "github.com/skupperproject/skupper/internal/fixtures" @@ -665,6 +666,148 @@ func TestGeneral(t *testing.T) { } } +func TestRecoveryPreservesRouterBridgeConfig(t *testing.T) { + flags := &flag.FlagSet{} + config, err := BoundConfig(flags) + assert.Assert(t, err) + + uid := "49b03ad4-d414-42be-bbb5-b32d7d4ca503" + site := f.addUID(f.site("mysite", "test", "", false, false), uid) + attachedConnectorBinding := f.attachedConnectorBinding("attached-a", "test", "attached") + attachedConnectorBinding.Spec.RoutingKey = "attached-a" + attachedPod := f.pod("pod-a", "attached", map[string]string{"app": "attached-a"}, nil, f.podStatus("10.1.1.20", corev1.PodRunning, f.podCondition(corev1.PodReady, corev1.ConditionTrue))) + attachedPod.UID = types.UID("6ffbd287-42cc-4b71-b0a7-44f1befcc847") + normalPod := f.pod("pod-b", "test", map[string]string{"app": "normal-a"}, nil, f.podStatus("10.1.1.30", corev1.PodRunning, f.podCondition(corev1.PodReady, corev1.ConditionTrue))) + normalPod.UID = types.UID("a2b30f15-4fe3-4788-a68d-36105d147435") + siteCA := &skupperv2alpha1.Certificate{ + ObjectMeta: metav1.ObjectMeta{ + Name: "skupper-site-ca", + Namespace: "test", + Labels: map[string]string{ + "internal.skupper.io/certificate": "true", + }, + Annotations: map[string]string{ + "internal.skupper.io/controlled": "true", + "internal.skupper.io/hosts-" + uid: "", + }, + OwnerReferences: []metav1.OwnerReference{ + { + APIVersion: skupperv2alpha1.SchemeGroupVersion.String(), + Kind: "Site", + Name: site.Name, + UID: types.UID(uid), + }, + }, + }, + Spec: skupperv2alpha1.CertificateSpec{ + Subject: "mysite site CA", + Signing: true, + }, + } + routerConfig := f.routerConfig("skupper-router", "test"). + tcpListener("listener/listener-a", "1024", "", ""). + tcpListener("listener/listener-a@pod-a", "1027", "backend-a.pod-a", ""). + tcpListener("listener/listener-b", "1025", "", ""). + tcpListener("multiAddress/mkl-a", "1026", "", ""). + tcpConnector("connector/normal-a@10.1.1.30", "10.1.1.30", "8084", "", ""). + tcpConnector("connector/attached-a@10.1.1.20", "10.1.1.20", "8083", "attached-a", "") + routerConfig.config.Bridges.TcpConnectors["connector/attached-a@10.1.1.20"] = qdr.TcpEndpoint{ + Name: "connector/attached-a@10.1.1.20", + SiteId: uid, + Host: "10.1.1.20", + Port: "8083", + Address: "attached-a", + ProcessID: "6ffbd287-42cc-4b71-b0a7-44f1befcc847", + } + routerConfigMap := routerConfig.asConfigMapWithOwner(site.Name, uid) + routerConfigMap.Labels = map[string]string{ + "internal.skupper.io/router-config": "", + } + networkStatus := f.skupperNetworkStatus("test", f.networkStatusInfo("mysite", "test", nil, map[string]string{"backend-a.pod-a": "10.1.1.10"}).info()) + + routerDeployment := &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{ + Name: "skupper-router", + Namespace: "test", + Labels: map[string]string{ + "skupper.io/component": "router", + "skupper.io/type": "site", + }, + }, + Spec: appsv1.DeploymentSpec{ + Selector: &metav1.LabelSelector{MatchLabels: f.routerSelector(false)}, + }, + } + clients, err := fakeclient.NewFakeClient(config.Namespace, []runtime.Object{routerConfigMap, networkStatus, routerDeployment, attachedPod, normalPod}, []runtime.Object{ + site, + siteCA, + f.connectorWithSelector("normal-a", "test", "app=normal-a", 8084), + f.attachedConnector("attached-a", "attached", "test", "app=attached-a", 8083), + attachedConnectorBinding, + f.listenerWithExposePodsByName("listener-a", "test", "backend-a", "backend-a", 8080), + f.listener("listener-b", "test", "backend-b", 8081), + f.multiKeyListener("mkl-a", "test", "backend-mkl", 8082), + }, "") + assert.Assert(t, err) + enableSSA(clients.GetDynamicClient()) + + var observed []qdr.BridgeConfig + clients.GetKubeClient().(*k8sfake.Clientset).PrependReactor("update", "configmaps", func(action k8stesting.Action) (bool, runtime.Object, error) { + update := action.(k8stesting.UpdateAction) + cm := update.GetObject().(*corev1.ConfigMap) + if cm.Namespace == "test" && cm.Name == "skupper-router" { + config, err := qdr.GetRouterConfigFromConfigMap(cm) + if err != nil { + return true, nil, err + } + observed = append(observed, config.Bridges) + } + return false, nil, nil + }) + + controller, err := NewController(clients, config) + assert.Assert(t, err) + stopCh := make(chan struct{}) + defer close(stopCh) + err = controller.init(stopCh) + assert.Assert(t, err) + + expectedTcpListeners := []string{ + "listener/listener-a", + "listener/listener-a@pod-a", + "listener/listener-b", + "multiAddress/mkl-a", + } + expectedTcpConnectors := []string{ + "connector/normal-a@10.1.1.30", + "connector/attached-a@10.1.1.20", + } + assert.Assert(t, len(observed) > 0, "expected at least one router config update after recovery") + for i, bridge := range observed { + for _, name := range expectedTcpListeners { + _, ok := bridge.TcpListeners[name] + assert.Assert(t, ok, "router config update %d was partial: missing %s", i, name) + } + for _, name := range expectedTcpConnectors { + _, ok := bridge.TcpConnectors[name] + assert.Assert(t, ok, "router config update %d was partial: missing %s", i, name) + } + } + + actual, err := clients.GetKubeClient().CoreV1().ConfigMaps("test").Get(context.Background(), "skupper-router", metav1.GetOptions{}) + assert.Assert(t, err) + configAfterRecovery, err := qdr.GetRouterConfigFromConfigMap(actual) + assert.Assert(t, err) + for _, name := range expectedTcpListeners { + _, ok := configAfterRecovery.Bridges.TcpListeners[name] + assert.Assert(t, ok, "final router config missing %s", name) + } + for _, name := range expectedTcpConnectors { + _, ok := configAfterRecovery.Bridges.TcpConnectors[name] + assert.Assert(t, ok, "final router config missing %s", name) + } +} + func TestUpdate(t *testing.T) { testTable := []struct { name string diff --git a/internal/kube/site/bindings.go b/internal/kube/site/bindings.go index 578f6142a..d0c894df6 100644 --- a/internal/kube/site/bindings.go +++ b/internal/kube/site/bindings.go @@ -28,6 +28,7 @@ type TargetSelection interface { Selector() string Close() List() []skupperv2alpha1.PodDetails + Sync(stopCh <-chan struct{}) bool } type TargetSelectionImpl struct { @@ -61,6 +62,10 @@ func (w *TargetSelectionImpl) List() []skupperv2alpha1.PodDetails { return w.watcher.pods() } +func (w *TargetSelectionImpl) Sync(stopCh <-chan struct{}) bool { + return w.watcher.Sync(stopCh) +} + func (w *TargetSelectionImpl) Updated(pods []skupperv2alpha1.PodDetails) error { err := w.site.updateRouterConfig(w.site.bindings) connector := w.site.bindings.GetConnector(w.name) @@ -127,3 +132,7 @@ func (w *PodWatcher) Close() { bindings_logger.Debug("Stopping pod watcher", w.context.Attr(), slog.String("selector", w.context.Selector())) close(w.stopCh) } + +func (w *PodWatcher) Sync(stopCh <-chan struct{}) bool { + return w.watcher.Sync(stopCh) +} diff --git a/internal/kube/site/bindings_test.go b/internal/kube/site/bindings_test.go index 8818256e0..735219d1c 100644 --- a/internal/kube/site/bindings_test.go +++ b/internal/kube/site/bindings_test.go @@ -45,12 +45,15 @@ func (m *MockBindingContext) Unexpose(host string) error { type MockTargetSelection struct { selector string podDetails []skupperv2alpha1.PodDetails + syncResult bool + syncCalled bool } func NewMockTargetSelection(selector string, podDetails []skupperv2alpha1.PodDetails) *MockTargetSelection { return &MockTargetSelection{ selector: selector, podDetails: podDetails, + syncResult: true, } } @@ -65,6 +68,36 @@ func (m *MockTargetSelection) List() []skupperv2alpha1.PodDetails { return m.podDetails } +func (m *MockTargetSelection) Sync(stopCh <-chan struct{}) bool { + m.syncCalled = true + return m.syncResult +} + +func TestExtendedBindings_SyncConnectorPods(t *testing.T) { + selector := NewMockTargetSelection("app=backend", nil) + bindings := &ExtendedBindings{ + selectors: map[string]TargetSelection{ + "backend": selector, + }, + } + + assert.Assert(t, bindings.SyncConnectorPods(make(chan struct{}))) + assert.Assert(t, selector.syncCalled) +} + +func TestExtendedBindings_SyncConnectorPodsReturnsFalse(t *testing.T) { + selector := NewMockTargetSelection("app=backend", nil) + selector.syncResult = false + bindings := &ExtendedBindings{ + selectors: map[string]TargetSelection{ + "backend": selector, + }, + } + + assert.Assert(t, !bindings.SyncConnectorPods(make(chan struct{}))) + assert.Assert(t, selector.syncCalled) +} + func TestBindingAdaptor_ConnectorUpdated(t *testing.T) { type fields struct { context BindingContext diff --git a/internal/kube/site/extended_bindings.go b/internal/kube/site/extended_bindings.go index ed95f3df9..a6a5ed3e6 100644 --- a/internal/kube/site/extended_bindings.go +++ b/internal/kube/site/extended_bindings.go @@ -394,6 +394,24 @@ func (b *ExtendedBindings) MapOverAttachedConnectors(cf AttachedConnectorFunctio } } +func (b *ExtendedBindings) SyncConnectorPods(stopCh <-chan struct{}) bool { + for _, selector := range b.selectors { + if selector != nil && !selector.Sync(stopCh) { + return false + } + } + return true +} + +func (b *ExtendedBindings) SyncAttachedConnectorPods(stopCh <-chan struct{}) bool { + for _, connector := range b.connectors { + if connector.watcher != nil && !connector.watcher.Sync(stopCh) { + return false + } + } + return true +} + func (b *ExtendedBindings) Apply(config *qdr.RouterConfig) bool { var updated bool desired := b.bindings.ToBridgeConfig() diff --git a/internal/kube/site/site.go b/internal/kube/site/site.go index 3e855ede4..837420e32 100644 --- a/internal/kube/site/site.go +++ b/internal/kube/site/site.go @@ -47,6 +47,7 @@ type Labelling interface { type Site struct { initialised bool + inRecovery bool site *skupperv2alpha1.Site name string namespace string @@ -131,7 +132,24 @@ func (s *Site) StartRecovery(site *skupperv2alpha1.Site) error { if err := s.migrateRouterDeploymentSelectors(context.TODO(), site); err != nil { return err } - return s.reconcile(site, true) + if err := s.reconcile(site, true); err != nil { + return err + } + s.inRecovery = true + return nil +} + +func (s *Site) FinishRecovery(siteDef *skupperv2alpha1.Site) error { + if !s.inRecovery { + return s.Reconcile(siteDef) + } + + err := s.Reconcile(siteDef) + s.inRecovery = false + if err != nil { + return err + } + return s.updateRecoveredRouterConfig() } // migrateRouterDeploymentSelectors deletes router Deployments whose @@ -624,6 +642,14 @@ func (s *Site) WatchPods(context PodWatchingContext, namespace string) *PodWatch return w } +func (s *Site) SyncConnectorPods(stopCh <-chan struct{}) bool { + return s.bindings.SyncConnectorPods(stopCh) +} + +func (s *Site) SyncAttachedConnectorPods(stopCh <-chan struct{}) bool { + return s.bindings.SyncAttachedConnectorPods(stopCh) +} + func (s *Site) Expose(exposed *ExposedPortSet) error { ctxt := context.TODO() current, err := s.clients.GetKubeClient().CoreV1().Services(s.namespace).Get(ctxt, exposed.Host, metav1.GetOptions{}) @@ -1009,7 +1035,7 @@ func (s *Site) updateRouterConfig(update qdr.ConfigUpdate) error { } func (s *Site) updateRouterConfigForGroup(update qdr.ConfigUpdate, group string) error { - if !s.initialised { + if !s.initialised || s.inRecovery { return nil } if err := kubeqdr.UpdateRouterConfig(s.clients.GetKubeClient(), group, s.namespace, context.TODO(), update, s.labelling); err != nil { @@ -1018,6 +1044,18 @@ func (s *Site) updateRouterConfigForGroup(update qdr.ConfigUpdate, group string) return nil } +func (s *Site) updateRecoveredRouterConfig() error { + groups := s.groups() + for i, group := range groups { + linkConfig := s.linkAccess.DesiredConfig(groups[:i], SSL_PROFILE_PATH) + update := ConfigUpdateList{s.bindings, s, linkConfig} + if err := s.updateRouterConfigForGroup(update, group); err != nil { + return err + } + } + return nil +} + func (s *Site) updateConnectorStatus(connector *skupperv2alpha1.Connector) error { updated, err := updateConnectorStatus(s.clients, connector) if err != nil {