diff --git a/config/crd/bases/skupper_link_crd.yaml b/config/crd/bases/skupper_link_crd.yaml index 218985fe2..705995cfe 100644 --- a/config/crd/bases/skupper_link_crd.yaml +++ b/config/crd/bases/skupper_link_crd.yaml @@ -61,6 +61,14 @@ spec: description: |- The configured routing cost of sending traffic over the link. type: integer + routingKeys: + type: array + description: |- + Optional. A list of routing key identifiers available in the local VAN + that are accessible to the remote network over this link. + This is only used in inter-network connections (multi-van). + items: + type: string settings: description: |- A map containing additional settings. Each map entry has a diff --git a/config/crd/bases/skupper_router_access_crd.yaml b/config/crd/bases/skupper_router_access_crd.yaml index 2966e4454..80bb1c40f 100644 --- a/config/crd/bases/skupper_router_access_crd.yaml +++ b/config/crd/bases/skupper_router_access_crd.yaml @@ -22,20 +22,27 @@ spec: roles: description: |- The named interfaces by which a router can be accessed. These - include "inter-router" for links between interior routers and - "edge" for links from edge routers to interior routers. + include "inter-router" for links between interior routers, + "edge" for links from edge routers to interior routers, + "inter-network" for multi-van links. type: array + maxItems: 4 items: type: object properties: name: - description: The role name. Either "inter-router" or "edge". + description: The role name. One of "inter-router", "edge" or "inter-network". type: string port: description: The port for the router to bind. Must not conflict with another role. type: integer required: - name + x-kubernetes-validations: + - rule: "self.all(r, self.filter(x, x.name == r.name).size() == 1)" + message: "spec.roles must not contain duplicate role names" + - rule: "self.filter(x, has(x.port) && x.port > 0).all(r, self.filter(x, has(x.port) && x.port == r.port).size() == 1)" + message: "spec.roles must not contain duplicate non-zero ports" generateTlsCredentials: description: |- When set, Skupper generates the TLS credentials to be @@ -89,6 +96,14 @@ spec: The hostnames and IPs secured by the router TLS certificate. items: type: string + routingKeys: + type: array + description: |- + Optional. A list of routing key identifiers available in the local VAN + that are accessible to the remote network over this ingress. + This is only used in inter-network connections (multi-van). + items: + type: string settings: description: |- Advanced. A map containing additional settings. Each map @@ -162,6 +177,18 @@ spec: - reason - status - type + roles: + type: array + description: |- + List of roles and their allocated ports. + If a role does not specify a port, the controller assigns a dynamic port and reports it here. + items: + type: object + properties: + name: + type: string + port: + type: integer endpoints: type: array description: |- diff --git a/config/crd/bases/skupper_site_crd.yaml b/config/crd/bases/skupper_site_crd.yaml index b0c2d4968..860741b93 100644 --- a/config/crd/bases/skupper_site_crd.yaml +++ b/config/crd/bases/skupper_site_crd.yaml @@ -77,6 +77,13 @@ spec: there is little benefit. Currently, edge sites cannot also have HA enabled. + networkId: + type: string + description: |- + Optional. An identifier for the network this site belongs to. + When set, the router uses this value to advertise inter-network + topology addresses, enabling multi-VAN connectivity. All sites + connected to this VAN must use the same network identifier. settings: description: |- Advanced. A map containing additional settings. Each map diff --git a/config/rbac/cluster/clusterrole.yaml b/config/rbac/cluster/clusterrole.yaml index 2e1deb9be..94c1f2bad 100644 --- a/config/rbac/cluster/clusterrole.yaml +++ b/config/rbac/cluster/clusterrole.yaml @@ -129,6 +129,12 @@ rules: - create - delete - update + - apiGroups: + - apiextensions.k8s.io + resources: + - customresourcedefinitions + verbs: + - get - apiGroups: - skupper.io resources: diff --git a/internal/cmd/skupper/debug/sweeper/ports.go b/internal/cmd/skupper/debug/sweeper/ports.go index d66b7f9f0..0c97ad264 100644 --- a/internal/cmd/skupper/debug/sweeper/ports.go +++ b/internal/cmd/skupper/debug/sweeper/ports.go @@ -8,8 +8,6 @@ import ( "strconv" "strings" "text/tabwriter" - - "github.com/skupperproject/skupper/internal/ports" ) // PortStat is the number of TCP adaptor connections on one router port, split @@ -20,6 +18,8 @@ type PortStat struct { Out int } +const MaxTCPPort int = 65535 + func (p PortStat) Total() int { return p.In + p.Out } // ListPorts summarizes the router's TCP adaptor connections by port, @@ -59,8 +59,8 @@ func FilterByPorts(conns []connInfo, portList []int) []connInfo { func ValidatePorts(portList []int) error { var portErrors []error for _, p := range portList { - if p < 1 || p > ports.MAX_PORT { - portErrors = append(portErrors, fmt.Errorf("port is not valid: %d is not between 1 and %d", p, ports.MAX_PORT)) + if p < 1 || p > MaxTCPPort { + portErrors = append(portErrors, fmt.Errorf("port is not valid: %d is not between 1 and %d", p, MaxTCPPort)) } } return errors.Join(portErrors...) diff --git a/internal/cmd/skupper/debug/sweeper/ports_test.go b/internal/cmd/skupper/debug/sweeper/ports_test.go index 060785d11..bcf1eb442 100644 --- a/internal/cmd/skupper/debug/sweeper/ports_test.go +++ b/internal/cmd/skupper/debug/sweeper/ports_test.go @@ -4,6 +4,8 @@ import ( "bytes" "reflect" "testing" + + "github.com/skupperproject/skupper/internal/ports" ) func TestSummarizePorts(t *testing.T) { @@ -115,7 +117,7 @@ func TestFilterByPorts(t *testing.T) { } func TestValidatePorts(t *testing.T) { - if err := ValidatePorts([]int{1, 8080, 65535}); err != nil { + if err := ValidatePorts([]int{1, 8080, ports.MAX_PORT}); err != nil { t.Errorf("ValidatePorts() rejected valid ports: %v", err) } for _, invalid := range [][]int{{0}, {-1}, {65536}, {8080, 70000}} { diff --git a/internal/cmd/skupper/link/kube/link_generate.go b/internal/cmd/skupper/link/kube/link_generate.go index 80f0dfc8b..426587dae 100644 --- a/internal/cmd/skupper/link/kube/link_generate.go +++ b/internal/cmd/skupper/link/kube/link_generate.go @@ -7,6 +7,7 @@ import ( "context" "errors" "fmt" + "slices" "strconv" "strings" "time" @@ -313,6 +314,10 @@ func getEndpointsByGroups(endpointList []v2alpha1.Endpoint) map[string][]v2alpha endpointGroup := make(map[string][]v2alpha1.Endpoint) for _, endpoint := range endpointList { + // ignore endpoints not used for site linking + if !slices.Contains([]string{"inter-router", "edge"}, endpoint.Name) { + continue + } if len(endpointGroup[endpoint.Group]) > 0 { endpointGroup[endpoint.Group] = append(endpointGroup[endpoint.Group], endpoint) } else { diff --git a/internal/fixtures/skupper_resources.go b/internal/fixtures/skupper_resources.go index 39407ae2b..4ccea2e6d 100644 --- a/internal/fixtures/skupper_resources.go +++ b/internal/fixtures/skupper_resources.go @@ -176,3 +176,16 @@ func SecuredAccess(name string, namespace string) *skupperv2alpha1.SecuredAccess }, } } + +func Link(name, namespace string) *skupperv2alpha1.Link { + return &skupperv2alpha1.Link{ + TypeMeta: metav1.TypeMeta{ + APIVersion: "skupper.io/v2alpha1", + Kind: "Link", + }, + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: namespace, + }, + } +} diff --git a/internal/kube/adaptor/config_sync.go b/internal/kube/adaptor/config_sync.go index 0b6d268e7..888fd4d2d 100644 --- a/internal/kube/adaptor/config_sync.go +++ b/internal/kube/adaptor/config_sync.go @@ -97,6 +97,9 @@ func (c *ConfigSync) configEvent(key string, configmap *corev1.ConfigMap) error if err != nil { return err } + if err := qdr.SyncNetwork(c.agentPool, desired.Network); err != nil { + return err + } if err := c.syncSslProfileCredentialsToDisk(desired.SslProfiles); err != nil { return err } diff --git a/internal/kube/client/client.go b/internal/kube/client/client.go index 2d4e655ca..b24f7733a 100644 --- a/internal/kube/client/client.go +++ b/internal/kube/client/client.go @@ -11,6 +11,7 @@ import ( openshiftroute "github.com/openshift/client-go/route/clientset/versioned" routev1client "github.com/openshift/client-go/route/clientset/versioned/typed/route/v1" + crdClient "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset" "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/runtime/serializer" "k8s.io/client-go/discovery" @@ -68,6 +69,7 @@ type Clients interface { GetRouteInterface() openshiftroute.Interface GetRouteClient() routev1client.RouteV1Interface GetSkupperClient() skupperclient.Interface + GetCrdClient() crdClient.Interface } // A Kube Client manages orchestration and communications with the network components @@ -81,6 +83,7 @@ type KubeClient struct { Dynamic dynamic.Interface Discovery discovery.DiscoveryInterface Skupper skupperclient.Interface + CrdClient crdClient.Interface } func (c *KubeClient) GetNamespace() string { @@ -111,6 +114,10 @@ func (c *KubeClient) GetSkupperClient() skupperclient.Interface { return c.Skupper } +func (c *KubeClient) GetCrdClient() crdClient.Interface { + return c.CrdClient +} + func NewClient(namespace string, context string, kubeConfigPath string) (*KubeClient, error) { loadingRules := clientcmd.NewDefaultClientConfigLoadingRules() if kubeConfigPath != "" { @@ -185,6 +192,10 @@ func NewClientFromRestConfig(restconfig *restclient.Config, namespace string) (* if err != nil { return nil, err } + c.CrdClient, err = crdClient.NewForConfig(cfg) + if err != nil { + return nil, err + } return c, nil } diff --git a/internal/kube/client/crds.go b/internal/kube/client/crds.go new file mode 100644 index 000000000..4b52be2d7 --- /dev/null +++ b/internal/kube/client/crds.go @@ -0,0 +1,49 @@ +package client + +import ( + "context" + "strings" + + apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1" + "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset" + v1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +func IsCrdAvailable(client clientset.Interface, name string) bool { + crd, err := getCrd(client, name) + return err == nil && crd != nil +} + +func IsCrdPathAvailable(client clientset.Interface, name string, path string) bool { + crd, err := getCrd(client, name) + if err != nil { + return false + } + paths := strings.Split(path, ".") + for _, crdVersion := range crd.Spec.Versions { + if crdVersion.Schema == nil || crdVersion.Schema.OpenAPIV3Schema == nil { + continue + } + schema := crdVersion.Schema.OpenAPIV3Schema + properties := schema.Properties + found := true + for _, item := range paths { + if prop, ok := properties[item]; !ok { + found = false + break + } else { + properties = prop.Properties + } + } + if found { + return true + } + } + return false +} + +func getCrd(client clientset.Interface, name string) (*apiextensionsv1.CustomResourceDefinition, error) { + crd, err := client.ApiextensionsV1().CustomResourceDefinitions().Get( + context.Background(), name, v1.GetOptions{}) + return crd, err +} diff --git a/internal/kube/client/crds_test.go b/internal/kube/client/crds_test.go new file mode 100644 index 000000000..00c6a668b --- /dev/null +++ b/internal/kube/client/crds_test.go @@ -0,0 +1,143 @@ +package client + +import ( + "testing" + + apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1" + "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset" + fakeCrdClient "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset/fake" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" +) + +// routerAccessCRD returns a minimal RouterAccess CRD whose OpenAPI schema +// contains a "status" property with a nested "roles" property, mirroring the +// real routeraccesses.skupper.io shape used by IsCrdPathAvailable. +func routerAccessCRD(allocatedPort bool) *apiextensionsv1.CustomResourceDefinition { + str := apiextensionsv1.JSONSchemaProps{Type: "string"} + rolesProp := apiextensionsv1.JSONSchemaProps{ + Type: "array", + Items: &apiextensionsv1.JSONSchemaPropsOrArray{ + Schema: &str, + }, + } + statusProp := apiextensionsv1.JSONSchemaProps{ + Type: "object", + Properties: map[string]apiextensionsv1.JSONSchemaProps{}, + } + if allocatedPort { + statusProp.Properties["roles"] = rolesProp + } + schema := &apiextensionsv1.JSONSchemaProps{ + Type: "object", + Properties: map[string]apiextensionsv1.JSONSchemaProps{ + "status": statusProp, + }, + } + return &apiextensionsv1.CustomResourceDefinition{ + ObjectMeta: metav1.ObjectMeta{ + Name: "routeraccesses.skupper.io", + }, + Spec: apiextensionsv1.CustomResourceDefinitionSpec{ + Versions: []apiextensionsv1.CustomResourceDefinitionVersion{ + { + Name: "v2alpha1", + Schema: &apiextensionsv1.CustomResourceValidation{ + OpenAPIV3Schema: schema, + }, + }, + }, + }, + } +} + +func newFakeCrdClient(objects ...runtime.Object) clientset.Interface { + return fakeCrdClient.NewSimpleClientset(objects...) +} + +func TestIsCrdAvailable(t *testing.T) { + crdWithRoles := routerAccessCRD(true) + + tests := []struct { + name string + client clientset.Interface + crdName string + want bool + }{ + { + name: "CRD not registered", + client: newFakeCrdClient(), + crdName: "routeraccesses.skupper.io", + want: false, + }, + { + name: "CRD registered", + client: newFakeCrdClient(crdWithRoles), + crdName: "routeraccesses.skupper.io", + want: true, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := IsCrdAvailable(tt.client, tt.crdName); got != tt.want { + t.Errorf("IsCrdAvailable() = %v, want %v", got, tt.want) + } + }) + } +} + +func TestIsCrdPathAvailable(t *testing.T) { + crdWithRoles := routerAccessCRD(true) + crdWithoutRoles := routerAccessCRD(false) + + tests := []struct { + name string + client clientset.Interface + crdName string + path string + want bool + }{ + { + name: "CRD not registered", + client: newFakeCrdClient(), + crdName: "routeraccesses.skupper.io", + path: "status.roles", + want: false, + }, + { + name: "path does not exist in schema", + client: newFakeCrdClient(crdWithRoles), + crdName: "routeraccesses.skupper.io", + path: "spec.nonexistent", + want: false, + }, + { + name: "first segment exists but second is missing", + client: newFakeCrdClient(crdWithRoles), + crdName: "routeraccesses.skupper.io", + path: "status.missing", + want: false, + }, + { + name: "nested path status.roles does not exists", + client: newFakeCrdClient(crdWithoutRoles), + crdName: "routeraccesses.skupper.io", + path: "status.roles", + want: false, + }, + { + name: "nested path status.roles exists", + client: newFakeCrdClient(crdWithRoles), + crdName: "routeraccesses.skupper.io", + path: "status.roles", + want: true, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := IsCrdPathAvailable(tt.client, tt.crdName, tt.path); got != tt.want { + t.Errorf("IsCrdPathAvailable() = %v, want %v", got, tt.want) + } + }) + } +} diff --git a/internal/kube/client/fake/fake_client.go b/internal/kube/client/fake/fake_client.go index ef2e3808f..af4ec9846 100644 --- a/internal/kube/client/fake/fake_client.go +++ b/internal/kube/client/fake/fake_client.go @@ -20,6 +20,7 @@ import ( "github.com/skupperproject/skupper/internal/kube/resource" skupperclientfake "github.com/skupperproject/skupper/pkg/generated/client/clientset/versioned/fake" fakeskupperv2alpha1 "github.com/skupperproject/skupper/pkg/generated/client/clientset/versioned/typed/skupper/v2alpha1/fake" + fakeCrdClient "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset/fake" testing "k8s.io/client-go/testing" ) @@ -68,7 +69,7 @@ func NewFakeClient(namespace string, k8sObjects []runtime.Object, skupperObjects fakeDiscoveryClient.Resources = append(fakeDiscoveryClient.Resources, fakedApiResources()...) } c.Route = routefake.NewSimpleClientset(routes...) - + c.CrdClient = fakeCrdClient.NewClientset() return c, nil } func gvrFromGvk(gvk schema.GroupVersionKind) (schema.GroupVersionResource, bool) { diff --git a/internal/kube/controller/controller.go b/internal/kube/controller/controller.go index 9967aa4b6..131560624 100644 --- a/internal/kube/controller/controller.go +++ b/internal/kube/controller/controller.go @@ -57,6 +57,7 @@ type Controller struct { labellingWatcher *watchers.ConfigMapWatcher attachableConnectors map[string]*skupperv2alpha1.AttachedConnector disableSecContext bool + dynamicPortsInRouterAccess bool log *slog.Logger namespaces *NamespaceConfig observedServices map[string]string @@ -99,14 +100,15 @@ func labelling() internalinterfaces.TweakListOptionsFunc { func NewController(cli internalclient.Clients, config *Config, options ...watchers.EventProcessorCustomizer) (*Controller, error) { controller := &Controller{ - eventProcessor: watchers.NewEventProcessor("Controller", cli, options...), - sites: map[string]*site.Site{}, - siteSizing: sizing.NewRegistry(), - labelling: labels.NewLabelsAndAnnotations(config.Namespace), - attachableConnectors: map[string]*skupperv2alpha1.AttachedConnector{}, - log: slog.New(slog.Default().Handler()).With(slog.String("component", "kube.controller")), - observedServices: map[string]string{}, - disableSecContext: config.DisableSecurityContext, + eventProcessor: watchers.NewEventProcessor("Controller", cli, options...), + sites: map[string]*site.Site{}, + siteSizing: sizing.NewRegistry(), + labelling: labels.NewLabelsAndAnnotations(config.Namespace), + attachableConnectors: map[string]*skupperv2alpha1.AttachedConnector{}, + log: slog.New(slog.Default().Handler()).With(slog.String("component", "kube.controller")), + observedServices: map[string]string{}, + dynamicPortsInRouterAccess: internalclient.IsCrdPathAvailable(cli.GetCrdClient(), "routeraccesses.skupper.io", "status.roles"), + disableSecContext: config.DisableSecurityContext, } hostname := os.Getenv("HOSTNAME") @@ -465,7 +467,7 @@ func (c *Controller) getSite(namespace string) *site.Site { if existing, ok := c.sites[namespace]; ok { return existing } - site := site.NewSite(namespace, c.eventProcessor, c.certMgr, c.accessMgr, c.siteSizing, c, c.disableSecContext) + site := site.NewSite(namespace, c.eventProcessor, c.certMgr, c.accessMgr, c.siteSizing, c, c.dynamicPortsInRouterAccess, c.disableSecContext) c.sites[namespace] = site return site } diff --git a/internal/kube/grants/tokens.go b/internal/kube/grants/tokens.go index e4ca6c692..eb40501aa 100644 --- a/internal/kube/grants/tokens.go +++ b/internal/kube/grants/tokens.go @@ -6,6 +6,7 @@ import ( "fmt" "io" "log/slog" + "slices" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -74,6 +75,9 @@ func (g *TokenGenerator) setValidHostsFromSite(site *skupperv2alpha1.Site) error } var hosts []string for _, endpoint := range site.Status.Endpoints { + if !slices.Contains([]string{"inter-router", "edge"}, endpoint.Name) { + continue + } hosts = append(hosts, endpoint.Host) } if len(hosts) == 0 { diff --git a/internal/kube/site/extended_bindings.go b/internal/kube/site/extended_bindings.go index 66c97866e..e33f55c75 100644 --- a/internal/kube/site/extended_bindings.go +++ b/internal/kube/site/extended_bindings.go @@ -8,6 +8,7 @@ import ( corev1 "k8s.io/api/core/v1" "github.com/skupperproject/skupper/internal/kube/watchers" + "github.com/skupperproject/skupper/internal/ports" "github.com/skupperproject/skupper/internal/qdr" "github.com/skupperproject/skupper/internal/site" skupperv2alpha1 "github.com/skupperproject/skupper/pkg/apis/skupper/v2alpha1" @@ -73,6 +74,14 @@ func (a *ExtendedBindings) cleanup() { } } +func (a *ExtendedBindings) GetPool() *ports.FreePorts { + return a.mapping.Pool +} + +func (a *ExtendedBindings) GetAllocatedPorts() map[int]string { + return a.mapping.GetAllocatedPorts() +} + func (a *ExtendedBindings) ConnectorUpdated(connector *skupperv2alpha1.Connector) bool { if selector, ok := a.selectors[connector.Name]; ok { if selector.Selector() == connector.Spec.Selector { diff --git a/internal/kube/site/site.go b/internal/kube/site/site.go index 7f650b630..8107e05ce 100644 --- a/internal/kube/site/site.go +++ b/internal/kube/site/site.go @@ -10,6 +10,7 @@ import ( "strings" internalnetwork "github.com/skupperproject/skupper/internal/network" + "github.com/skupperproject/skupper/internal/ports" appsv1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" rbacv1 "k8s.io/api/rbac/v1" @@ -47,29 +48,30 @@ type Labelling interface { } type Site struct { - initialised bool - site *skupperv2alpha1.Site - name string - namespace string - clients *watchers.EventProcessor - bindings *ExtendedBindings - links map[string]*site.Link - errors map[string]string - linkAccess site.RouterAccessMap - certs certificates.CertificateManager - access SecuredAccessFactory - accessMapping securedAccessMap - sizes *sizing.Registry - routerPods map[string]*corev1.Pod - logger *slog.Logger - currentGroups []string - labelling Labelling - profiles *secrets.ProfilesWatcher - disableSecCtx bool - leadListeners map[string]string -} - -func NewSite(namespace string, eventProcessor *watchers.EventProcessor, certs certificates.CertificateManager, access SecuredAccessFactory, sizes *sizing.Registry, labelling Labelling, disableSecCtx bool) *Site { + initialised bool + site *skupperv2alpha1.Site + name string + namespace string + clients *watchers.EventProcessor + bindings *ExtendedBindings + links map[string]*site.Link + errors map[string]string + linkAccess site.RouterAccessMap + certs certificates.CertificateManager + access SecuredAccessFactory + accessMapping securedAccessMap + sizes *sizing.Registry + routerPods map[string]*corev1.Pod + logger *slog.Logger + currentGroups []string + labelling Labelling + profiles *secrets.ProfilesWatcher + disableSecCtx bool + dynamicIngressPorts bool + leadListeners map[string]string +} + +func NewSite(namespace string, eventProcessor *watchers.EventProcessor, certs certificates.CertificateManager, access SecuredAccessFactory, sizes *sizing.Registry, labelling Labelling, dynamicIngressPorts bool, disableSecCtx bool) *Site { logger := slog.New(slog.Default().Handler()) site := &Site{ bindings: NewExtendedBindings(eventProcessor, SSL_PROFILE_PATH), @@ -85,9 +87,10 @@ func NewSite(namespace string, eventProcessor *watchers.EventProcessor, certs ce logger: logger.With( slog.String("component", "kube.site.site"), ), - labelling: labelling, - disableSecCtx: disableSecCtx, - leadListeners: map[string]string{}, + labelling: labelling, + disableSecCtx: disableSecCtx, + dynamicIngressPorts: dynamicIngressPorts, + leadListeners: map[string]string{}, } site.profiles = secrets.NewProfilesWatcher( sslSecretsWatcher(namespace, eventProcessor), @@ -196,6 +199,13 @@ func (s *Site) routerMode() qdr.Mode { } } +func (s *Site) networkId() string { + if s.site != nil { + return s.site.Spec.NetworkId + } + return "" +} + const SSL_PROFILE_PATH = "/etc/skupper-router-certs" const PROXY_PROFILE_PATH = "/etc/skupper-router-proxies" @@ -335,6 +345,7 @@ func (s *Site) initialRouterConfig() *qdr.RouterConfig { // IsNotProtectedListener to include the complete list of "protected" listeners. // rc := qdr.InitialConfig(s.name+"-${HOSTNAME}", s.site.GetSiteId(), version.Version, s.isEdge(), 3) + rc.Network.NetworkId = s.networkId() rc.AddAddress(qdr.Address{ Prefix: "mc", Distribution: "multicast", @@ -606,6 +617,22 @@ func (s *Site) Apply(config *qdr.RouterConfig) bool { updated = true config.Metadata.Mode = mode } + if networkId := s.networkId(); config.Network.NetworkId != networkId { + updated = true + config.Network.NetworkId = networkId + if networkId == "" { + for name := range qdr.FilterAutoLinks(config.AutoLinks, qdr.FilterAutoLinkExternalAddress) { + config.RemoveAutoLink(name) + } + } else { + for name := range qdr.FilterListeners(config.Listeners, qdr.IsInterVANListener) { + config.AddAutoLink(site.AutoLinkForListener(name, networkId)) + } + for name := range qdr.FilterConnectors(config.Connectors, qdr.IsInterVANConnector) { + config.AddAutoLink(site.AutoLinkForConnector(name, networkId)) + } + } + } if dcc := s.site.Spec.GetRouterDataConnectionCount(); config.Metadata.DataConnectionCount != dcc { updated = true config.Metadata.DataConnectionCount = dcc @@ -1355,7 +1382,8 @@ func (s *Site) link(linkconfig *skupperv2alpha1.Link) error { config = existing } } else { - config, err := s.newLink(linkconfig) + var err error + config, err = s.newLink(linkconfig) if err == nil { s.links[linkconfig.ObjectMeta.Name] = config } else { @@ -1364,7 +1392,11 @@ func (s *Site) link(linkconfig *skupperv2alpha1.Link) error { } if s.initialised { if config != nil { - s.logger.Info("Connecting site using token", + connectionTarget := "site" + if config.Definition().IsInterVAN() { + connectionTarget = "van" + } + s.logger.Info(fmt.Sprintf("Connecting %s using token", connectionTarget), slog.String("namespace", s.namespace), slog.String("token", linkconfig.ObjectMeta.Name)) if currentProxyProfileName != "" && prevProxyProfileName != "" && currentProxyProfileName != prevProxyProfileName { @@ -1703,8 +1735,8 @@ func asSecuredAccessSpec(routerAccess *skupperv2alpha1.RouterAccess, group strin for _, role := range routerAccess.Spec.Roles { spec.Ports = append(spec.Ports, skupperv2alpha1.SecuredAccessPort{ Name: role.Name, - Port: role.Port, - TargetPort: role.Port, + Port: int(routerAccess.GetPortForRole(role.Name)), + TargetPort: int(routerAccess.GetPortForRole(role.Name)), Protocol: "TCP", }) } @@ -1736,20 +1768,88 @@ func (s *Site) checkSecuredAccess() error { return nil } +func (s *Site) hasPortConflict(ra *skupperv2alpha1.RouterAccess) (bool, string, int) { + var usedPorts = s.bindings.GetAllocatedPorts() + for _, role := range ra.Spec.Roles { + if role.Port == 0 { + continue + } + if name, ok := usedPorts[role.Port]; ok { + return true, fmt.Sprintf("routing key: %s", name), role.Port + } + } + return s.linkAccess.HasPortConflict(ra) + +} + func (s *Site) CheckRouterAccess(name string, la *skupperv2alpha1.RouterAccess) error { + if !s.initialised { + if s.linkAccess != nil && la != nil && la.Status.StatusType != skupperv2alpha1.StatusError { + s.linkAccess[name] = la + } + return nil + } + var allocatedPorts []int32 + statusChanged := false specChanged := false if la == nil { + if existing, ok := s.linkAccess[name]; ok && existing.Status.StatusType != skupperv2alpha1.StatusError { + s.getPool().ReleaseAll(existing.GetAllocatedPorts()...) + } delete(s.linkAccess, name) specChanged = true + } else if conflicts, withName, withPort := s.hasPortConflict(la); conflicts || la.MixesDynamicAndStaticPorts() { + if la.Status.StatusType != skupperv2alpha1.StatusError { + var err error + if la.MixesDynamicAndStaticPorts() { + err = fmt.Errorf("RouterAccess %q mixes static and dynamic ports", name) + s.logger.Error("RouterAccess mixes static and dynamic ports", slog.String("name", name)) + } else { + err = fmt.Errorf("RouterAccess %q conflicts with %q on port %d", name, withName, withPort) + s.logger.Error("RouterAccess port conflicts", + slog.String("name", name), + slog.String("with", withName), + slog.Int("port", withPort), + ) + } + la.SetConfigured(err) + s.updateRouterAccessStatus(la) + } + // forces router access removal + delete(s.linkAccess, name) + specChanged = true + la = nil } else { if existing, ok := s.linkAccess[name]; ok { specChanged = !reflect.DeepEqual(existing.Spec, la.Spec) } + if unusedPorts := la.GetUnusedPorts(); len(unusedPorts) > 0 { + s.getPool().ReleaseAll(unusedPorts...) + la.ReleaseUnusedPorts(unusedPorts) + statusChanged = true + } + var err error + for _, role := range la.Spec.Roles { + port := int(la.GetPortForRole(role.Name)) + if port == 0 { + if !s.dynamicIngressPorts { + return fmt.Errorf("dynamic port allocation support is not available") + } + port, err = s.getPool().NextFreePort() + if err != nil { + s.getPool().ReleaseAll(allocatedPorts...) + return err + } + allocatedPorts = append(allocatedPorts, int32(port)) + } else { + s.getPool().InUse(port) + } + if s.dynamicIngressPorts && la.AllocatePort(role.Name, port) { + statusChanged = true + } + } s.linkAccess[name] = la } - if !s.initialised { - return nil - } var configuredErr error if la != nil { configuredErr = s.missingTlsCredentialsErr(la.Spec.TlsCredentials) @@ -1801,8 +1901,13 @@ func (s *Site) CheckRouterAccess(name string, la *skupperv2alpha1.RouterAccess) if configuredErr != nil { err = stderrors.Join(configuredErr, err) } - if la != nil && la.SetConfigured(err) { + if la != nil && (la.SetConfigured(err) || statusChanged) { if err := s.updateRouterAccessStatus(la); err != nil { + if len(allocatedPorts) > 0 { + la.ReleaseUnusedPorts(allocatedPorts) + s.getPool().ReleaseAll(allocatedPorts...) + s.linkAccess[name] = la + } return err } } @@ -1886,6 +1991,13 @@ func (s *Site) TLSPriorValidRevisions() uint64 { return revisions } +func (s *Site) getPool() *ports.FreePorts { + if s.bindings == nil { + return nil + } + return s.bindings.GetPool() +} + func podState(pod *corev1.Pod) skupperv2alpha1.ConditionState { for _, c := range pod.Status.Conditions { if c.Status == corev1.ConditionFalse { diff --git a/internal/kube/site/site_test.go b/internal/kube/site/site_test.go index f7e4ef005..59b573ca3 100644 --- a/internal/kube/site/site_test.go +++ b/internal/kube/site/site_test.go @@ -2,6 +2,7 @@ package site import ( "context" + "fmt" "log/slog" "maps" "testing" @@ -1770,3 +1771,151 @@ func Test_updateAccessTokensForDeletedSite(t *testing.T) { }) } } + +func TestSite_hasPortConflict(t *testing.T) { + tests := []struct { + name string + bindingPort int // 0 means no binding port seeded + bindingKey string // routing key for the binding listener + linkAccessName string // if non-empty, seed s.linkAccess[linkAccessName] with linkAccessPort + linkAccessPort int + ra *skupperv2alpha1.RouterAccess + wantConflict bool + wantConflictName string + wantConflictPort int + }{ + { + name: "no ports anywhere — no conflict", + ra: &skupperv2alpha1.RouterAccess{ + ObjectMeta: metav1.ObjectMeta{Name: "new-ra"}, + Spec: skupperv2alpha1.RouterAccessSpec{ + Roles: []skupperv2alpha1.RouterAccessRole{ + {Name: "inter-router", Port: 55671}, + }, + }, + }, + wantConflict: false, + }, + { + name: "RA port collides with binding", + bindingPort: 8080, + bindingKey: "backend", + ra: &skupperv2alpha1.RouterAccess{ + ObjectMeta: metav1.ObjectMeta{Name: "new-ra"}, + Spec: skupperv2alpha1.RouterAccessSpec{ + Roles: []skupperv2alpha1.RouterAccessRole{ + {Name: "inter-router", Port: 8080}, + }, + }, + }, + wantConflict: true, + wantConflictName: "routing key: backend", + wantConflictPort: 8080, + }, + { + name: "RA port=0 skipped, no conflict", + bindingPort: 8080, + bindingKey: "backend", + ra: &skupperv2alpha1.RouterAccess{ + ObjectMeta: metav1.ObjectMeta{Name: "new-ra"}, + Spec: skupperv2alpha1.RouterAccessSpec{ + Roles: []skupperv2alpha1.RouterAccessRole{ + {Name: "inter-router", Port: 0}, + }, + }, + }, + wantConflict: false, + }, + { + name: "RA port collides with existing RouterAccess", + linkAccessName: "existing", + linkAccessPort: 55671, + ra: &skupperv2alpha1.RouterAccess{ + ObjectMeta: metav1.ObjectMeta{Name: "new-ra"}, + Spec: skupperv2alpha1.RouterAccessSpec{ + Roles: []skupperv2alpha1.RouterAccessRole{ + {Name: "inter-router", Port: 55671}, + }, + }, + }, + wantConflict: true, + wantConflictName: "router access: existing", + wantConflictPort: 55671, + }, + { + name: "RA port in both binding and linkAccess — binding wins", + bindingPort: 1024, + bindingKey: "backend", + linkAccessName: "existing", + linkAccessPort: 1024, + ra: &skupperv2alpha1.RouterAccess{ + ObjectMeta: metav1.ObjectMeta{Name: "new-ra"}, + Spec: skupperv2alpha1.RouterAccessSpec{ + Roles: []skupperv2alpha1.RouterAccessRole{ + {Name: "inter-router", Port: 1024}, + }, + }, + }, + wantConflict: true, + wantConflictName: "routing key: backend", + wantConflictPort: 1024, + }, + { + name: "self-update: own RA in linkAccess — no conflict", + linkAccessName: "my-ra", + linkAccessPort: 55671, + ra: &skupperv2alpha1.RouterAccess{ + ObjectMeta: metav1.ObjectMeta{Name: "my-ra"}, + Spec: skupperv2alpha1.RouterAccessSpec{ + Roles: []skupperv2alpha1.RouterAccessRole{ + {Name: "inter-router", Port: 55671}, + }, + }, + }, + wantConflict: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + s, err := newSiteMocks("test", nil, nil, "", false) + if err != nil { + t.Fatalf("newSiteMocks: %v", err) + } + + // Seed binding ports when requested. + if tt.bindingPort != 0 { + config := qdr.InitialConfig("router", "site-1", "test", false, 3) + config.Bridges.AddTcpListener(qdr.TcpEndpoint{ + Name: qdr.TcpListenerNamePrefix + tt.bindingKey, + Port: fmt.Sprintf("%d", tt.bindingPort), + Address: tt.bindingKey, + }) + s.bindings.mapping = qdr.RecoverPortMapping(&config) + } + + // Seed linkAccess when requested. + if tt.linkAccessName != "" { + s.linkAccess[tt.linkAccessName] = &skupperv2alpha1.RouterAccess{ + ObjectMeta: metav1.ObjectMeta{Name: tt.linkAccessName}, + Spec: skupperv2alpha1.RouterAccessSpec{ + Roles: []skupperv2alpha1.RouterAccessRole{ + {Name: "inter-router", Port: tt.linkAccessPort}, + }, + }, + } + } + + gotConflict, gotName, gotPort := s.hasPortConflict(tt.ra) + if gotConflict != tt.wantConflict { + t.Errorf("conflict = %v, want %v", gotConflict, tt.wantConflict) + } + if gotName != tt.wantConflictName { + t.Errorf("conflictName = %q, want %q", gotName, tt.wantConflictName) + } + if gotPort != tt.wantConflictPort { + t.Errorf("conflictPort = %d, want %d", gotPort, tt.wantConflictPort) + } + }) + } +} diff --git a/internal/kube/watchers/watchers.go b/internal/kube/watchers/watchers.go index 76320b416..9639e4ab4 100644 --- a/internal/kube/watchers/watchers.go +++ b/internal/kube/watchers/watchers.go @@ -35,6 +35,7 @@ import ( skupperclient "github.com/skupperproject/skupper/pkg/generated/client/clientset/versioned" skupperv2alpha1interfaces "github.com/skupperproject/skupper/pkg/generated/client/informers/externalversions/internalinterfaces" skupperv2alpha1informer "github.com/skupperproject/skupper/pkg/generated/client/informers/externalversions/skupper/v2alpha1" + crdClient "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset" ) // ResourceChange is the form in which events are added to the @@ -78,6 +79,7 @@ type EventProcessor struct { dynamicClient dynamic.Interface discoveryClient discovery.DiscoveryInterface skupperClient skupperclient.Interface + crdClient crdClient.Interface metrics *metricsQueue queue workqueue.RateLimitingInterface resync time.Duration @@ -97,6 +99,7 @@ func NewEventProcessor(name string, clients internalclient.Clients, options ...E discoveryClient: clients.GetDiscoveryClient(), dynamicClient: clients.GetDynamicClient(), skupperClient: clients.GetSkupperClient(), + crdClient: clients.GetCrdClient(), queue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), name), resync: time.Minute * 5, resyncShort: time.Second * 30, @@ -173,6 +176,10 @@ func (c *EventProcessor) GetSkupperClient() skupperclient.Interface { return c.skupperClient } +func (c *EventProcessor) GetCrdClient() crdClient.Interface { + return c.crdClient +} + // Starts the event processing loop in a new go routine. func (c *EventProcessor) Start(stopCh <-chan struct{}) { go wait.Until(c.run, time.Second, stopCh) diff --git a/internal/ports/ports.go b/internal/ports/ports.go index fdcac2437..d106694cf 100644 --- a/internal/ports/ports.go +++ b/internal/ports/ports.go @@ -128,6 +128,16 @@ func (ports *FreePorts) String() string { return "[" + strings.Join(parts, ", ") + "]" } +func (ports *FreePorts) ReleaseAll(portsToRelease ...int32) bool { + var changed bool + for _, port := range portsToRelease { + if ports.Release(int(port)) { + changed = true + } + } + return changed +} + func (ports *FreePorts) Release(port int) bool { var i int for i = 0; i < len(ports.Available) && port >= (ports.Available[i].Start-1); i++ { diff --git a/internal/ports/ports_test.go b/internal/ports/ports_test.go index 028fc6730..15db7f87a 100644 --- a/internal/ports/ports_test.go +++ b/internal/ports/ports_test.go @@ -253,3 +253,27 @@ func TestMergePortRangeFailed(t *testing.T) { t.Errorf(`merge should not succeed`) } } + +func TestReleaseAll(t *testing.T) { + ports := NewFreePorts() + a, _ := ports.NextFreePort() + b, _ := ports.NextFreePort() + c, _ := ports.NextFreePort() + + if ports.String() != "[(1027-65535)]" { + t.Errorf("unexpected state before ReleaseAll: %s", ports) + } + + result := ports.ReleaseAll(int32(a), int32(b), int32(c)) + if !result { + t.Errorf("expected ReleaseAll to return true") + } + if ports.String() != "[(1024-65535)]" { + t.Errorf("expected full range after ReleaseAll, got: %s", ports) + } + + result = ports.ReleaseAll() + if result { + t.Errorf("expected ReleaseAll to return false for empty input") + } +} diff --git a/internal/qdr/amqp_mgmt.go b/internal/qdr/amqp_mgmt.go index 1ad19b779..735009bfa 100644 --- a/internal/qdr/amqp_mgmt.go +++ b/internal/qdr/amqp_mgmt.go @@ -6,12 +6,11 @@ import ( "fmt" "log/slog" "os" + "slices" "strconv" "strings" "time" - "slices" - amqp "github.com/interconnectedcloud/go-amqp" "github.com/skupperproject/skupper/api/types" "github.com/skupperproject/skupper/internal/config" @@ -1216,6 +1215,24 @@ func asConnector(record Record) Connector { } } +func asAutoLink(record Record) AutoLink { + return AutoLink{ + Name: record.AsString("name"), + Address: record.AsString("address"), + ExternalAddress: record.AsString("externalAddress"), + Direction: record.AsString("direction"), + Connection: record.AsString("connection"), + ContainerId: record.AsString("containerId"), + OperStatus: OperStatus(record.AsString("operStatus")), + } +} + +func asNetwork(record Record) Network { + return Network{ + NetworkId: record.AsString("networkId"), + } +} + func asInt32(s string) int32 { ival, _ := strconv.Atoi(s) return int32(ival) @@ -1376,6 +1393,31 @@ func (a *Agent) UpdateListenerConfig(changes *ListenerDifference) error { return nil } +func (a *Agent) GetAutoLinks() (map[string]AutoLink, error) { + results, err := a.Query("io.skupper.router.router.config.autoLink", []string{}) + if err != nil { + return nil, err + } + autoLinks := map[string]AutoLink{} + for _, record := range results { + c := asAutoLink(record) + autoLinks[c.Name] = c + } + return autoLinks, nil +} + +func (a *Agent) GetNetwork() (Network, error) { + results, err := a.Query("io.skupper.router.network", []string{}) + if err != nil { + return Network{}, err + } + var network Network + for _, record := range results { + network = asNetwork(record) + } + return network, nil +} + func (a *Agent) GetLocalListeners() (map[string]Listener, error) { results, err := a.Query("io.skupper.router.listener", []string{}) if err != nil { @@ -1544,6 +1586,29 @@ func (a *Agent) ReloadProxyProfile(name string) error { return nil } +func (a *Agent) UpdateAutoLinkConfig(changes *AutoLinkDifference) error { + for _, deleted := range changes.Deleted { + if err := a.Delete("io.skupper.router.router.config.autoLink", deleted.Name); err != nil { + return fmt.Errorf("error deleting autoLink: %s - %w", deleted.Name, err) + } + } + + for _, added := range changes.Added { + if err := a.Create("io.skupper.router.router.config.autoLink", added.Name, &added); err != nil { + return fmt.Errorf("error adding autoLink: %s - %w", added.Name, err) + } + } + return nil +} + +func (a *Agent) UpdateNetworkConfig(desired Network) error { + if err := a.Update("io.skupper.router.network", "network/0", desired); err != nil { + return fmt.Errorf("error updating network config: %w", err) + } + time.Sleep(time.Second * 2) + return nil +} + func ConnectedSitesInfo(selfId string, routers []Router) types.TransportConnectedSites { var connectedSites types.TransportConnectedSites var self *Router diff --git a/internal/qdr/port_mapping.go b/internal/qdr/port_mapping.go index 3ac375679..874da85f6 100644 --- a/internal/qdr/port_mapping.go +++ b/internal/qdr/port_mapping.go @@ -10,15 +10,23 @@ import ( type PortMapping struct { mappings map[string]int - pool *ports.FreePorts + Pool *ports.FreePorts logger *slog.Logger } +func (p *PortMapping) GetAllocatedPorts() map[int]string { + ports := map[int]string{} + for key, port := range p.mappings { + ports[port] = key + } + return ports +} + func (p *PortMapping) GetPortForKey(key string) (int, error) { if existing, ok := p.mappings[key]; ok { return existing, nil } - allocated, err := p.pool.NextFreePort() + allocated, err := p.Pool.NextFreePort() if err != nil { return 0, err } @@ -29,7 +37,7 @@ func (p *PortMapping) GetPortForKey(key string) (int, error) { func (p *PortMapping) ReleasePortForKey(key string) { if existing, ok := p.mappings[key]; ok { - p.pool.Release(existing) + p.Pool.Release(existing) delete(p.mappings, key) } } @@ -40,7 +48,7 @@ func (p *PortMapping) recovered(key string, portstr string) { p.logger.Error("Failed to convert port to int", slog.String("port", portstr), slog.Any("error", err)) return } - p.pool.InUse(port) + p.Pool.InUse(port) p.mappings[key] = port } @@ -61,12 +69,12 @@ func portMappingKey(listener TcpEndpoint) string { func RecoverPortMapping(config *RouterConfig) *PortMapping { mapping := &PortMapping{ mappings: map[string]int{}, - pool: ports.NewFreePorts(), + Pool: ports.NewFreePorts(), logger: slog.New(slog.Default().Handler()).With(slog.String("component", "qdr.portMapping")), } if config != nil { for _, listener := range config.Listeners { - mapping.pool.InUse(int(listener.Port)) + mapping.Pool.InUse(int(listener.Port)) } for _, listener := range config.Bridges.TcpListeners { diff --git a/internal/qdr/qdr.go b/internal/qdr/qdr.go index 64b4e7b02..bb20da7b9 100644 --- a/internal/qdr/qdr.go +++ b/internal/qdr/qdr.go @@ -7,6 +7,7 @@ import ( "net" path_ "path" "reflect" + "slices" "strconv" "strings" @@ -16,11 +17,13 @@ import ( type RouterConfig struct { Metadata RouterMetadata + Network Network SslProfiles map[string]SslProfile ProxyProfiles map[string]ProxyProfile Listeners map[string]Listener Connectors map[string]Connector Addresses map[string]Address + AutoLinks map[string]AutoLink LogConfig map[string]LogConfig SiteConfig *SiteConfig Bridges BridgeConfig @@ -275,6 +278,28 @@ func (r *RouterConfig) AddAddress(a Address) { r.Addresses[a.Prefix] = a } +func (r *RouterConfig) AddAutoLink(a AutoLink) bool { + if r.AutoLinks == nil { + r.AutoLinks = map[string]AutoLink{} + } + if existing, ok := r.AutoLinks[a.Name]; ok && existing == a { + return false + } + r.AutoLinks[a.Name] = a + return true +} + +func (r *RouterConfig) RemoveAutoLink(name string) bool { + if r.AutoLinks == nil { + return false + } + if _, ok := r.AutoLinks[name]; ok { + delete(r.AutoLinks, name) + return true + } + return false +} + func (r *RouterConfig) AddTcpConnector(e TcpEndpoint) { r.Bridges.AddTcpConnector(e) } @@ -430,10 +455,11 @@ func (r *RouterConfig) SetLogLevels(levels map[string]string) bool { type Role string const ( - RoleInterRouter Role = "inter-router" - RoleEdge = "edge" - RoleNormal = "normal" - RoleDefault = "" + RoleInterRouter Role = "inter-router" + RoleEdge = "edge" + RoleNormal = "normal" + RoleInterNetwork = "inter-network" + RoleDefault = "" ) func asRole(name string) Role { @@ -446,6 +472,9 @@ func asRole(name string) Role { if name == "normal" { return RoleNormal } + if name == "inter-network" { + return RoleInterNetwork + } return RoleDefault } @@ -454,6 +483,8 @@ func GetRole(name string) Role { return RoleEdge } else if name == "normal" { return RoleNormal + } else if name == "inter-network" { + return RoleInterNetwork } return RoleInterRouter } @@ -473,6 +504,25 @@ type RouterMetadata struct { Metadata string `json:"metadata,omitempty"` } +type Network struct { + NetworkId string `json:"networkId,omitempty"` +} + +func (n Network) toRecord() Record { + result := make(map[string]any) + result["name"] = "network/0" + result["networkId"] = n.NetworkId + return result +} + +func (n Network) IsSet() bool { + return n.NetworkId != "" +} + +func (n Network) Equals(other Network) bool { + return n == other +} + type SslProfile struct { Name string `json:"name,omitempty"` CertFile string `json:"certFile,omitempty"` @@ -661,6 +711,11 @@ func (c *Connector) SetMaxSessionFrames(value int) { c.MaxSessionFrames = value } +func (c *Connector) FilterAutoLinks(autoLink AutoLink) bool { + return autoLink.Name == "link/"+c.Name || + strings.HasPrefix(autoLink.Name, fmt.Sprintf("link/%s/", c.Name)) +} + type Distribution string const ( @@ -674,6 +729,127 @@ type Address struct { Distribution string `json:"distribution,omitempty"` } +type OperStatus string + +const ( + OperStatusInactive OperStatus = "inactive" + OperStatusAttaching OperStatus = "attaching" + OperStatusFailed OperStatus = "failed" + OperStatusActive OperStatus = "active" + OperStatusQuiescing OperStatus = "quiescing" + OperStatusIdle OperStatus = "idle" +) + +// AutoLink fields information: +// - Address is required when direction is out +// - Direction is required +type AutoLink struct { + Name string `json:"name,omitempty"` + Address string `json:"address,omitempty"` + ExternalAddress string `json:"externalAddress,omitempty"` + Direction string `json:"direction,omitempty"` + Connection string `json:"connection,omitempty"` + ContainerId string `json:"containerId,omitempty"` + OperStatus OperStatus `json:"operStatus,omitempty"` +} + +func (a *AutoLink) toRecord() Record { + result := make(map[string]any) + if a.Name != "" { + result["name"] = a.Name + } + if a.Address != "" { + result["address"] = a.Address + } + if a.ExternalAddress != "" { + result["externalAddress"] = a.ExternalAddress + } + if a.Direction != "" { + result["direction"] = a.Direction + } + if a.Connection != "" { + result["connection"] = a.Connection + } + if a.ContainerId != "" { + result["containerId"] = a.ContainerId + } + if a.OperStatus != "" { + result["operStatus"] = a.OperStatus + } + return result +} + +func (a *AutoLink) GetOperStatus() OperStatus { + return a.OperStatus +} + +func (a *AutoLink) Equivalent(other *AutoLink) bool { + if other == nil { + return false + } + return a.Address == other.Address && + a.ExternalAddress == other.ExternalAddress && + a.Direction == other.Direction && + a.Connection == other.Connection && + a.ContainerId == other.ContainerId +} + +type AutoLinkDifference struct { + Deleted []AutoLink + Added []AutoLink +} + +func AutoLinksDifference(actual map[string]AutoLink, desired map[string]AutoLink) *AutoLinkDifference { + result := AutoLinkDifference{} + for key, v1 := range desired { + actualValue, ok := actual[key] + if !ok { + result.Added = append(result.Added, v1) + } + + //in case the autoLink exists but has changed some of its values, it needs to be recreated again + if ok && !v1.Equivalent(&actualValue) { + result.Deleted = append(result.Deleted, v1) + result.Added = append(result.Added, v1) + } + } + for key, v1 := range actual { + _, ok := desired[key] + if !ok { + result.Deleted = append(result.Deleted, v1) + } + } + return &result +} + +func (a *AutoLinkDifference) Empty() bool { + return len(a.Deleted) == 0 && len(a.Added) == 0 +} + +type AutoLinkFilter func(AutoLink) bool + +func FilterAutoLinks(autoLinks map[string]AutoLink, fn AutoLinkFilter) map[string]AutoLink { + var res = map[string]AutoLink{} + for name, link := range autoLinks { + if fn(link) { + res[name] = link + } + } + return res +} + +func FilterAutoLinkListeners(autoLink AutoLink) bool { + return strings.HasPrefix(autoLink.Name, "routerAccess/") +} + +func FilterAutoLinkConnectors(autoLink AutoLink) bool { + return strings.HasPrefix(autoLink.Name, "link/") +} + +func FilterAutoLinkExternalAddress(autoLink AutoLink) bool { + return autoLink.ExternalAddress != "" +} + type TcpEndpoint struct { Name string `json:"name,omitempty"` Host string `json:"host,omitempty"` @@ -784,7 +960,9 @@ func RouterConfigEquals(actual, desired string) bool { func UnmarshalRouterConfig(config string) (RouterConfig, error) { result := RouterConfig{ Metadata: RouterMetadata{}, + Network: Network{}, Addresses: map[string]Address{}, + AutoLinks: map[string]AutoLink{}, SslProfiles: map[string]SslProfile{}, ProxyProfiles: map[string]ProxyProfile{}, Listeners: map[string]Listener{}, @@ -822,6 +1000,13 @@ func UnmarshalRouterConfig(config string) (RouterConfig, error) { return result, fmt.Errorf("Invalid %s element got %#v", entityType, element[1]) } result.Metadata = metadata + case "network": + network := Network{} + err = convert(element[1], &network) + if err != nil { + return result, fmt.Errorf("Invalid %s element got %#v", entityType, element[1]) + } + result.Network = network case "address": address := Address{} err = convert(element[1], &address) @@ -829,6 +1014,13 @@ func UnmarshalRouterConfig(config string) (RouterConfig, error) { return result, fmt.Errorf("Invalid %s element got %#v", entityType, element[1]) } result.Addresses[address.Prefix] = address + case "autoLink": + autoLink := AutoLink{} + err = convert(element[1], &autoLink) + if err != nil { + return result, fmt.Errorf("Invalid %s element got %#v", entityType, element[1]) + } + result.AutoLinks[autoLink.Name] = autoLink case "connector": connector := Connector{} err = convert(element[1], &connector) @@ -905,6 +1097,13 @@ func MarshalRouterConfig(config RouterConfig) (string, error) { config.Metadata, } elements = append(elements, tuple) + if config.Network.IsSet() { + tuple := []interface{}{ + "network", + config.Network, + } + elements = append(elements, tuple) + } for _, e := range config.SslProfiles { tuple := []interface{}{ "sslProfile", @@ -940,6 +1139,13 @@ func MarshalRouterConfig(config RouterConfig) (string, error) { } elements = append(elements, tuple) } + for _, e := range config.AutoLinks { + tuple := []interface{}{ + "autoLink", + e, + } + elements = append(elements, tuple) + } for _, e := range config.Bridges.TcpConnectors { tuple := []interface{}{ "tcpConnector", @@ -1023,6 +1229,8 @@ func (r *RouterConfig) UpdateConfigMap(configmap *corev1.ConfigMap) (bool, error type ListenerPredicate func(Listener) bool +type ConnectorPredicate func(Connector) bool + func IsNotProtectedListener(l Listener) bool { protectedNames := [3]string{"@9090", "amqp", "amqps"} for _, name := range protectedNames { @@ -1033,6 +1241,21 @@ func IsNotProtectedListener(l Listener) bool { return true } +func IsSupportedAndNotProtectedListener(l Listener) bool { + if !IsNotProtectedListener(l) { + return false + } + return slices.Contains([]Role{RoleInterRouter, RoleEdge, RoleNormal, RoleInterNetwork}, l.Role) +} + +func IsInterVANListener(l Listener) bool { + return l.Role == RoleInterNetwork +} + +func IsInterVANConnector(c Connector) bool { + return c.Role == RoleInterNetwork +} + func FilterListeners(in map[string]Listener, predicate ListenerPredicate) map[string]Listener { results := map[string]Listener{} for key, listener := range in { @@ -1043,6 +1266,16 @@ func FilterListeners(in map[string]Listener, predicate ListenerPredicate) map[st return results } +func FilterConnectors(in map[string]Connector, predicate ConnectorPredicate) map[string]Connector { + results := map[string]Connector{} + for key, connector := range in { + if predicate(connector) { + results[key] = connector + } + } + return results +} + func (config *RouterConfig) GetMatchingListeners(predicate ListenerPredicate) map[string]Listener { return FilterListeners(config.Listeners, predicate) } diff --git a/internal/qdr/qdr_test.go b/internal/qdr/qdr_test.go index 8d063fec2..012b0191d 100644 --- a/internal/qdr/qdr_test.go +++ b/internal/qdr/qdr_test.go @@ -805,3 +805,132 @@ func TestTcpEndpointEquivalentHttpModes(t *testing.T) { t.Errorf("expected http1 vs none to be not equivalent") } } + +func TestFilterListeners(t *testing.T) { + interNetworkListener := Listener{Name: "l-in", Role: RoleInterNetwork} + interRouterListener := Listener{Name: "l-ir", Role: RoleInterRouter} + edgeListener := Listener{Name: "l-e", Role: RoleEdge} + + tests := []struct { + name string + input map[string]Listener + predicate ListenerPredicate + wantKeys []string + }{ + { + name: "empty map", + input: map[string]Listener{}, + predicate: IsInterVANListener, + wantKeys: []string{}, + }, + { + name: "all match", + input: map[string]Listener{ + "l-in": interNetworkListener, + "l-in2": {Name: "l-in2", Role: RoleInterNetwork}, + }, + predicate: IsInterVANListener, + wantKeys: []string{"l-in", "l-in2"}, + }, + { + name: "none match", + input: map[string]Listener{ + "l-ir": interRouterListener, + "l-e": edgeListener, + }, + predicate: IsInterVANListener, + wantKeys: []string{}, + }, + { + name: "mixed - only inter-network returned", + input: map[string]Listener{ + "l-in": interNetworkListener, + "l-ir": interRouterListener, + "l-e": edgeListener, + }, + predicate: IsInterVANListener, + wantKeys: []string{"l-in"}, + }, + { + name: "not protected listener", + input: map[string]Listener{ + "amqp": {Name: "amqp", Role: RoleNormal}, + "amqps": {Name: "amqps", Role: RoleNormal}, + "@9090": {Name: "@9090", Role: RoleNormal}, + "l-ir": interRouterListener, + }, + predicate: IsNotProtectedListener, + wantKeys: []string{"l-ir"}, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := FilterListeners(tt.input, tt.predicate) + assert.Equal(t, len(got), len(tt.wantKeys)) + for _, key := range tt.wantKeys { + if _, ok := got[key]; !ok { + t.Errorf("FilterListeners() missing expected key %q", key) + } + } + }) + } +} + +func TestFilterConnectors(t *testing.T) { + interNetworkConnector := Connector{Name: "c-in", Role: RoleInterNetwork} + interRouterConnector := Connector{Name: "c-ir", Role: RoleInterRouter} + edgeConnector := Connector{Name: "c-e", Role: RoleEdge} + + tests := []struct { + name string + input map[string]Connector + predicate ConnectorPredicate + wantKeys []string + }{ + { + name: "empty map", + input: map[string]Connector{}, + predicate: IsInterVANConnector, + wantKeys: []string{}, + }, + { + name: "all match", + input: map[string]Connector{ + "c-in": interNetworkConnector, + "c-in2": {Name: "c-in2", Role: RoleInterNetwork}, + }, + predicate: IsInterVANConnector, + wantKeys: []string{"c-in", "c-in2"}, + }, + { + name: "none match", + input: map[string]Connector{ + "c-ir": interRouterConnector, + "c-e": edgeConnector, + }, + predicate: IsInterVANConnector, + wantKeys: []string{}, + }, + { + name: "mixed - only inter-network returned", + input: map[string]Connector{ + "c-in": interNetworkConnector, + "c-ir": interRouterConnector, + "c-e": edgeConnector, + }, + predicate: IsInterVANConnector, + wantKeys: []string{"c-in"}, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := FilterConnectors(tt.input, tt.predicate) + assert.Equal(t, len(got), len(tt.wantKeys)) + for _, key := range tt.wantKeys { + if _, ok := got[key]; !ok { + t.Errorf("FilterConnectors() missing expected key %q", key) + } + } + }) + } +} diff --git a/internal/qdr/sync_router_ops.go b/internal/qdr/sync_router_ops.go index 79b8fd677..283077e1f 100644 --- a/internal/qdr/sync_router_ops.go +++ b/internal/qdr/sync_router_ops.go @@ -60,6 +60,24 @@ func SyncBridgeConfig(agentPool *AgentPool, desired *BridgeConfig) error { return nil } +func SyncNetwork(agentPool *AgentPool, desired Network) error { + agent, err := agentPool.Get() + if err != nil { + return fmt.Errorf("Could not get management agent : %s", err) + } + defer agentPool.Put(agent) + actual, err := agent.GetNetwork() + if err != nil { + return fmt.Errorf("error retrieving network: %s", err) + } + if !actual.Equals(desired) { + if err = agent.UpdateNetworkConfig(desired); err != nil { + return fmt.Errorf("error updating network config: %s", err) + } + } + return nil +} + func SyncRouterConfig(agentPool *AgentPool, desired *RouterConfig, checkCertFilesExist bool) error { if err := syncConnectors(agentPool, desired, checkCertFilesExist); err != nil { return err @@ -67,6 +85,9 @@ func SyncRouterConfig(agentPool *AgentPool, desired *RouterConfig, checkCertFile if err := syncListeners(agentPool, desired); err != nil { return err } + if err := syncAutoLinks(agentPool, desired); err != nil { + return err + } return nil } @@ -111,6 +132,25 @@ func syncListeners(agentPool *AgentPool, desired *RouterConfig) error { return nil } +func syncAutoLinks(agentPool *AgentPool, desired *RouterConfig) error { + agent, err := agentPool.Get() + if err != nil { + return fmt.Errorf("Could not get management agent : %s", err) + } + defer agentPool.Put(agent) + actual, err := agent.GetAutoLinks() + if err != nil { + return fmt.Errorf("Error retrieving autoLinks: %s", err) + } + + if differences := AutoLinksDifference(actual, desired.AutoLinks); !differences.Empty() { + if err = agent.UpdateAutoLinkConfig(differences); err != nil { + return fmt.Errorf("Error syncing autoLinks: %s", err) + } + } + return nil +} + func syncBridgeConfig(agent *Agent, desired *BridgeConfig) (bool, error) { actual, err := agent.GetLocalBridgeConfig() if err != nil { diff --git a/internal/site/link.go b/internal/site/link.go index 05ae02f8c..7e574b730 100644 --- a/internal/site/link.go +++ b/internal/site/link.go @@ -1,6 +1,7 @@ package site import ( + "fmt" "reflect" "strings" @@ -42,9 +43,14 @@ func (l *Link) Apply(current *qdr.RouterConfig) bool { if l.definition == nil { return false } - role := qdr.RoleInterRouter - if current.IsEdge() { - role = qdr.RoleEdge + var role qdr.Role + if !l.definition.IsInterVAN() { + role = qdr.RoleInterRouter + if current.IsEdge() { + role = qdr.RoleEdge + } + } else { + role = qdr.RoleInterNetwork } endpoint, ok := l.definition.Spec.GetEndpointForRole(string(role)) if !ok { @@ -76,9 +82,47 @@ func (l *Link) Apply(current *qdr.RouterConfig) bool { } else if prevProxyProfileName != "" { current.RemoveProxyProfile(prevProxyProfileName) } + if l.definition.IsInterVAN() { + desired := l.desiredAutoLinks(current.Network.NetworkId) + diff := qdr.AutoLinksDifference(qdr.FilterAutoLinks(current.AutoLinks, connector.FilterAutoLinks), desired) + for _, autoLinkDel := range diff.Deleted { + current.RemoveAutoLink(autoLinkDel.Name) + } + for _, autoLinkAdd := range diff.Added { + current.AddAutoLink(autoLinkAdd) + } + } return true //TODO: optimise by indicating if no change was actually needed } +func (l *Link) desiredAutoLinks(networkId string) map[string]qdr.AutoLink { + var res = map[string]qdr.AutoLink{} + if networkId != "" { + al := AutoLinkForConnector(l.name, networkId) + res[al.Name] = al + } + for _, routingKey := range l.definition.Spec.RoutingKeys { + autoLinkName := fmt.Sprintf("link/%s/%s", l.name, routingKey) + res[autoLinkName] = qdr.AutoLink{ + Name: autoLinkName, + Address: routingKey, + Direction: qdr.DirectionIn, + Connection: l.name, + } + } + return res +} + +func AutoLinkForConnector(connectorName, networkId string) qdr.AutoLink { + autoLinkName := fmt.Sprintf("link/%s", connectorName) + return qdr.AutoLink{ + Name: autoLinkName, + ExternalAddress: "_xtopo/" + networkId, + Direction: qdr.DirectionIn, + Connection: connectorName, + } +} + func sslProfileName(link *skupperv2alpha1.Link) string { return link.Spec.TlsCredentials + "-profile" } @@ -99,6 +143,9 @@ func (m LinkMap) Apply(current *qdr.RouterConfig) bool { current.RemoveConnector(connector.Name) current.RemoveSslProfile(connector.SslProfile) current.RemoveProxyProfile(connector.ProxyProfile) + for name := range qdr.FilterAutoLinks(current.AutoLinks, connector.FilterAutoLinks) { + current.RemoveAutoLink(name) + } } } } @@ -119,8 +166,15 @@ type RemoveConnector struct { name string } +func (o *RemoveConnector) filterAutoLink(autoLink qdr.AutoLink) bool { + return autoLink.Name == fmt.Sprintf("link/%s", o.name) || + strings.HasPrefix(autoLink.Name, fmt.Sprintf("link/%s/", o.name)) +} + func (o *RemoveConnector) Apply(current *qdr.RouterConfig) bool { - if changed, connector := current.RemoveConnector(o.name); changed { + var changed bool + var connector qdr.Connector + if changed, connector = current.RemoveConnector(o.name); changed { unreferenced := current.UnreferencedSslProfiles() if _, ok := unreferenced[connector.SslProfile]; ok { current.RemoveSslProfile(connector.SslProfile) @@ -129,9 +183,13 @@ func (o *RemoveConnector) Apply(current *qdr.RouterConfig) bool { if _, ok := unreferencedProxyProfiles[connector.ProxyProfile]; ok { current.RemoveProxyProfile(connector.ProxyProfile) } - return true } - return false + for name := range qdr.FilterAutoLinks(current.AutoLinks, o.filterAutoLink) { + if current.RemoveAutoLink(name) { + changed = true + } + } + return changed } func NewRemoveConnector(name string) qdr.ConfigUpdate { diff --git a/internal/site/link_test.go b/internal/site/link_test.go index e04afab6b..49affc6bb 100644 --- a/internal/site/link_test.go +++ b/internal/site/link_test.go @@ -27,10 +27,11 @@ func TestLink_Apply(t *testing.T) { current qdr.RouterConfig } tests := []struct { - name string - fields fields - args args - want bool + name string + fields fields + args args + want bool + wantRole qdr.Role }{ { name: "no definition", @@ -116,7 +117,36 @@ func TestLink_Apply(t *testing.T) { args: args{ current: qdr.InitialConfig(id, siteId, version, true, helloAge), }, - want: true, + want: true, + wantRole: qdr.RoleEdge, + }, + { + name: "inter-network definition with endpoint", + fields: fields{ + name: "link1", + sslProfilePath: "/etc/skupper-router-certs/skupper-internal/ca.crt", + proxyConfig: ProxyConfig{}, + definition: &skupperv2alpha1.Link{ + ObjectMeta: v1.ObjectMeta{ + Name: "remote-site", + Namespace: "test", + }, + Spec: skupperv2alpha1.LinkSpec{ + Endpoints: []skupperv2alpha1.Endpoint{ + { + Name: string(qdr.RoleInterNetwork), + Host: "10.10.10.1", + Port: "35671", + }, + }, + }, + }, + }, + args: args{ + current: qdr.InitialConfig(id, siteId, version, notEdge, helloAge), + }, + want: true, + wantRole: qdr.RoleInterNetwork, }, } for _, tt := range tests { @@ -126,6 +156,9 @@ func TestLink_Apply(t *testing.T) { if got := l.Apply(&tt.args.current); got != tt.want { t.Errorf("Link.Apply() = %v, want %v", got, tt.want) } + if tt.wantRole != "" { + assert.Equal(t, tt.args.current.Connectors[tt.fields.name].Role, tt.wantRole) + } }) } } @@ -485,3 +518,113 @@ func TestRemoveConnector_Apply(t *testing.T) { }) } } + +func TestAutoLinkForConnector(t *testing.T) { + tests := []struct { + name string + connectorName string + networkId string + wantAutoLink qdr.AutoLink + }{ + { + name: "basic connector and networkId", + connectorName: "link1", + networkId: "net-a", + wantAutoLink: qdr.AutoLink{ + Name: "link/link1", + ExternalAddress: "_xtopo/net-a", + Direction: qdr.DirectionIn, + Connection: "link1", + }, + }, + { + name: "connector with different networkId", + connectorName: "my-link", + networkId: "production-network", + wantAutoLink: qdr.AutoLink{ + Name: "link/my-link", + ExternalAddress: "_xtopo/production-network", + Direction: qdr.DirectionIn, + Connection: "my-link", + }, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := AutoLinkForConnector(tt.connectorName, tt.networkId) + assert.Equal(t, got.Name, tt.wantAutoLink.Name) + assert.Equal(t, got.ExternalAddress, tt.wantAutoLink.ExternalAddress) + assert.Equal(t, got.Direction, tt.wantAutoLink.Direction) + assert.Equal(t, got.Connection, tt.wantAutoLink.Connection) + }) + } +} + +func TestLink_DesiredAutoLinks(t *testing.T) { + id := "router-1" + siteId := "site-1" + version := "v2.0" + notEdge := false + helloAge := 10 + + interNetworkEndpoint := skupperv2alpha1.Endpoint{ + Name: "inter-network", + Host: "10.0.0.1", + Port: "35671", + } + + tests := []struct { + name string + networkId string + routingKeys []string + wantAutoLinkKeys []string + }{ + { + name: "no networkId and no routingKeys", + networkId: "", + routingKeys: nil, + wantAutoLinkKeys: []string{}, + }, + { + name: "networkId set, no routingKeys", + networkId: "net-a", + routingKeys: nil, + wantAutoLinkKeys: []string{"link/link1"}, + }, + { + name: "no networkId, two routingKeys", + networkId: "", + routingKeys: []string{"key1", "key2"}, + wantAutoLinkKeys: []string{"link/link1/key1", "link/link1/key2"}, + }, + { + name: "networkId set, two routingKeys", + networkId: "net-a", + routingKeys: []string{"key1", "key2"}, + wantAutoLinkKeys: []string{"link/link1", "link/link1/key1", "link/link1/key2"}, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + config := qdr.InitialConfig(id, siteId, version, notEdge, helloAge) + config.Network.NetworkId = tt.networkId + + l := NewLink("link1", "/etc/skupper-router-certs/skupper-internal/ca.crt", &ProxyConfig{}) + l.definition = &skupperv2alpha1.Link{ + Spec: skupperv2alpha1.LinkSpec{ + Endpoints: []skupperv2alpha1.Endpoint{interNetworkEndpoint}, + RoutingKeys: tt.routingKeys, + }, + } + + l.Apply(&config) + + assert.Equal(t, len(config.AutoLinks), len(tt.wantAutoLinkKeys)) + for _, key := range tt.wantAutoLinkKeys { + if _, ok := config.AutoLinks[key]; !ok { + t.Errorf("AutoLinks missing expected key %q", key) + } + } + }) + } +} diff --git a/internal/site/routeraccess.go b/internal/site/routeraccess.go index 8b44463d5..f30c2f63d 100644 --- a/internal/site/routeraccess.go +++ b/internal/site/routeraccess.go @@ -1,6 +1,7 @@ package site import ( + "fmt" "strconv" "github.com/skupperproject/skupper/internal/qdr" @@ -18,7 +19,7 @@ func (m RouterAccessMap) desiredListeners() map[string]qdr.Listener { Name: name, Role: qdr.GetRole(role.Name), Host: ra.Spec.BindHost, - Port: role.GetPort(), + Port: ra.GetPortForRole(role.Name), SslProfile: ra.Spec.TlsCredentials, SaslMechanisms: "EXTERNAL", AuthenticatePeer: true, @@ -40,7 +41,7 @@ func (m RouterAccessMap) desiredConnectors(targetGroups []string) []qdr.Connecto Name: name, Host: group, Role: qdr.RoleInterRouter, - Port: strconv.Itoa(role.Port), + Port: strconv.Itoa(int(ra.GetPortForRole(role.Name))), SslProfile: ra.Spec.TlsCredentials, Cost: 1, } @@ -50,6 +51,25 @@ func (m RouterAccessMap) desiredConnectors(targetGroups []string) []qdr.Connecto return connectors } +func (m RouterAccessMap) desiredAutoLinks() map[string]qdr.AutoLink { + var autoLinks = map[string]qdr.AutoLink{} + for raName, ra := range m { + if ra.FindRole(qdr.RoleInterNetwork) == nil { + continue + } + for _, routingKey := range ra.Spec.RoutingKeys { + autoLinkName := fmt.Sprintf("routerAccess/%s/%s", raName, routingKey) + autoLinks[autoLinkName] = qdr.AutoLink{ + Name: autoLinkName, + Address: routingKey, + Direction: qdr.DirectionIn, + Connection: fmt.Sprintf("%s-inter-network", raName), + } + } + } + return autoLinks +} + func (m RouterAccessMap) findInterRouterRole() (*skupperv2alpha1.RouterAccessRole, *skupperv2alpha1.RouterAccess) { for _, value := range m { if role := value.FindRole("inter-router"); role != nil { @@ -59,6 +79,26 @@ func (m RouterAccessMap) findInterRouterRole() (*skupperv2alpha1.RouterAccessRol return nil, nil } +func (m RouterAccessMap) HasPortConflict(ra *skupperv2alpha1.RouterAccess) (bool, string, int) { + var usedPorts = map[int32]string{} + for _, cur := range m { + for _, curRole := range cur.Spec.Roles { + if port := cur.GetPortForRole(curRole.Name); port != 0 { + usedPorts[port] = cur.Name + } + } + } + for _, role := range ra.Spec.Roles { + if role.Port == 0 { + continue + } + if name, ok := usedPorts[int32(role.Port)]; ok && name != ra.Name { + return true, fmt.Sprintf("router access: %s", name), role.Port + } + } + return false, "", 0 +} + func (m RouterAccessMap) DesiredConfig(targetGroups []string, profilePath string) *RouterAccessConfig { return m.DesiredConfigWithAvailableCredentials(targetGroups, profilePath, nil) } @@ -79,6 +119,7 @@ func (m RouterAccessMap) DesiredConfigWithAvailableCredentials(targetGroups []st return &RouterAccessConfig{ listeners: source.desiredListeners(), connectors: source.desiredConnectors(targetGroups), + autoLinks: source.desiredAutoLinks(), profilePath: profilePath, } } @@ -86,6 +127,7 @@ func (m RouterAccessMap) DesiredConfigWithAvailableCredentials(targetGroups []st type RouterAccessConfig struct { listeners map[string]qdr.Listener connectors []qdr.Connector + autoLinks map[string]qdr.AutoLink profilePath string } @@ -100,7 +142,8 @@ func (g *RouterAccessConfig) Apply(config *qdr.RouterConfig) bool { } } for _, value := range lc.Added { - if config.AddListener(value) && config.AddSslProfile(qdr.ConfigureSslProfile(value.SslProfile, g.profilePath, true)) { + if config.AddListener(value) { + config.AddSslProfile(qdr.ConfigureSslProfile(value.SslProfile, g.profilePath, true)) changed = true } } @@ -114,5 +157,34 @@ func (g *RouterAccessConfig) Apply(config *qdr.RouterConfig) bool { config.RemoveSslProfile(name) changed = true } + // Check if networkId related autoLinks are needed for inter-network listeners + if config.Network.IsSet() { + for listenerName := range qdr.FilterListeners(config.Listeners, qdr.IsInterVANListener) { + al := AutoLinkForListener(listenerName, config.Network.NetworkId) + g.autoLinks[al.Name] = al + } + } + // Update listener related autoLinks + autoLinksDiff := qdr.AutoLinksDifference(qdr.FilterAutoLinks(config.AutoLinks, qdr.FilterAutoLinkListeners), g.autoLinks) + for _, ld := range autoLinksDiff.Deleted { + if config.RemoveAutoLink(ld.Name) { + changed = true + } + } + for _, la := range autoLinksDiff.Added { + if config.AddAutoLink(la) { + changed = true + } + } return changed } + +func AutoLinkForListener(listenerName, networkId string) qdr.AutoLink { + autoLinkName := fmt.Sprintf("routerAccess/%s", listenerName) + return qdr.AutoLink{ + Name: autoLinkName, + ExternalAddress: fmt.Sprintf("_xtopo/%s", networkId), + Direction: qdr.DirectionIn, + Connection: listenerName, + } +} diff --git a/internal/site/routeraccess_test.go b/internal/site/routeraccess_test.go index bfd0fc492..eb6d099c6 100644 --- a/internal/site/routeraccess_test.go +++ b/internal/site/routeraccess_test.go @@ -356,6 +356,7 @@ func TestRouterAccessMap_DesiredConfig(t *testing.T) { Cost: 1, }, }, + autoLinks: map[string]qdr.AutoLink{}, }, }, { @@ -399,6 +400,7 @@ func TestRouterAccessMap_DesiredConfig(t *testing.T) { }, }, connectors: nil, + autoLinks: map[string]qdr.AutoLink{}, }, }, { @@ -442,13 +444,116 @@ func TestRouterAccessMap_DesiredConfig(t *testing.T) { }, }, connectors: nil, + autoLinks: map[string]qdr.AutoLink{}, + }, + }, + { + name: "inter-network sans routingKeys", + m: map[string]*skupperv2alpha1.RouterAccess{ + "default": &skupperv2alpha1.RouterAccess{ + ObjectMeta: v1.ObjectMeta{ + Name: "my-ra", + Namespace: "test", + }, + Spec: skupperv2alpha1.RouterAccessSpec{ + AccessType: "loadbalancer", + Roles: []skupperv2alpha1.RouterAccessRole{ + { + Name: "inter-network", + Port: 35671, + }, + }, + TlsCredentials: "skupper", + BindHost: "10.10.10.1", + }, + }, + }, + args: args{ + targetGroups: []string{"my-target-group"}, + profilePath: "", + }, + want: RouterAccessConfig{ + listeners: map[string]qdr.Listener{ + "my-ra-inter-network": qdr.Listener{ + Name: "my-ra-inter-network", + Role: "inter-network", + Host: "10.10.10.1", + Port: 35671, + RouteContainer: false, + Http: false, + Cost: 0, + SslProfile: "skupper", + SaslMechanisms: "EXTERNAL", + AuthenticatePeer: true, + }, + }, + connectors: nil, + autoLinks: map[string]qdr.AutoLink{}, + }, + }, + { + name: "inter-network with routingKeys", + m: map[string]*skupperv2alpha1.RouterAccess{ + "my-ra": &skupperv2alpha1.RouterAccess{ + ObjectMeta: v1.ObjectMeta{ + Name: "my-ra", + Namespace: "test", + }, + Spec: skupperv2alpha1.RouterAccessSpec{ + AccessType: "loadbalancer", + Roles: []skupperv2alpha1.RouterAccessRole{ + { + Name: "inter-network", + Port: 35671, + }, + }, + TlsCredentials: "skupper", + BindHost: "10.10.10.1", + RoutingKeys: []string{"key1", "key2"}, + }, + }, + }, + args: args{ + targetGroups: []string{"my-target-group"}, + profilePath: "", + }, + want: RouterAccessConfig{ + listeners: map[string]qdr.Listener{ + "my-ra-inter-network": qdr.Listener{ + Name: "my-ra-inter-network", + Role: "inter-network", + Host: "10.10.10.1", + Port: 35671, + RouteContainer: false, + Http: false, + Cost: 0, + SslProfile: "skupper", + SaslMechanisms: "EXTERNAL", + AuthenticatePeer: true, + }, + }, + connectors: nil, + autoLinks: map[string]qdr.AutoLink{ + "routerAccess/my-ra/key1": qdr.AutoLink{ + Name: "routerAccess/my-ra/key1", + Address: "key1", + Direction: "in", + Connection: "my-ra-inter-network", + }, + "routerAccess/my-ra/key2": qdr.AutoLink{ + Name: "routerAccess/my-ra/key2", + Address: "key2", + Direction: "in", + Connection: "my-ra-inter-network", + }, + }, }, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { if got := tt.m.DesiredConfig(tt.args.targetGroups, tt.args.profilePath); !reflect.DeepEqual(*got, tt.want) { - t.Errorf("RouterAccessMap.DesiredConfig() = %v, want %v", *got, tt.want) + t.Errorf("RouterAccessMap.DesiredConfig() = %+v, want %+v", *got, tt.want) } }) } @@ -477,3 +582,326 @@ func TestRouterAccessMap_DesiredConfigWithAvailableCredentials(t *testing.T) { t.Fatal("expected listeners when TLS secret allowed") } } + +func TestRouterAccessMap_HasPortConflict(t *testing.T) { + tests := []struct { + name string + m RouterAccessMap + ra *skupperv2alpha1.RouterAccess + wantConflict bool + wantConflictName string + wantConflictPort int + }{ + { + name: "empty map — no conflict", + m: RouterAccessMap{}, + ra: &skupperv2alpha1.RouterAccess{ + ObjectMeta: v1.ObjectMeta{Name: "new-ra"}, + Spec: skupperv2alpha1.RouterAccessSpec{ + Roles: []skupperv2alpha1.RouterAccessRole{ + {Name: "inter-router", Port: 55671}, + }, + }, + }, + wantConflict: false, + }, + { + name: "new ra, distinct port from existing — no conflict", + m: RouterAccessMap{ + "existing": &skupperv2alpha1.RouterAccess{ + ObjectMeta: v1.ObjectMeta{Name: "existing-ra"}, + Spec: skupperv2alpha1.RouterAccessSpec{ + Roles: []skupperv2alpha1.RouterAccessRole{ + {Name: "inter-router", Port: 55671}, + }, + }, + }, + }, + ra: &skupperv2alpha1.RouterAccess{ + ObjectMeta: v1.ObjectMeta{Name: "new-ra"}, + Spec: skupperv2alpha1.RouterAccessSpec{ + Roles: []skupperv2alpha1.RouterAccessRole{ + {Name: "edge", Port: 45671}, + }, + }, + }, + wantConflict: false, + }, + { + name: "new ra, same port as existing — conflict", + m: RouterAccessMap{ + "existing": &skupperv2alpha1.RouterAccess{ + ObjectMeta: v1.ObjectMeta{Name: "existing-ra"}, + Spec: skupperv2alpha1.RouterAccessSpec{ + Roles: []skupperv2alpha1.RouterAccessRole{ + {Name: "inter-router", Port: 55671}, + }, + }, + }, + }, + ra: &skupperv2alpha1.RouterAccess{ + ObjectMeta: v1.ObjectMeta{Name: "new-ra"}, + Spec: skupperv2alpha1.RouterAccessSpec{ + Roles: []skupperv2alpha1.RouterAccessRole{ + {Name: "inter-router", Port: 55671}, + }, + }, + }, + wantConflict: true, + wantConflictName: "router access: existing-ra", + wantConflictPort: 55671, + }, + { + name: "ra already in map updating its own port — self, no conflict", + m: RouterAccessMap{ + "my-ra": &skupperv2alpha1.RouterAccess{ + ObjectMeta: v1.ObjectMeta{Name: "my-ra"}, + Spec: skupperv2alpha1.RouterAccessSpec{ + Roles: []skupperv2alpha1.RouterAccessRole{ + {Name: "inter-router", Port: 55671}, + }, + }, + }, + }, + ra: &skupperv2alpha1.RouterAccess{ + ObjectMeta: v1.ObjectMeta{Name: "my-ra"}, + Spec: skupperv2alpha1.RouterAccessSpec{ + Roles: []skupperv2alpha1.RouterAccessRole{ + {Name: "inter-router", Port: 55671}, + }, + }, + }, + wantConflict: false, + }, + { + name: "candidate role port 0 is skipped — no conflict", + m: RouterAccessMap{ + "existing": &skupperv2alpha1.RouterAccess{ + ObjectMeta: v1.ObjectMeta{Name: "existing-ra"}, + Spec: skupperv2alpha1.RouterAccessSpec{ + Roles: []skupperv2alpha1.RouterAccessRole{ + {Name: "inter-router", Port: 55671}, + }, + }, + }, + }, + ra: &skupperv2alpha1.RouterAccess{ + ObjectMeta: v1.ObjectMeta{Name: "new-ra"}, + Spec: skupperv2alpha1.RouterAccessSpec{ + Roles: []skupperv2alpha1.RouterAccessRole{ + {Name: "inter-router", Port: 0}, + }, + }, + }, + wantConflict: false, + }, + { + name: "two existing ras both with unresolved ports — port-0 key bug, no spurious conflict", + m: RouterAccessMap{ + "first": &skupperv2alpha1.RouterAccess{ + ObjectMeta: v1.ObjectMeta{Name: "first-ra"}, + Spec: skupperv2alpha1.RouterAccessSpec{ + Roles: []skupperv2alpha1.RouterAccessRole{ + {Name: "inter-router", Port: 0}, + }, + }, + }, + "second": &skupperv2alpha1.RouterAccess{ + ObjectMeta: v1.ObjectMeta{Name: "second-ra"}, + Spec: skupperv2alpha1.RouterAccessSpec{ + Roles: []skupperv2alpha1.RouterAccessRole{ + {Name: "edge", Port: 0}, + }, + }, + }, + }, + ra: &skupperv2alpha1.RouterAccess{ + ObjectMeta: v1.ObjectMeta{Name: "new-ra"}, + Spec: skupperv2alpha1.RouterAccessSpec{ + Roles: []skupperv2alpha1.RouterAccessRole{ + {Name: "inter-router", Port: 55671}, + }, + }, + }, + wantConflict: false, + }, + { + name: "existing ra has allocated port in Status — candidate wants same port — conflict", + m: RouterAccessMap{ + "existing": &skupperv2alpha1.RouterAccess{ + ObjectMeta: v1.ObjectMeta{Name: "existing-ra"}, + Spec: skupperv2alpha1.RouterAccessSpec{ + Roles: []skupperv2alpha1.RouterAccessRole{ + {Name: "inter-router", Port: 0}, + }, + }, + Status: skupperv2alpha1.RouterAccessStatus{ + Roles: []skupperv2alpha1.RouterAccessRole{ + {Name: "inter-router", Port: 55671}, + }, + }, + }, + }, + ra: &skupperv2alpha1.RouterAccess{ + ObjectMeta: v1.ObjectMeta{Name: "new-ra"}, + Spec: skupperv2alpha1.RouterAccessSpec{ + Roles: []skupperv2alpha1.RouterAccessRole{ + {Name: "inter-router", Port: 55671}, + }, + }, + }, + wantConflict: true, + wantConflictName: "router access: existing-ra", + wantConflictPort: 55671, + }, + { + name: "multiple existing ras — conflict with second one", + m: RouterAccessMap{ + "first": &skupperv2alpha1.RouterAccess{ + ObjectMeta: v1.ObjectMeta{Name: "first-ra"}, + Spec: skupperv2alpha1.RouterAccessSpec{ + Roles: []skupperv2alpha1.RouterAccessRole{ + {Name: "inter-router", Port: 55671}, + }, + }, + }, + "second": &skupperv2alpha1.RouterAccess{ + ObjectMeta: v1.ObjectMeta{Name: "second-ra"}, + Spec: skupperv2alpha1.RouterAccessSpec{ + Roles: []skupperv2alpha1.RouterAccessRole{ + {Name: "edge", Port: 45671}, + }, + }, + }, + }, + ra: &skupperv2alpha1.RouterAccess{ + ObjectMeta: v1.ObjectMeta{Name: "new-ra"}, + Spec: skupperv2alpha1.RouterAccessSpec{ + Roles: []skupperv2alpha1.RouterAccessRole{ + {Name: "edge", Port: 45671}, + }, + }, + }, + wantConflict: true, + wantConflictName: "router access: second-ra", + wantConflictPort: 45671, + }, + { + name: "candidate with two roles — only second conflicts", + m: RouterAccessMap{ + "existing": &skupperv2alpha1.RouterAccess{ + ObjectMeta: v1.ObjectMeta{Name: "existing-ra"}, + Spec: skupperv2alpha1.RouterAccessSpec{ + Roles: []skupperv2alpha1.RouterAccessRole{ + {Name: "inter-router", Port: 55671}, + }, + }, + }, + }, + ra: &skupperv2alpha1.RouterAccess{ + ObjectMeta: v1.ObjectMeta{Name: "new-ra"}, + Spec: skupperv2alpha1.RouterAccessSpec{ + Roles: []skupperv2alpha1.RouterAccessRole{ + {Name: "edge", Port: 45671}, + {Name: "inter-router", Port: 55671}, + }, + }, + }, + wantConflict: true, + wantConflictName: "router access: existing-ra", + wantConflictPort: 55671, + }, + { + name: "candidate with two roles — neither conflicts", + m: RouterAccessMap{ + "existing": &skupperv2alpha1.RouterAccess{ + ObjectMeta: v1.ObjectMeta{Name: "existing-ra"}, + Spec: skupperv2alpha1.RouterAccessSpec{ + Roles: []skupperv2alpha1.RouterAccessRole{ + {Name: "inter-router", Port: 55671}, + }, + }, + }, + }, + ra: &skupperv2alpha1.RouterAccess{ + ObjectMeta: v1.ObjectMeta{Name: "new-ra"}, + Spec: skupperv2alpha1.RouterAccessSpec{ + Roles: []skupperv2alpha1.RouterAccessRole{ + {Name: "edge", Port: 45671}, + {Name: "inter-network", Port: 35671}, + }, + }, + }, + wantConflict: false, + }, + { + name: "candidate with modified port — no conflicts", + m: RouterAccessMap{ + "my-ra": &skupperv2alpha1.RouterAccess{ + ObjectMeta: v1.ObjectMeta{Name: "my-ra"}, + Spec: skupperv2alpha1.RouterAccessSpec{ + Roles: []skupperv2alpha1.RouterAccessRole{ + {Name: "inter-router", Port: 55671}, + }, + }, + }, + }, + ra: &skupperv2alpha1.RouterAccess{ + ObjectMeta: v1.ObjectMeta{Name: "my-ra"}, + Spec: skupperv2alpha1.RouterAccessSpec{ + Roles: []skupperv2alpha1.RouterAccessRole{ + {Name: "inter-network", Port: 55673}, + }, + }, + }, + wantConflict: false, + }, + { + name: "candidate conflicts when modifying port", + m: RouterAccessMap{ + "my-ra": &skupperv2alpha1.RouterAccess{ + ObjectMeta: v1.ObjectMeta{Name: "my-ra"}, + Spec: skupperv2alpha1.RouterAccessSpec{ + Roles: []skupperv2alpha1.RouterAccessRole{ + {Name: "inter-router", Port: 55671}, + }, + }, + }, + "other-ra": &skupperv2alpha1.RouterAccess{ + ObjectMeta: v1.ObjectMeta{Name: "other-ra"}, + Spec: skupperv2alpha1.RouterAccessSpec{ + Roles: []skupperv2alpha1.RouterAccessRole{ + {Name: "inter-router", Port: 55673}, + }, + }, + }, + }, + ra: &skupperv2alpha1.RouterAccess{ + ObjectMeta: v1.ObjectMeta{Name: "my-ra"}, + Spec: skupperv2alpha1.RouterAccessSpec{ + Roles: []skupperv2alpha1.RouterAccessRole{ + {Name: "inter-network", Port: 55673}, + }, + }, + }, + wantConflict: true, + wantConflictName: "router access: other-ra", + wantConflictPort: 55673, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + gotConflict, gotName, gotPort := tt.m.HasPortConflict(tt.ra) + if gotConflict != tt.wantConflict { + t.Errorf("HasPortConflict() conflict = %v, want %v", gotConflict, tt.wantConflict) + } + if gotName != tt.wantConflictName { + t.Errorf("HasPortConflict() conflicting name = %q, want %q", gotName, tt.wantConflictName) + } + if gotPort != tt.wantConflictPort { + t.Errorf("HasPortConflict() conflicting port = %v, want %v", gotPort, tt.wantConflictPort) + } + }) + } +} diff --git a/pkg/apis/skupper/v2alpha1/link_types_test.go b/pkg/apis/skupper/v2alpha1/link_types_test.go new file mode 100644 index 000000000..3b59e0a84 --- /dev/null +++ b/pkg/apis/skupper/v2alpha1/link_types_test.go @@ -0,0 +1,60 @@ +package v2alpha1 + +import ( + "testing" +) + +func TestLink_IsInterVAN(t *testing.T) { + tests := []struct { + name string + endpoints []Endpoint + want bool + }{ + { + name: "no endpoints", + endpoints: nil, + want: false, + }, + { + name: "inter-router endpoint only", + endpoints: []Endpoint{ + {Name: "inter-router", Host: "10.0.0.1", Port: "55671"}, + }, + want: false, + }, + { + name: "edge endpoint only", + endpoints: []Endpoint{ + {Name: "edge", Host: "10.0.0.1", Port: "45671"}, + }, + want: false, + }, + { + name: "inter-network endpoint only", + endpoints: []Endpoint{ + {Name: "inter-network", Host: "10.0.0.1", Port: "35671"}, + }, + want: true, + }, + { + name: "inter-network and inter-router endpoints", + endpoints: []Endpoint{ + {Name: "inter-router", Host: "10.0.0.1", Port: "55671"}, + {Name: "inter-network", Host: "10.0.0.2", Port: "35671"}, + }, + want: true, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + l := &Link{ + Spec: LinkSpec{ + Endpoints: tt.endpoints, + }, + } + if got := l.IsInterVAN(); got != tt.want { + t.Errorf("Link.IsInterVAN() = %v, want %v", got, tt.want) + } + }) + } +} diff --git a/pkg/apis/skupper/v2alpha1/types.go b/pkg/apis/skupper/v2alpha1/types.go index e51819398..4240b6e34 100644 --- a/pkg/apis/skupper/v2alpha1/types.go +++ b/pkg/apis/skupper/v2alpha1/types.go @@ -199,6 +199,7 @@ type SiteSpec struct { DefaultIssuer string `json:"defaultIssuer,omitempty"` Edge bool `json:"edge,omitempty"` HA bool `json:"ha,omitempty"` + NetworkId string `json:"networkId,omitempty"` Settings map[string]string `json:"settings,omitempty"` } @@ -572,6 +573,11 @@ func (l *Link) IsReady() bool { meta.IsStatusConditionTrue(l.Status.Conditions, CONDITION_TYPE_OPERATIONAL) } +func (l *Link) IsInterVAN() bool { + _, found := l.Spec.GetEndpointForRole("inter-network") + return found +} + // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object // LinkList contains a List of Link instances @@ -585,6 +591,7 @@ type LinkSpec struct { Endpoints []Endpoint `json:"endpoints"` TlsCredentials string `json:"tlsCredentials,omitempty"` Cost int `json:"cost,omitempty"` + RoutingKeys []string `json:"routingKeys,omitempty"` Settings map[string]string `json:"settings,omitempty"` } @@ -972,30 +979,130 @@ type RouterAccessRole struct { Port int `json:"port,omitempty"` } -func (role RouterAccessRole) GetPort() int32 { - if role.Port != 0 { - return int32(role.Port) - } else if role.Name == "edge" { - return 45671 - } else { - return 55671 +func (r *RouterAccess) GetPortForRole(name string) int32 { + for _, role := range r.Spec.Roles { + if role.Name == name { + if role.Port > 0 { + return int32(role.Port) + } + return r.GetAllocatedPortForRole(name) + } + } + return 0 +} + +func (r *RouterAccess) GetAllocatedPortForRole(name string) int32 { + for _, role := range r.Status.Roles { + if role.Name == name { + return int32(role.Port) + } } + return 0 +} + +func (r *RouterAccess) GetAllocatedPorts() []int32 { + var ports []int32 + for _, role := range r.Spec.Roles { + if role.Port != 0 { + ports = append(ports, int32(role.Port)) + continue + } + if allocatedPort := r.GetAllocatedPortForRole(role.Name); allocatedPort != 0 { + ports = append(ports, allocatedPort) + } + } + return ports +} + +func (r *RouterAccess) MixesDynamicAndStaticPorts() bool { + var hasStatic, hasDynamic bool + for _, role := range r.Spec.Roles { + if role.Port == 0 { + hasDynamic = true + } else { + hasStatic = true + } + } + return hasStatic && hasDynamic +} + +func (r *RouterAccess) GetUnusedPorts() []int32 { + var ports []int32 + for _, role := range r.Status.Roles { + specRole := r.FindRole(role.Name) + if specRole == nil || (specRole.Port > 0 && role.Port > 0 && specRole.Port != role.Port) { + ports = append(ports, int32(role.Port)) + } + } + return ports +} + +func (r *RouterAccess) AllocatePort(role string, port int) bool { + if port == 0 { + return false + } + var ports []RouterAccessRole + for _, portStatus := range r.Status.Roles { + if portStatus.Name == role { + if portStatus.Port == port { + return false + } + continue + } + ports = append(ports, portStatus) + } + ports = append(ports, RouterAccessRole{ + Name: role, + Port: port, + }) + r.Status.Roles = ports + return true +} + +func (r *RouterAccess) ReleaseUnusedPorts(unusedPorts []int32) { + var ports []RouterAccessRole + for _, portStatus := range r.Status.Roles { + keep := true + for _, port := range unusedPorts { + if portStatus.Port == int(port) { + keep = false + break + } + } + if keep { + ports = append(ports, portStatus) + } + } + r.Status.Roles = ports } type RouterAccessSpec struct { - AccessType string `json:"accessType,omitempty"` + AccessType string `json:"accessType,omitempty"` + // +kubebuilder:validation:MaxItems=4 + // +kubebuilder:validation:XValidation:rule="self.all(r, self.filter(x, x.name == r.name).size() == 1)",message="spec.roles must not contain duplicate role names" + // +kubebuilder:validation:XValidation:rule="self.filter(x, has(x.port) && x.port > 0).all(r, self.filter(x, has(x.port) && x.port == r.port).size() == 1)",message="spec.roles must not contain duplicate non-zero ports" Roles []RouterAccessRole `json:"roles"` TlsCredentials string `json:"tlsCredentials"` GenerateTlsCredentials bool `json:"generateTlsCredentials,omitempty"` Issuer string `json:"issuer,omitempty"` BindHost string `json:"bindHost,omitempty"` SubjectAlternativeNames []string `json:"subjectAlternativeNames,omitempty"` + RoutingKeys []string `json:"routingKeys,omitempty"` Settings map[string]string `json:"settings,omitempty"` } type RouterAccessStatus struct { Status `json:",inline"` - Endpoints []Endpoint `json:"endpoints,omitempty"` + Roles []RouterAccessRole `json:"roles,omitempty"` + Endpoints []Endpoint `json:"endpoints,omitempty"` +} + +func (s *RouterAccessStatus) GetEndpoints() []Endpoint { + return s.Endpoints +} + +func (s *RouterAccessStatus) SetEndpoints(endpoints []Endpoint) { + s.Endpoints = endpoints } // +genclient diff --git a/pkg/apis/skupper/v2alpha1/zz_generated.deepcopy.go b/pkg/apis/skupper/v2alpha1/zz_generated.deepcopy.go index 4deb897ce..0c85c9b57 100644 --- a/pkg/apis/skupper/v2alpha1/zz_generated.deepcopy.go +++ b/pkg/apis/skupper/v2alpha1/zz_generated.deepcopy.go @@ -780,6 +780,11 @@ func (in *LinkSpec) DeepCopyInto(out *LinkSpec) { *out = make([]Endpoint, len(*in)) copy(*out, *in) } + if in.RoutingKeys != nil { + in, out := &in.RoutingKeys, &out.RoutingKeys + *out = make([]string, len(*in)) + copy(*out, *in) + } if in.Settings != nil { in, out := &in.Settings, &out.Settings *out = make(map[string]string, len(*in)) @@ -1205,6 +1210,11 @@ func (in *RouterAccessSpec) DeepCopyInto(out *RouterAccessSpec) { *out = make([]string, len(*in)) copy(*out, *in) } + if in.RoutingKeys != nil { + in, out := &in.RoutingKeys, &out.RoutingKeys + *out = make([]string, len(*in)) + copy(*out, *in) + } if in.Settings != nil { in, out := &in.Settings, &out.Settings *out = make(map[string]string, len(*in)) @@ -1229,6 +1239,11 @@ func (in *RouterAccessSpec) DeepCopy() *RouterAccessSpec { func (in *RouterAccessStatus) DeepCopyInto(out *RouterAccessStatus) { *out = *in in.Status.DeepCopyInto(&out.Status) + if in.Roles != nil { + in, out := &in.Roles, &out.Roles + *out = make([]RouterAccessRole, len(*in)) + copy(*out, *in) + } if in.Endpoints != nil { in, out := &in.Endpoints, &out.Endpoints *out = make([]Endpoint, len(*in)) diff --git a/tests/integration/kube/controller/site_test.go b/tests/integration/kube/controller/site_test.go index d1913fda4..bc0966b79 100644 --- a/tests/integration/kube/controller/site_test.go +++ b/tests/integration/kube/controller/site_test.go @@ -8,6 +8,7 @@ import ( "time" "github.com/skupperproject/skupper/internal/fixtures" + "github.com/skupperproject/skupper/internal/qdr" "gotest.tools/v3/assert" "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -15,6 +16,8 @@ import ( skupperv2alpha1 "github.com/skupperproject/skupper/pkg/apis/skupper/v2alpha1" ) +type RouterConfigReadyFn func(config *qdr.RouterConfig) bool + func TestSimpleSite(t *testing.T) { tc := setup(t) namespace := "simple-site" @@ -47,4 +50,175 @@ func TestSimpleSite(t *testing.T) { assert.Equal(t, deployment.Labels["skupper.io/component"], "router") assert.Equal(t, deployment.Labels["application"], "skupper-router") assert.Equal(t, len(deployment.Spec.Template.Spec.Containers), 2) + + t.Run("set network-id", func(t *testing.T) { + actualSite.Spec.NetworkId = "my-van" + actualSite, err = tc.clients.GetSkupperClient().SkupperV2alpha1().Sites(namespace).Update(ctx, actualSite, metav1.UpdateOptions{}) + assert.NilError(t, err) + waitForRouterConfigState(t, tc, namespace, func(config *qdr.RouterConfig) bool { + return config.Network.NetworkId != "" + }) + }) + + t.Run("add routerAccess with dynamic port allocation and keys", func(t *testing.T) { + ra := fixtures.RouterAccess("inter-van-ra", namespace) + ra.Spec.AccessType = "local" + ra.Spec.Roles = append(ra.Spec.Roles, skupperv2alpha1.RouterAccessRole{ + Name: "inter-network", + }) + ra.Spec.RoutingKeys = append(ra.Spec.RoutingKeys, "key1", "key2") + ra, err = tc.clients.GetSkupperClient().SkupperV2alpha1().RouterAccesses(namespace).Create(ctx, ra, metav1.CreateOptions{}) + assert.NilError(t, err) + }) + + t.Run("validate expected autoLinks present", func(t *testing.T) { + expectedAutoLinks := []qdr.AutoLink{ + { + Name: "routerAccess/inter-van-ra-inter-network", + ExternalAddress: "_xtopo/my-van", + Direction: "in", + Connection: "inter-van-ra-inter-network", + }, + { + Name: "routerAccess/inter-van-ra/key1", + Address: "key1", + Direction: "in", + Connection: "inter-van-ra-inter-network", + }, + { + Name: "routerAccess/inter-van-ra/key2", + Address: "key2", + Direction: "in", + Connection: "inter-van-ra-inter-network", + }, + } + waitForRouterConfigState(t, tc, namespace, func(config *qdr.RouterConfig) bool { + if len(config.AutoLinks) != len(expectedAutoLinks) { + return false + } + for name, autoLink := range config.AutoLinks { + var found bool + for _, wantedAutoLink := range expectedAutoLinks { + if wantedAutoLink.Name == name && autoLink == wantedAutoLink { + found = true + break + } + } + if !found { + return false + } + } + return true + }) + }) + + t.Run("create an inter-van link with exposed routingKeys", func(t *testing.T) { + link := fixtures.Link("link-van-1", namespace) + link.Spec.Endpoints = []skupperv2alpha1.Endpoint{ + { + Name: "inter-network", + Group: "skupper-router", + Host: "link-host", + Port: "35671", + }, + } + link.Spec.RoutingKeys = []string{"key1", "key2"} + _, err = tc.clients.GetSkupperClient().SkupperV2alpha1().Links(namespace).Create(ctx, link, metav1.CreateOptions{}) + assert.NilError(t, err) + }) + + t.Run("validate inter-network connector and autoLinks created", func(t *testing.T) { + expectedAutoLinks := []qdr.AutoLink{ + { + Name: "link/link-van-1", + ExternalAddress: "_xtopo/my-van", + Direction: "in", + Connection: "link-van-1", + }, + { + Name: "link/link-van-1/key1", + Address: "key1", + Direction: "in", + Connection: "link-van-1", + }, + { + Name: "link/link-van-1/key2", + Address: "key2", + Direction: "in", + Connection: "link-van-1", + }, + } + waitForRouterConfigState(t, tc, namespace, func(config *qdr.RouterConfig) bool { + connector, ok := config.Connectors["link-van-1"] + if !ok { + return false + } + assert.Equal(t, "inter-network", string(connector.Role)) + for _, wanted := range expectedAutoLinks { + got, ok := config.AutoLinks[wanted.Name] + if !ok { + return false + } + assert.Equal(t, wanted, got) + } + return true + }) + }) + + t.Run("remove networkId", func(t *testing.T) { + actualSite, err = tc.clients.GetSkupperClient().SkupperV2alpha1().Sites(namespace).Get(ctx, actualSite.Name, metav1.GetOptions{}) + assert.NilError(t, err) + actualSite.Spec.NetworkId = "" + actualSite, err = tc.clients.GetSkupperClient().SkupperV2alpha1().Sites(namespace).Update(ctx, actualSite, metav1.UpdateOptions{}) + assert.NilError(t, err) + waitForRouterConfigState(t, tc, namespace, func(config *qdr.RouterConfig) bool { + return config.Network.NetworkId == "" + }) + }) + + t.Run("ensure topology address autoLinks removed", func(t *testing.T) { + waitForRouterConfigState(t, tc, namespace, func(config *qdr.RouterConfig) bool { + return len(config.AutoLinks) == 4 + }) + }) + + t.Run("remove router access keys", func(t *testing.T) { + ra, err := tc.clients.GetSkupperClient().SkupperV2alpha1().RouterAccesses(namespace).Get(ctx, "inter-van-ra", metav1.GetOptions{}) + assert.NilError(t, err) + ra.Spec.RoutingKeys = nil + _, err = tc.clients.GetSkupperClient().SkupperV2alpha1().RouterAccesses(namespace).Update(ctx, ra, metav1.UpdateOptions{}) + assert.NilError(t, err) + }) + + t.Run("remove inter-van link keys", func(t *testing.T) { + link, err := tc.clients.GetSkupperClient().SkupperV2alpha1().Links(namespace).Get(ctx, "link-van-1", metav1.GetOptions{}) + assert.NilError(t, err) + link.Spec.RoutingKeys = nil + _, err = tc.clients.GetSkupperClient().SkupperV2alpha1().Links(namespace).Update(ctx, link, metav1.UpdateOptions{}) + assert.NilError(t, err) + }) + + t.Run("assert no autoLinks left but listener and connector are present", func(t *testing.T) { + waitForRouterConfigState(t, tc, namespace, func(config *qdr.RouterConfig) bool { + if len(config.AutoLinks) > 0 { + return false + } + _, listenerFound := config.Listeners["inter-van-ra-inter-network"] + _, connectorFound := config.Connectors["link-van-1"] + assert.Assert(t, listenerFound) + assert.Assert(t, connectorFound) + return true + }) + }) +} + +func waitForRouterConfigState(t *testing.T, tc *testContext, namespace string, state RouterConfigReadyFn) (routerConfig *qdr.RouterConfig) { + waitFor(t, 10*time.Second, 100*time.Millisecond, func() (bool, error) { + cm, err := tc.clients.GetKubeClient().CoreV1().ConfigMaps(namespace).Get(context.Background(), "skupper-router", metav1.GetOptions{}) + assert.NilError(t, err) + routerConfig, err = qdr.GetRouterConfigFromConfigMap(cm) + assert.NilError(t, err) + return state(routerConfig), nil + }) + return } diff --git a/tests/integration/kube/controller/suite_test.go b/tests/integration/kube/controller/suite_test.go index e4926642a..6fdb98db6 100644 --- a/tests/integration/kube/controller/suite_test.go +++ b/tests/integration/kube/controller/suite_test.go @@ -19,10 +19,13 @@ import ( "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/client-go/rest" + "k8s.io/client-go/tools/clientcmd" + clientcmdapi "k8s.io/client-go/tools/clientcmd/api" "sigs.k8s.io/controller-runtime/pkg/envtest" ) const controllerInstallNamespace = "skupper-system" +const envTestKubeConfig = "/tmp/envtest-kubeconfig.yaml" var ( envTestConfig *rest.Config @@ -44,11 +47,15 @@ func TestMain(m *testing.M) { if err != nil { panic(err) } + + serializeConfig() + defer func() { stopSharedController() if err := testEnv.Stop(); err != nil { fmt.Fprintf(os.Stderr, "envtest teardown warning: %v\n", err) } + os.Remove(envTestKubeConfig) }() if err := startSharedController(); err != nil { @@ -58,6 +65,27 @@ func TestMain(m *testing.M) { m.Run() } +func serializeConfig() { + kubeconfig := clientcmdapi.NewConfig() + kubeconfig.Clusters["envtest"] = &clientcmdapi.Cluster{ + Server: envTestConfig.Host, + CertificateAuthorityData: envTestConfig.CAData, + } + kubeconfig.AuthInfos["envtest"] = &clientcmdapi.AuthInfo{ + ClientCertificateData: envTestConfig.CertData, + ClientKeyData: envTestConfig.KeyData, + } + kubeconfig.Contexts["envtest"] = &clientcmdapi.Context{ + Cluster: "envtest", + AuthInfo: "envtest", + } + kubeconfig.CurrentContext = "envtest" + + if err := clientcmd.WriteToFile(*kubeconfig, envTestKubeConfig); err != nil { + panic(err) + } +} + func startSharedController() error { os.Setenv("NAMESPACE", controllerInstallNamespace) os.Setenv("CONTROLLER_NAME", "test-controller")