Skip to content

Recover Full Site Before Reconfiguring Rotuer #2519

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
c-kruse wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Recover Full Site Before Reconfiguring Rotuer #2519

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
147 changes: 115 additions & 32 deletions internal/kube/controller/controller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ import (
"log/slog"
"os"
"regexp"
"time"

appsv1 "k8s.io/api/apps/v1"
corev1 "k8s.io/api/core/v1"
Expand All @@ -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 {
Expand Down Expand Up @@ -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))
Expand Down Expand Up @@ -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
Expand All @@ -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() {
Expand Down Expand Up @@ -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
}
Comment on lines +350 to +353

@coderabbitai coderabbitai Bot Jul 2, 2026 •
edited
Loading

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎯 Functional Correctness | 🟠 Major | ⚡ Quick win

Filter attached connector recovery by target site namespace.

checkAttachedConnector applies the object to connector.Spec.SiteNamespace, so this recovery filter can skip externally-namespaced attached connectors that target a controlled site. Use the target site namespace for the control check.

🐛 Proposed fix 
 	for _, connector := range c.attachedConnectorWatcher.List() {
-		if !c.namespaces.isControlled(connector.Namespace) {
+		if !c.namespaces.isControlled(connector.Spec.SiteNamespace) {
 			continue
 		}
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
for _, connector := range c.attachedConnectorWatcher.List() {
if !c.namespaces.isControlled(connector.Namespace) {
continue
}
for _, connector := range c.attachedConnectorWatcher.List() {
if !c.namespaces.isControlled(connector.Spec.SiteNamespace) {
continue
}

@AryanP123 AryanP123 Jul 7, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

TestRecoveryPreservesRouterBridgeConfig might not catch that because default test config has requireExplicitControl=false, so isControlled("attached") still passes I believe.

@coderabbitai coderabbitai Bot Jul 7, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
rg -n "requireExplicitControl|func.*isControlled|type NamespaceConfig" internal/kube/controller/*.go

Length of output: 1127

🏁 Script executed:

#!/bin/bash
sed -n '1,80p' internal/kube/controller/namespaces.go
echo ---
sed -n '1,50p' internal/kube/controller/config.go

Length of output: 3644

@AryanP123, you're right — with the default test config, Config.requireExplicitControl() evaluates to false (since WatchingAllNamespaces() is likely true and RequireExplicitControl is unset), so NamespaceConfig.isControlled falls through to return true regardless of which namespace is checked. That means TestRecoveryPreservesRouterBridgeConfig wouldn't distinguish between checking connector.Namespace vs connector.Spec.SiteNamespace — it's not exercising the namespace-scoping logic at all.

The original correctness concern about using connector.Spec.SiteNamespace (to match what checkAttachedConnector actually operates on) still stands independently of test coverage. If you want to guard against regressions here, a follow-up test could set RequireExplicitControl: true (or watch a specific namespace) and create an AttachedConnector whose .Namespace differs from .Spec.SiteNamespace, then assert recovery still applies it when the site namespace is controlled. Happy to help sketch that out if useful, otherwise this can be tracked as a known test gap.

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()),
)
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.
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),
Expand All @@ -341,8 +426,6 @@ func (c *Controller) init(stopCh <-chan struct{}) error {
)
}
}
c.certMgr.Recover()
c.accessRecovery.Recover()
if c.startGrantServer != nil {
c.startGrantServer()
}
Expand Down
143 changes: 143 additions & 0 deletions internal/kube/controller/controller_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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
Expand Down
9 changes: 9 additions & 0 deletions internal/kube/site/bindings.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@ type TargetSelection interface {
Selector() string
Close()
List() []skupperv2alpha1.PodDetails
Sync(stopCh <-chan struct{}) bool
}

type TargetSelectionImpl struct {
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
}
Loading
Loading