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 {