diff --git a/api/net/v1alpha1/types.go b/api/net/v1alpha1/types.go index 98cce7d8ee..fbb0d6c2ae 100644 --- a/api/net/v1alpha1/types.go +++ b/api/net/v1alpha1/types.go @@ -19,6 +19,18 @@ const ( // and read intentionally until then. SiteLabelKey = "net.unbounded-cloud.io/site" + // GatewayPoolLabelKey identifies the single GatewayPool assignment adopted + // by the node agent. An absent label means the node is a worker. + GatewayPoolLabelKey = "net.unbounded-cloud.io/gateway-pool" + + // EffectiveSiteAnnotation records the Site adopted by a node-agent Pod. + EffectiveSiteAnnotation = "net.unbounded-cloud.io/effective-site" + + // EffectiveGatewayPoolAnnotation records the GatewayPool adopted by a + // node-agent Pod. A present empty value means the worker role; an absent + // key means acknowledgment is pending. + EffectiveGatewayPoolAnnotation = "net.unbounded-cloud.io/effective-gateway-pool" + // NodeDiscoveredPublicIPAnnotation contains the translated source IP // discovered by the node agent. It does not prove that the WireGuard port is // reachable through symmetric NAT or CGNAT. diff --git a/cmd/unbounded-net-controller/main.go b/cmd/unbounded-net-controller/main.go index de0f635dfa..b369a0ff8e 100644 --- a/cmd/unbounded-net-controller/main.go +++ b/cmd/unbounded-net-controller/main.go @@ -727,6 +727,21 @@ func run(cfg *config.Config, forceNotLeader bool) error { }() } + nodeAssignmentCtrl, err := controller.NewNodeAssignmentController( + clientset, + informerFactory.Core().V1().Nodes(), + nodeAuthCaches.pods, + ) + if err != nil { + klog.Errorf("Failed to create Node assignment controller: %v", err) + } else { + go func() { + if err := nodeAssignmentCtrl.Run(ctx, 2); err != nil { + klog.Errorf("Node assignment controller error: %v", err) + } + }() + } + // Create and start peering aggregation controller peeringAggCtrl, err := controller.NewPeeringAggregationController(dynamicClient, dynamicInformerFactory) if err != nil { diff --git a/cmd/unbounded-net-controller/node_auth_informers.go b/cmd/unbounded-net-controller/node_auth_informers.go index 2dcc236225..2c74c729ca 100644 --- a/cmd/unbounded-net-controller/node_auth_informers.go +++ b/cmd/unbounded-net-controller/node_auth_informers.go @@ -5,14 +5,17 @@ package main import ( "context" + "fmt" "time" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/client-go/informers" coreinformers "k8s.io/client-go/informers/core/v1" "k8s.io/client-go/kubernetes" + "k8s.io/client-go/tools/cache" "github.com/Azure/unbounded/internal/net/authn" + netcontroller "github.com/Azure/unbounded/internal/net/controller" ) // These watches run on every serving replica, independently of leadership. @@ -43,6 +46,12 @@ func newNodeAuthInformers(ctx context.Context, client kubernetes.Interface, name result.pods.Informer() result.serviceAccounts.Informer() + if err := result.pods.Informer().AddIndexers(cache.Indexers{ + netcontroller.NodeAgentPodNodeNameIndex: netcontroller.NodeAgentPodNodeNameIndexFunc, + }); err != nil { + panic(fmt.Sprintf("add node-agent Pod Node index: %v", err)) + } + return result } diff --git a/cmd/unbounded-net-controller/node_auth_informers_test.go b/cmd/unbounded-net-controller/node_auth_informers_test.go index 4997ab8034..c38d5aca5c 100644 --- a/cmd/unbounded-net-controller/node_auth_informers_test.go +++ b/cmd/unbounded-net-controller/node_auth_informers_test.go @@ -26,6 +26,7 @@ import ( "k8s.io/client-go/tools/cache" "github.com/Azure/unbounded/internal/net/authn" + netcontroller "github.com/Azure/unbounded/internal/net/controller" ) func nodeAuthObjects() (*corev1.Pod, *corev1.ServiceAccount, *authn.KubernetesServiceAccountIdentity) { @@ -182,6 +183,11 @@ func TestNodeAuthInformersScopeAndLifecycle(t *testing.T) { t.Fatal(err) } + indexed, err := caches.pods.Informer().GetIndexer().ByIndex(netcontroller.NodeAgentPodNodeNameIndex, pod.Spec.NodeName) + if err != nil || len(indexed) != 1 || indexed[0].(*corev1.Pod).UID != pod.UID { + t.Fatalf("node-agent Pods indexed by Node = %#v, err=%v", indexed, err) + } + if _, err := caches.serviceAccounts.Lister().ServiceAccounts(sa.Namespace).Get(sa.Name); err != nil { t.Fatalf("unlabeled service account must be watched: %v", err) } diff --git a/deploy/net/controller/02-rbac.yaml.tmpl b/deploy/net/controller/02-rbac.yaml.tmpl index d479483aa8..0da9facb00 100644 --- a/deploy/net/controller/02-rbac.yaml.tmpl +++ b/deploy/net/controller/02-rbac.yaml.tmpl @@ -13,8 +13,9 @@ rules: # Nodes: read, patch, and update for CIDR assignment and site labeling. # Cannot be further scoped -- the controller must operate on any node in the cluster. # Fields restricted by ValidatingAdmissionPolicy (09-vap.yaml.tmpl): - # labels/annotations: only net.unbounded-cloud.io/* prefix - # spec: only podCIDR and podCIDRs (Kubernetes enforces immutability after initial set) + # labels: only net.unbounded-cloud.io/* and the canonical Site label + # annotations: only net.unbounded-cloud.io/* + # spec: only pod CIDRs, plus the node agent's exact gateway taint - apiGroups: [""] resources: ["nodes"] verbs: ["get", "list", "watch", "patch", "update"] @@ -186,10 +187,12 @@ rules: resources: ["services"] resourceNames: ["unbounded-net-controller"] verbs: ["get"] - # Pods: dashboard status and local OIDC bound-object validation. + # Pods: dashboard status, local OIDC bound-object validation, and targeted + # replacement of node-agent Pods after acknowledged assignment changes. + # ValidatingAdmissionPolicy restricts delete to DaemonSet-owned node Pods. - apiGroups: [""] resources: ["pods"] - verbs: ["get", "list", "watch"] + verbs: ["get", "list", "watch", "delete"] # ServiceAccounts: local OIDC bound-object validation (controller namespace only). - apiGroups: [""] resources: ["serviceaccounts"] diff --git a/deploy/net/controller/09-vap.yaml.tmpl b/deploy/net/controller/09-vap.yaml.tmpl index 5ab6fb2613..fbe34b1ea7 100644 --- a/deploy/net/controller/09-vap.yaml.tmpl +++ b/deploy/net/controller/09-vap.yaml.tmpl @@ -63,11 +63,10 @@ spec: --- # Restrict what fields unbounded-net service accounts may modify on Node objects. -# The controller and node agent may ONLY modify: -# - metadata.labels with the net.unbounded-cloud.io/ prefix -# - metadata.annotations with the net.unbounded-cloud.io/ prefix -# - spec.podCIDR and spec.podCIDRs (Kubernetes enforces immutability after initial set) -# All other node fields (spec.taints, spec.unschedulable, status, etc.) must remain unchanged. +# The controller owns networking labels and pod CIDRs. +# The node agent may only modify its public key, MTU, discovery annotations, +# and exact gateway taint on its token-bound Node. +# All other node fields (spec.unschedulable, status, etc.) must remain unchanged. # The CEL expressions iterate over map keys dynamically rather than enumerating fields, # so new fields added to the Node spec in future Kubernetes versions are automatically protected. apiVersion: admissionregistration.k8s.io/v1 @@ -91,14 +90,41 @@ spec: request.userInfo.username == 'system:serviceaccount:{{ default "unbounded-system" .Namespace }}:unbounded-net-controller' || request.userInfo.username == 'system:serviceaccount:{{ default "unbounded-system" .Namespace }}:unbounded-net-node' validations: + - expression: >- + request.userInfo.username != 'system:serviceaccount:{{ default "unbounded-system" .Namespace }}:unbounded-net-node' || + object.metadata.?labels.orValue({}) == oldObject.metadata.?labels.orValue({}) + message: "unbounded-net-node may not modify Node labels" + - expression: >- + request.userInfo.username != 'system:serviceaccount:{{ default "unbounded-system" .Namespace }}:unbounded-net-node' || + (object.spec.?podCIDR.orValue('') == oldObject.spec.?podCIDR.orValue('') && + object.spec.?podCIDRs.orValue([]) == oldObject.spec.?podCIDRs.orValue([])) + message: "unbounded-net-node may not modify controller-owned pod CIDRs" # Status must be unchanged. - expression: "object.status == oldObject.status" message: "unbounded-net may not modify node status" - # Spec: only podCIDR and podCIDRs may change. All other spec fields are protected. + # Spec: only podCIDR and podCIDRs may change generally. The node agent may + # additionally add or remove its exact gateway taint. + - expression: >- + object.spec.all(k, + k in ['podCIDR', 'podCIDRs'] || + (request.userInfo.username == 'system:serviceaccount:{{ default "unbounded-system" .Namespace }}:unbounded-net-node' && k == 'taints') || + (k in oldObject.spec && object.spec[k] == oldObject.spec[k])) && + oldObject.spec.all(k, + k in ['podCIDR', 'podCIDRs'] || + (request.userInfo.username == 'system:serviceaccount:{{ default "unbounded-system" .Namespace }}:unbounded-net-node' && k == 'taints') || + (k in object.spec && oldObject.spec[k] == object.spec[k])) + message: "unbounded-net may only modify pod CIDRs and the node agent's gateway taint" - expression: >- - object.spec.all(k, k in ['podCIDR', 'podCIDRs'] || (k in oldObject.spec && object.spec[k] == oldObject.spec[k])) && - oldObject.spec.all(k, k in ['podCIDR', 'podCIDRs'] || (k in object.spec && oldObject.spec[k] == object.spec[k])) - message: "unbounded-net may only modify spec.podCIDR and spec.podCIDRs on nodes" + request.userInfo.username != 'system:serviceaccount:{{ default "unbounded-system" .Namespace }}:unbounded-net-node' || + (request.userInfo.extra[?'authentication.kubernetes.io/node-name'][0].orValue('') == object.metadata.name && + request.userInfo.extra[?'authentication.kubernetes.io/node-uid'][0].orValue('') == object.metadata.uid && + object.spec.?taints.orValue([]).filter(t, !(t.key == 'net.unbounded-cloud.io/gateway-node' && + t.?value.orValue('') == 'true' && t.effect == 'NoSchedule')) == + oldObject.spec.?taints.orValue([]).filter(t, !(t.key == 'net.unbounded-cloud.io/gateway-node' && + t.?value.orValue('') == 'true' && t.effect == 'NoSchedule')) && + object.spec.?taints.orValue([]).filter(t, t.key == 'net.unbounded-cloud.io/gateway-node' && + t.?value.orValue('') == 'true' && t.effect == 'NoSchedule').size() <= 1) + message: "unbounded-net-node may only add or remove its exact gateway taint on its own Node" - expression: >- request.userInfo.username != 'system:serviceaccount:{{ default "unbounded-system" .Namespace }}:unbounded-net-node' || (((!has(object.metadata.annotations) || !('net.unbounded-cloud.io/wireguard-port' in object.metadata.annotations)) && @@ -133,6 +159,19 @@ spec: oldObject.metadata.annotations.all(k, k.startsWith('net.unbounded-cloud.io/') || (k in object.metadata.annotations && oldObject.metadata.annotations[k] == object.metadata.annotations[k]))) message: "unbounded-net may only modify annotations with net.unbounded-cloud.io/ prefix" + - expression: >- + request.userInfo.username != 'system:serviceaccount:{{ default "unbounded-system" .Namespace }}:unbounded-net-node' || + (object.metadata.?annotations.orValue({}).all(k, + k in ['net.unbounded-cloud.io/wg-pubkey', 'net.unbounded-cloud.io/tunnel-mtu', + 'net.unbounded-cloud.io/discovered-public-ip', 'net.unbounded-cloud.io/discovered-public-ip-expires-at'] || + (k in oldObject.metadata.?annotations.orValue({}) && + object.metadata.annotations[k] == oldObject.metadata.annotations[k])) && + oldObject.metadata.?annotations.orValue({}).all(k, + k in ['net.unbounded-cloud.io/wg-pubkey', 'net.unbounded-cloud.io/tunnel-mtu', + 'net.unbounded-cloud.io/discovered-public-ip', 'net.unbounded-cloud.io/discovered-public-ip-expires-at'] || + (k in object.metadata.?annotations.orValue({}) && + oldObject.metadata.annotations[k] == object.metadata.annotations[k]))) + message: "unbounded-net-node may only modify agent-owned Node annotations" --- apiVersion: admissionregistration.k8s.io/v1 kind: ValidatingAdmissionPolicyBinding @@ -145,6 +184,40 @@ spec: policyName: unbounded-net-node-field-restriction validationActions: ["Deny"] +--- +# GatewayPool names are written verbatim into Node label values. Names are +# immutable, so only CREATE is checked; existing longer pools can still be +# updated, drained, and finalized. +apiVersion: admissionregistration.k8s.io/v1 +kind: ValidatingAdmissionPolicy +metadata: + name: unbounded-net-gateway-pool-name + labels: + app.kubernetes.io/name: unbounded-net-controller + app.kubernetes.io/component: controller +spec: + failurePolicy: Fail + matchConstraints: + resourceRules: + - apiGroups: ["net.unbounded-cloud.io"] + apiVersions: ["v1alpha1"] + operations: ["CREATE"] + resources: ["gatewaypools"] + validations: + - expression: "object.metadata.name.size() <= 63" + message: "GatewayPool names must fit the 63-character Kubernetes label value limit" +--- +apiVersion: admissionregistration.k8s.io/v1 +kind: ValidatingAdmissionPolicyBinding +metadata: + name: unbounded-net-gateway-pool-name + labels: + app.kubernetes.io/name: unbounded-net-controller + app.kubernetes.io/component: controller +spec: + policyName: unbounded-net-gateway-pool-name + validationActions: ["Deny"] + --- # Protect the ownership of public IP annotations on every Node. The discovered # address and its expiry belong to the agent running on that Node. The declared @@ -196,3 +269,110 @@ metadata: spec: policyName: unbounded-net-public-ip-annotation-ownership validationActions: ["Deny"] + +--- +# Restrict node-agent Pod updates to effective assignment acknowledgment by +# the service account token bound to that exact Pod. +apiVersion: admissionregistration.k8s.io/v1 +kind: ValidatingAdmissionPolicy +metadata: + name: unbounded-net-effective-assignment-ownership + labels: + app.kubernetes.io/name: unbounded-net-node + app.kubernetes.io/component: node +spec: + failurePolicy: Fail + matchConstraints: + resourceRules: + - apiGroups: [""] + apiVersions: ["v1"] + operations: ["UPDATE"] + resources: ["pods"] + matchConditions: + - name: "is-unbounded-net-node-sa" + expression: "request.userInfo.username == 'system:serviceaccount:{{ default "unbounded-system" .Namespace }}:unbounded-net-node'" + validations: + - expression: >- + request.userInfo.extra[?'authentication.kubernetes.io/pod-name'][0].orValue('') == object.metadata.name && + request.userInfo.extra[?'authentication.kubernetes.io/pod-uid'][0].orValue('') == object.metadata.uid + message: "unbounded-net-node may only acknowledge assignment on its own bound Pod" + - expression: >- + object.spec == oldObject.spec && + has(object.status) == has(oldObject.status) && + (!has(object.status) || object.status == oldObject.status) && + has(object.metadata.labels) == has(oldObject.metadata.labels) && + (!has(object.metadata.labels) || object.metadata.labels == oldObject.metadata.labels) && + has(object.metadata.ownerReferences) == has(oldObject.metadata.ownerReferences) && + (!has(object.metadata.ownerReferences) || object.metadata.ownerReferences == oldObject.metadata.ownerReferences) && + has(object.metadata.finalizers) == has(oldObject.metadata.finalizers) && + (!has(object.metadata.finalizers) || object.metadata.finalizers == oldObject.metadata.finalizers) + message: "unbounded-net-node may only modify effective assignment annotations on its own Pod" + - expression: >- + (!has(object.metadata.annotations) && !has(oldObject.metadata.annotations)) || + (!has(oldObject.metadata.annotations) && object.metadata.annotations.all(k, + k in ['net.unbounded-cloud.io/effective-site', 'net.unbounded-cloud.io/effective-gateway-pool'])) || + (!has(object.metadata.annotations) && oldObject.metadata.annotations.all(k, + k in ['net.unbounded-cloud.io/effective-site', 'net.unbounded-cloud.io/effective-gateway-pool'])) || + (has(object.metadata.annotations) && has(oldObject.metadata.annotations) && + object.metadata.annotations.all(k, + k in ['net.unbounded-cloud.io/effective-site', 'net.unbounded-cloud.io/effective-gateway-pool'] || + (k in oldObject.metadata.annotations && object.metadata.annotations[k] == oldObject.metadata.annotations[k])) && + oldObject.metadata.annotations.all(k, + k in ['net.unbounded-cloud.io/effective-site', 'net.unbounded-cloud.io/effective-gateway-pool'] || + (k in object.metadata.annotations && oldObject.metadata.annotations[k] == object.metadata.annotations[k]))) + message: "unbounded-net-node may only modify effective assignment annotations" +--- +apiVersion: admissionregistration.k8s.io/v1 +kind: ValidatingAdmissionPolicyBinding +metadata: + name: unbounded-net-effective-assignment-ownership + labels: + app.kubernetes.io/name: unbounded-net-node + app.kubernetes.io/component: node +spec: + policyName: unbounded-net-effective-assignment-ownership + validationActions: ["Deny"] + +--- +# Restrict controller Pod deletion to the node-agent Pods owned by the +# unbounded-net-node DaemonSet. +apiVersion: admissionregistration.k8s.io/v1 +kind: ValidatingAdmissionPolicy +metadata: + name: unbounded-net-node-pod-delete-restriction + labels: + app.kubernetes.io/name: unbounded-net-controller + app.kubernetes.io/component: controller +spec: + failurePolicy: Fail + matchConstraints: + resourceRules: + - apiGroups: [""] + apiVersions: ["v1"] + operations: ["DELETE"] + resources: ["pods"] + matchConditions: + - name: "is-unbounded-net-controller-sa" + expression: "request.userInfo.username == 'system:serviceaccount:{{ default "unbounded-system" .Namespace }}:unbounded-net-controller'" + validations: + - expression: >- + oldObject.metadata.namespace == '{{ default "unbounded-system" .Namespace }}' && + oldObject.metadata.labels['app.kubernetes.io/name'] == 'unbounded-net-node' && + oldObject.metadata.labels['app.kubernetes.io/component'] == 'node' && + oldObject.metadata.ownerReferences.exists(owner, + owner.controller == true && + owner.apiVersion == 'apps/v1' && + owner.kind == 'DaemonSet' && + owner.name == 'unbounded-net-node') + message: "unbounded-net controller may only delete DaemonSet-owned unbounded-net-node Pods" +--- +apiVersion: admissionregistration.k8s.io/v1 +kind: ValidatingAdmissionPolicyBinding +metadata: + name: unbounded-net-node-pod-delete-restriction + labels: + app.kubernetes.io/name: unbounded-net-controller + app.kubernetes.io/component: controller +spec: + policyName: unbounded-net-node-pod-delete-restriction + validationActions: ["Deny"] diff --git a/deploy/net/controller/rbac_test.go b/deploy/net/controller/rbac_test.go index 45f251af0f..8dcd314dad 100644 --- a/deploy/net/controller/rbac_test.go +++ b/deploy/net/controller/rbac_test.go @@ -62,6 +62,10 @@ func TestBoundObjectCacheRBAC(t *testing.T) { t.Fatalf("missing Pod %s permission in controller namespace", verb) } } + + if !grantsPodRead(role.Rules, "delete") { + t.Fatal("missing targeted Pod delete permission in controller namespace") + } } } diff --git a/deploy/net/node/02-rbac.yaml.tmpl b/deploy/net/node/02-rbac.yaml.tmpl index 5e4de5c3ad..120e6bcd66 100644 --- a/deploy/net/node/02-rbac.yaml.tmpl +++ b/deploy/net/node/02-rbac.yaml.tmpl @@ -67,3 +67,36 @@ subjects: - kind: ServiceAccount name: unbounded-net-node namespace: {{ default "unbounded-system" .Namespace }} +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: Role +metadata: + name: unbounded-net-node + namespace: {{ default "unbounded-system" .Namespace }} + labels: + app.kubernetes.io/name: unbounded-net-node + app.kubernetes.io/component: node +rules: + # Pods: acknowledge the effective assignment on this agent's own Pod. + # ValidatingAdmissionPolicy restricts these verbs to the bound Pod and the + # two effective assignment annotations. + - apiGroups: [""] + resources: ["pods"] + verbs: ["get", "patch"] +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: RoleBinding +metadata: + name: unbounded-net-node + namespace: {{ default "unbounded-system" .Namespace }} + labels: + app.kubernetes.io/name: unbounded-net-node + app.kubernetes.io/component: node +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: Role + name: unbounded-net-node +subjects: + - kind: ServiceAccount + name: unbounded-net-node + namespace: {{ default "unbounded-system" .Namespace }} diff --git a/deploy/net/node/03-daemonset.yaml.tmpl b/deploy/net/node/03-daemonset.yaml.tmpl index 910e30d6f1..69d5fd28c7 100644 --- a/deploy/net/node/03-daemonset.yaml.tmpl +++ b/deploy/net/node/03-daemonset.yaml.tmpl @@ -69,6 +69,14 @@ spec: valueFrom: fieldRef: fieldPath: metadata.namespace + - name: POD_NAME + valueFrom: + fieldRef: + fieldPath: metadata.name + - name: POD_UID + valueFrom: + fieldRef: + fieldPath: metadata.uid args: - -v=$(LOG_LEVEL) - --config-file=/etc/unbounded-net/config.yaml diff --git a/deploy/net/node/rbac_test.go b/deploy/net/node/rbac_test.go new file mode 100644 index 0000000000..c42f6fc78b --- /dev/null +++ b/deploy/net/node/rbac_test.go @@ -0,0 +1,112 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package node + +import ( + "bytes" + "io" + "os" + "path/filepath" + "slices" + "testing" + + appsv1 "k8s.io/api/apps/v1" + rbacv1 "k8s.io/api/rbac/v1" + utilyaml "k8s.io/apimachinery/pkg/util/yaml" + + "github.com/Azure/unbounded/hack/cmd/render-manifests/render" +) + +func TestOwnPodAcknowledgmentRBAC(t *testing.T) { + for _, namespace := range []string{"unbounded-system", "custom-system"} { + t.Run(namespace, func(t *testing.T) { + output := t.TempDir() + if err := render.Render(".", output, map[string]string{"Namespace": namespace}); err != nil { + t.Fatal(err) + } + + raw, err := os.ReadFile(filepath.Join(output, "02-rbac.yaml")) + if err != nil { + t.Fatal(err) + } + + decoder := utilyaml.NewYAMLOrJSONDecoder(bytes.NewReader(raw), 4096) + found := false + + for { + var role rbacv1.Role + if err := decoder.Decode(&role); err != nil { + if err == io.EOF { + break + } + + t.Fatal(err) + } + + if role.Kind != "Role" || role.Namespace != namespace { + continue + } + + for _, rule := range role.Rules { + if slices.Equal(rule.APIGroups, []string{""}) && + slices.Equal(rule.Resources, []string{"pods"}) { + if !slices.Equal(rule.Verbs, []string{"get", "patch"}) || len(rule.ResourceNames) != 0 { + t.Fatalf("unexpected own-Pod permission: %+v", rule) + } + + found = true + } + } + } + + if !found { + t.Fatal("missing namespaced own-Pod get/patch permission") + } + }) + } +} + +func TestNodePodIdentityEnvironment(t *testing.T) { + output := t.TempDir() + if err := render.Render(".", output, nil); err != nil { + t.Fatal(err) + } + + raw, err := os.ReadFile(filepath.Join(output, "03-daemonset.yaml")) + if err != nil { + t.Fatal(err) + } + + var daemonSet appsv1.DaemonSet + + decoder := utilyaml.NewYAMLOrJSONDecoder(bytes.NewReader(raw), 4096) + for daemonSet.Kind != "DaemonSet" { + if err := decoder.Decode(&daemonSet); err != nil { + t.Fatal(err) + } + } + + if len(daemonSet.Spec.Template.Spec.Containers) != 1 { + t.Fatalf("containers = %d, want 1", len(daemonSet.Spec.Template.Spec.Containers)) + } + + fields := map[string]string{} + + for _, env := range daemonSet.Spec.Template.Spec.Containers[0].Env { + if env.ValueFrom != nil && env.ValueFrom.FieldRef != nil { + fields[env.Name] = env.ValueFrom.FieldRef.FieldPath + } + } + + for name, field := range map[string]string{ + "NODE_NAME": "spec.nodeName", + "POD_NAMESPACE": "metadata.namespace", + "POD_NAME": "metadata.name", + "POD_UID": "metadata.uid", + } { + if fields[name] != field { + t.Fatalf("%s fieldRef = %q, want %q", name, fields[name], field) + } + } +} diff --git a/internal/net/controller/gatewaypool_controller.go b/internal/net/controller/gatewaypool_controller.go index fc6a650529..75be11a13a 100644 --- a/internal/net/controller/gatewaypool_controller.go +++ b/internal/net/controller/gatewaypool_controller.go @@ -7,12 +7,14 @@ package controller import ( "context" "encoding/json" + stderrors "errors" "fmt" "net" "net/netip" "reflect" "sort" "strconv" + "strings" "sync" "sync/atomic" "time" @@ -25,6 +27,7 @@ import ( "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/types" utilruntime "k8s.io/apimachinery/pkg/util/runtime" + "k8s.io/apimachinery/pkg/util/validation" "k8s.io/apimachinery/pkg/util/wait" "k8s.io/client-go/dynamic" "k8s.io/client-go/dynamic/dynamicinformer" @@ -58,6 +61,8 @@ const ( // nodes for gateway-to-gateway peering. Each gateway node gets a unique // port starting from this value. gatewayWireguardPortBase int32 = 51821 + + gatewayAssignmentsQueueKey = "\x00gateway-assignments" ) // GatewayPoolController manages GatewayPool status based on node selector matches. @@ -120,6 +125,7 @@ func NewGatewayPoolController( if _, err := nodeInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ AddFunc: func(obj interface{}) { gc.enqueuePoolsForNodes(obj.(*corev1.Node)) //nolint:errcheck + gc.workqueue.Add(gatewayAssignmentsQueueKey) }, UpdateFunc: func(old, new interface{}) { oldNode := old.(*corev1.Node) //nolint:errcheck @@ -134,6 +140,7 @@ func NewGatewayPoolController( nodeAnnotationChanged(oldNode, newNode, WireGuardPortAnnotation) || !stringSlicesEqual(oldNode.Spec.PodCIDRs, newNode.Spec.PodCIDRs) { gc.enqueuePoolsForNodes(oldNode, newNode) + gc.workqueue.Add(gatewayAssignmentsQueueKey) } }, DeleteFunc: func(obj interface{}) { @@ -143,6 +150,7 @@ func NewGatewayPoolController( } gc.enqueuePoolsForNodes(node) + gc.workqueue.Add(gatewayAssignmentsQueueKey) }, }); err != nil { return nil, fmt.Errorf("failed to add node event handler: %w", err) @@ -152,12 +160,14 @@ func NewGatewayPoolController( if _, err := gatewayPoolInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{ AddFunc: func(obj interface{}) { gc.enqueuePool(obj) + gc.workqueue.Add(gatewayAssignmentsQueueKey) }, UpdateFunc: func(old, new interface{}) { gc.enqueuePool(new) + gc.workqueue.Add(gatewayAssignmentsQueueKey) }, DeleteFunc: func(obj interface{}) { - // Nothing to clean up + gc.workqueue.Add(gatewayAssignmentsQueueKey) }, }); err != nil { return nil, fmt.Errorf("failed to add gateway pool event handler: %w", err) @@ -205,6 +215,7 @@ func (gc *GatewayPoolController) Run(ctx context.Context, workers int) error { // Rebuild the gateway port allocator from existing pool statuses gc.seedGatewayPorts() + gc.workqueue.Add(gatewayAssignmentsQueueKey) klog.Info("Starting GatewayPool workers") @@ -232,7 +243,14 @@ func (gc *GatewayPoolController) processNextWorkItem(ctx context.Context) bool { defer gc.workqueue.Done(key) start := time.Now() - err := gc.syncPool(ctx, key) + + var err error + if key == gatewayAssignmentsQueueKey { + err = gc.reconcileNodeGatewayPoolAssignments(ctx) + } else { + err = gc.syncPool(ctx, key) + } + duration := time.Since(start).Seconds() reconciliationDuration.WithLabelValues("GatewayPools").Observe(duration) @@ -246,7 +264,13 @@ func (gc *GatewayPoolController) processNextWorkItem(ctx context.Context) bool { reconciliationErrors.WithLabelValues("GatewayPools").Inc() reconciliationTotal.WithLabelValues("GatewayPools", "error").Inc() workqueueRetries.WithLabelValues("GatewayPools").Inc() - utilruntime.HandleError(fmt.Errorf("error syncing gateway pool %s: %v", key, err)) + + if key == gatewayAssignmentsQueueKey { + utilruntime.HandleError(fmt.Errorf("error syncing node gateway pool assignments: %v", err)) + } else { + utilruntime.HandleError(fmt.Errorf("error syncing gateway pool %s: %v", key, err)) + } + gc.workqueue.AddRateLimited(key) return true @@ -793,6 +817,102 @@ func (gc *GatewayPoolController) preferredGatewayPoolForNode(node *corev1.Node, return eligible[0], true } +func (gc *GatewayPoolController) reconcileNodeGatewayPoolAssignments(ctx context.Context) error { + gc.updatePoolsCache() + + nodes, err := gc.nodeLister.List(labels.Everything()) + if err != nil { + return fmt.Errorf("list nodes: %w", err) + } + + now := time.Now() + + var assignmentErrors []error + + for _, node := range nodes { + if err := gc.reconcileNodeGatewayPoolAssignment(ctx, node, now); err != nil { + assignmentErrors = append(assignmentErrors, err) + } + } + + return stderrors.Join(assignmentErrors...) +} + +func (gc *GatewayPoolController) reconcileNodeGatewayPoolAssignment(ctx context.Context, node *corev1.Node, now time.Time) error { + desired, _ := gc.preferredGatewayPoolForNode(node, now) + if desired != "" { + if problems := validation.IsValidLabelValue(desired); len(problems) != 0 { + return fmt.Errorf("GatewayPool %q cannot be assigned to Node %q because its raw name is not a valid Kubernetes label value: %s", desired, node.Name, strings.Join(problems, "; ")) + } + } + + current := "" + if node.Labels != nil { + current = node.Labels[unboundednetv1alpha1.GatewayPoolLabelKey] + } + + hasPort := node.Annotations != nil && node.Annotations[WireGuardPortAnnotation] != "" + if current == desired && (desired != "" || !hasPort) { + return nil + } + + if node.UID == "" || node.ResourceVersion == "" { + return fmt.Errorf("cannot patch GatewayPool assignment on Node %q without cached UID and resourceVersion", node.Name) + } + + patchOperations := []map[string]interface{}{ + {"op": "test", "path": "/metadata/uid", "value": string(node.UID)}, + {"op": "test", "path": "/metadata/resourceVersion", "value": node.ResourceVersion}, + } + + if desired == "" { + if current != "" { + patchOperations = append(patchOperations, map[string]interface{}{ + "op": "remove", + "path": "/metadata/labels/" + escapeJSONPointer(unboundednetv1alpha1.GatewayPoolLabelKey), + }) + } + + if hasPort { + patchOperations = append(patchOperations, map[string]interface{}{ + "op": "remove", + "path": "/metadata/annotations/" + escapeJSONPointer(WireGuardPortAnnotation), + }) + } + } else if len(node.Labels) == 0 { + patchOperations = append(patchOperations, map[string]interface{}{ + "op": "add", + "path": "/metadata/labels", + "value": map[string]string{ + unboundednetv1alpha1.GatewayPoolLabelKey: desired, + }, + }) + } else { + patchOperations = append(patchOperations, map[string]interface{}{ + "op": "add", + "path": "/metadata/labels/" + escapeJSONPointer(unboundednetv1alpha1.GatewayPoolLabelKey), + "value": desired, + }) + } + + patch, err := json.Marshal(patchOperations) + if err != nil { + return fmt.Errorf("marshal GatewayPool assignment patch for Node %q: %w", node.Name, err) + } + + if _, err := gc.clientset.CoreV1().Nodes().Patch(ctx, node.Name, types.JSONPatchType, patch, metav1.PatchOptions{}); err != nil { + return fmt.Errorf( + "patch GatewayPool assignment on Node %q from cached UID %q and resourceVersion %q; retry after informer refresh: %w", + node.Name, + node.UID, + node.ResourceVersion, + err, + ) + } + + return nil +} + // getNodeExternalIPs returns the external IPs of a node. func getNodeExternalIPs(node *corev1.Node) []string { var externalIPs []string diff --git a/internal/net/controller/gatewaypool_helpers_test.go b/internal/net/controller/gatewaypool_helpers_test.go index 87663a33aa..f8006e2eef 100644 --- a/internal/net/controller/gatewaypool_helpers_test.go +++ b/internal/net/controller/gatewaypool_helpers_test.go @@ -14,6 +14,7 @@ import ( corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/client-go/dynamic/fake" @@ -31,6 +32,23 @@ type fakeInformer struct { store cache.Store } +type orderedNodeLister struct { + corev1listers.NodeLister + nodes []*corev1.Node +} + +func (l orderedNodeLister) List(selector labels.Selector) ([]*corev1.Node, error) { + var result []*corev1.Node + + for _, node := range l.nodes { + if selector.Matches(labels.Set(node.Labels)) { + result = append(result, node) + } + } + + return result, nil +} + func (f *fakeInformer) GetStore() cache.Store { return f.store } func (f *fakeInformer) HasSynced() bool { return true } @@ -120,6 +138,7 @@ func TestNodeEligibilityAndPreferredPoolHelpers(t *testing.T) { NodeSelector: map[string]string{"role": "gateway"}, }, } + poolExternalB := &unboundednetv1alpha1.GatewayPool{ ObjectMeta: metav1.ObjectMeta{Name: "pool-a"}, Spec: unboundednetv1alpha1.GatewayPoolSpec{ @@ -1097,3 +1116,279 @@ func TestCleanupStaleGatewayPoolNodesPreservesPreferredPoolHandoff(t *testing.T) t.Fatalf("spec.gatewayPool = %q, want %q", got, internalPool.Name) } } + +func TestReconcileNodeGatewayPoolAssignments(t *testing.T) { + t.Run("writes raw preferred pool and clears demoted state", func(t *testing.T) { + gateway := makeGatewayNode("gateway", map[string]string{"role": "gateway"}, true, "pub") + gateway.UID = "gateway-uid" + gateway.ResourceVersion = "1" + worker := makeGatewayNode("worker", map[string]string{ + "example.com/preserved": "true", + "role": "worker", + unboundednetv1alpha1.GatewayPoolLabelKey: "old-pool", + }, true, "pub-worker") + worker.UID = "worker-uid" + worker.ResourceVersion = "2" + worker.Annotations[WireGuardPortAnnotation] = "51822" + worker.Annotations["example.com/preserved"] = "true" + + pool := unboundednetv1alpha1.GatewayPool{ + ObjectMeta: metav1.ObjectMeta{Name: "pool.raw"}, + Spec: unboundednetv1alpha1.GatewayPoolSpec{ + Type: gatewayPoolTypeExternal, + NodeSelector: map[string]string{"role": "gateway"}, + }, + } + + client := kubefake.NewSimpleClientset(gateway, worker) + + indexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{}) + for _, node := range []*corev1.Node{gateway, worker} { + if err := indexer.Add(node); err != nil { + t.Fatal(err) + } + } + + controller := &GatewayPoolController{ + clientset: client, + nodeLister: corev1listers.NewNodeLister(indexer), + gatewayPoolInformer: &fakeInformer{store: gatewayPoolStore(t, &pool)}, + } + if err := controller.reconcileNodeGatewayPoolAssignments(t.Context()); err != nil { + t.Fatal(err) + } + + gotGateway, err := client.CoreV1().Nodes().Get(t.Context(), gateway.Name, metav1.GetOptions{}) + if err != nil { + t.Fatal(err) + } + + if gotGateway.Labels[unboundednetv1alpha1.GatewayPoolLabelKey] != pool.Name { + t.Fatalf("gateway labels = %#v, want raw pool name %q", gotGateway.Labels, pool.Name) + } + + gotWorker, err := client.CoreV1().Nodes().Get(t.Context(), worker.Name, metav1.GetOptions{}) + if err != nil { + t.Fatal(err) + } + + if _, ok := gotWorker.Labels[unboundednetv1alpha1.GatewayPoolLabelKey]; ok { + t.Fatalf("worker retained GatewayPool label: %#v", gotWorker.Labels) + } + + if _, ok := gotWorker.Annotations[WireGuardPortAnnotation]; ok { + t.Fatalf("worker retained WireGuard port annotation: %#v", gotWorker.Annotations) + } + + if gotWorker.Labels["example.com/preserved"] != "true" || gotWorker.Annotations["example.com/preserved"] != "true" { + t.Fatalf("worker unrelated metadata was not preserved: labels=%#v annotations=%#v", gotWorker.Labels, gotWorker.Annotations) + } + }) + + t.Run("adds labels map when absent", func(t *testing.T) { + node := makeGatewayNode("gateway", nil, true, "pub") + node.UID = "gateway-uid" + node.ResourceVersion = "1" + pool := unboundednetv1alpha1.GatewayPool{ + ObjectMeta: metav1.ObjectMeta{Name: "pool.raw"}, + Spec: unboundednetv1alpha1.GatewayPoolSpec{ + Type: gatewayPoolTypeExternal, + }, + } + client := kubefake.NewSimpleClientset(node) + + indexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{}) + if err := indexer.Add(node); err != nil { + t.Fatal(err) + } + + controller := &GatewayPoolController{ + clientset: client, + nodeLister: corev1listers.NewNodeLister(indexer), + gatewayPoolInformer: &fakeInformer{store: gatewayPoolStore(t, &pool)}, + } + + if err := controller.reconcileNodeGatewayPoolAssignments(t.Context()); err != nil { + t.Fatal(err) + } + + got, err := client.CoreV1().Nodes().Get(t.Context(), node.Name, metav1.GetOptions{}) + if err != nil { + t.Fatal(err) + } + + if got.Labels[unboundednetv1alpha1.GatewayPoolLabelKey] != pool.Name { + t.Fatalf("gateway labels = %#v, want raw pool name %q", got.Labels, pool.Name) + } + }) + + t.Run("invalid raw name fails explicitly", func(t *testing.T) { + node := makeGatewayNode("gateway", map[string]string{"role": "gateway"}, true, "pub") + node.UID = "gateway-uid" + node.ResourceVersion = "1" + pool := unboundednetv1alpha1.GatewayPool{ + ObjectMeta: metav1.ObjectMeta{Name: strings.Repeat("a", 64)}, + Spec: unboundednetv1alpha1.GatewayPoolSpec{ + Type: gatewayPoolTypeExternal, + NodeSelector: map[string]string{"role": "gateway"}, + }, + } + validNode := makeGatewayNode("valid-gateway", map[string]string{"role": "valid-gateway"}, true, "pub-valid") + validNode.UID = "valid-uid" + validNode.ResourceVersion = "1" + validPool := &unboundednetv1alpha1.GatewayPool{ + ObjectMeta: metav1.ObjectMeta{Name: "valid-pool"}, + Spec: unboundednetv1alpha1.GatewayPoolSpec{ + Type: gatewayPoolTypeExternal, + NodeSelector: map[string]string{"role": "valid-gateway"}, + }, + } + client := kubefake.NewSimpleClientset(node, validNode) + + indexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{}) + if err := indexer.Add(node); err != nil { + t.Fatal(err) + } + + if err := indexer.Add(validNode); err != nil { + t.Fatal(err) + } + + controller := &GatewayPoolController{ + clientset: client, + nodeLister: corev1listers.NewNodeLister(indexer), + gatewayPoolInformer: &fakeInformer{store: gatewayPoolStore(t, &pool, validPool)}, + } + controller.nodeLister = orderedNodeLister{ + NodeLister: controller.nodeLister, + nodes: []*corev1.Node{node, validNode}, + } + + err := controller.reconcileNodeGatewayPoolAssignments(t.Context()) + if err == nil || !strings.Contains(err.Error(), "not a valid Kubernetes label value") { + t.Fatalf("error = %v", err) + } + + got, getErr := client.CoreV1().Nodes().Get(t.Context(), node.Name, metav1.GetOptions{}) + if getErr != nil { + t.Fatal(getErr) + } + + if _, ok := got.Labels[unboundednetv1alpha1.GatewayPoolLabelKey]; ok { + t.Fatalf("invalid pool silently assigned: %#v", got.Labels) + } + + gotValid, err := client.CoreV1().Nodes().Get(t.Context(), validNode.Name, metav1.GetOptions{}) + if err != nil { + t.Fatal(err) + } + + if gotValid.Labels[unboundednetv1alpha1.GatewayPoolLabelKey] != validPool.Name { + t.Fatalf("valid assignment blocked by invalid pool: %#v", gotValid.Labels) + } + }) + + t.Run("same-name replacement is not modified", func(t *testing.T) { + cached := makeGatewayNode("gateway", map[string]string{"role": "gateway"}, true, "pub") + cached.UID = "old-uid" + cached.ResourceVersion = "1" + replacement := cached.DeepCopy() + replacement.UID = "replacement-uid" + replacement.ResourceVersion = "2" + replacement.Labels = map[string]string{"role": "worker", "example.com/preserved": "true"} + + pool := unboundednetv1alpha1.GatewayPool{ + ObjectMeta: metav1.ObjectMeta{Name: "pool.raw"}, + Spec: unboundednetv1alpha1.GatewayPoolSpec{ + Type: gatewayPoolTypeExternal, + NodeSelector: map[string]string{"role": "gateway"}, + }, + } + client := kubefake.NewSimpleClientset(replacement) + + indexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{}) + if err := indexer.Add(cached); err != nil { + t.Fatal(err) + } + + controller := &GatewayPoolController{ + clientset: client, + nodeLister: corev1listers.NewNodeLister(indexer), + gatewayPoolInformer: &fakeInformer{store: gatewayPoolStore(t, &pool)}, + } + + err := controller.reconcileNodeGatewayPoolAssignments(t.Context()) + if err == nil || !strings.Contains(err.Error(), "retry after informer refresh") { + t.Fatalf("error = %v, want cached identity precondition failure", err) + } + + got, getErr := client.CoreV1().Nodes().Get(t.Context(), replacement.Name, metav1.GetOptions{}) + if getErr != nil { + t.Fatal(getErr) + } + + if _, ok := got.Labels[unboundednetv1alpha1.GatewayPoolLabelKey]; ok { + t.Fatalf("replacement node was assigned from stale cache: %#v", got.Labels) + } + + if got.UID != replacement.UID || got.Labels["example.com/preserved"] != "true" { + t.Fatalf("replacement node metadata changed: %#v", got.ObjectMeta) + } + }) + + t.Run("changed resource version is not assigned from stale selector", func(t *testing.T) { + cached := makeGatewayNode("gateway", map[string]string{"role": "gateway"}, true, "pub") + cached.UID = "gateway-uid" + cached.ResourceVersion = "1" + fresh := cached.DeepCopy() + fresh.ResourceVersion = "2" + fresh.Labels = map[string]string{"role": "worker"} + + pool := unboundednetv1alpha1.GatewayPool{ + ObjectMeta: metav1.ObjectMeta{Name: "pool.raw"}, + Spec: unboundednetv1alpha1.GatewayPoolSpec{ + Type: gatewayPoolTypeExternal, + NodeSelector: map[string]string{"role": "gateway"}, + }, + } + client := kubefake.NewSimpleClientset(fresh) + + indexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{}) + if err := indexer.Add(cached); err != nil { + t.Fatal(err) + } + + controller := &GatewayPoolController{ + clientset: client, + nodeLister: corev1listers.NewNodeLister(indexer), + gatewayPoolInformer: &fakeInformer{store: gatewayPoolStore(t, &pool)}, + } + + err := controller.reconcileNodeGatewayPoolAssignments(t.Context()) + if err == nil || !strings.Contains(err.Error(), "retry after informer refresh") { + t.Fatalf("error = %v, want cached resourceVersion precondition failure", err) + } + + got, getErr := client.CoreV1().Nodes().Get(t.Context(), fresh.Name, metav1.GetOptions{}) + if getErr != nil { + t.Fatal(getErr) + } + + if _, ok := got.Labels[unboundednetv1alpha1.GatewayPoolLabelKey]; ok { + t.Fatalf("fresh node was assigned from stale selector: %#v", got.Labels) + } + }) +} + +func gatewayPoolStore(t *testing.T, pools ...*unboundednetv1alpha1.GatewayPool) cache.Store { + t.Helper() + + store := cache.NewStore(cache.MetaNamespaceKeyFunc) + for _, pool := range pools { + if err := store.Add(toGatewayPoolUnstructured(t, pool)); err != nil { + t.Fatal(err) + } + } + + return store +} diff --git a/internal/net/controller/node_assignment_controller.go b/internal/net/controller/node_assignment_controller.go new file mode 100644 index 0000000000..3d966e437a --- /dev/null +++ b/internal/net/controller/node_assignment_controller.go @@ -0,0 +1,290 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package controller + +import ( + "context" + "fmt" + + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + coreinformers "k8s.io/client-go/informers/core/v1" + "k8s.io/client-go/kubernetes" + corev1listers "k8s.io/client-go/listers/core/v1" + "k8s.io/client-go/tools/cache" + "k8s.io/client-go/util/workqueue" + "k8s.io/klog/v2" + + unboundednetv1alpha1 "github.com/Azure/unbounded/api/net/v1alpha1" +) + +const ( + NodeAgentPodNodeNameIndex = "unboundedNetNodeName" + nodeAgentDaemonSetName = "unbounded-net-node" +) + +// NodeAssignmentController replaces only node-agent Pods whose acknowledged +// effective assignment differs from the desired labels on their Node. +type NodeAssignmentController struct { + clientset kubernetes.Interface + + nodeLister corev1listers.NodeLister + nodeSynced cache.InformerSynced + podIndexer cache.Indexer + podSynced cache.InformerSynced + + queue workqueue.TypedRateLimitingInterface[string] +} + +func NewNodeAssignmentController( + clientset kubernetes.Interface, + nodeInformer coreinformers.NodeInformer, + podInformer coreinformers.PodInformer, +) (*NodeAssignmentController, error) { + controller := &NodeAssignmentController{ + clientset: clientset, + nodeLister: nodeInformer.Lister(), + nodeSynced: nodeInformer.Informer().HasSynced, + podIndexer: podInformer.Informer().GetIndexer(), + podSynced: podInformer.Informer().HasSynced, + queue: workqueue.NewTypedRateLimitingQueueWithConfig( + workqueue.DefaultTypedControllerRateLimiter[string](), + workqueue.TypedRateLimitingQueueConfig[string]{Name: "NodeAssignments"}, + ), + } + + if _, err := nodeInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ + AddFunc: func(obj interface{}) { + if node, ok := obj.(*corev1.Node); ok { + controller.queue.Add(node.Name) + } + }, + UpdateFunc: func(oldObj, newObj interface{}) { + oldNode, oldOK := oldObj.(*corev1.Node) + + newNode, newOK := newObj.(*corev1.Node) + if !oldOK || !newOK { + return + } + + if NodeSiteLabel(oldNode) != NodeSiteLabel(newNode) || + nodeGatewayPoolLabel(oldNode) != nodeGatewayPoolLabel(newNode) { + controller.queue.Add(newNode.Name) + } + }, + DeleteFunc: func(obj interface{}) { + if node := nodeFromDeleteEvent(obj); node != nil { + controller.queue.Add(node.Name) + } + }, + }); err != nil { + return nil, fmt.Errorf("add Node assignment event handler: %w", err) + } + + if _, err := podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ + AddFunc: controller.enqueuePod, + UpdateFunc: func(_, newObj interface{}) { controller.enqueuePod(newObj) }, + DeleteFunc: controller.enqueuePod, + }); err != nil { + return nil, fmt.Errorf("add node-agent Pod assignment event handler: %w", err) + } + + return controller, nil +} + +func NodeAgentPodNodeNameIndexFunc(obj interface{}) ([]string, error) { + pod, ok := obj.(*corev1.Pod) + if !ok || pod.Spec.NodeName == "" { + return nil, nil + } + + return []string{pod.Spec.NodeName}, nil +} + +func (c *NodeAssignmentController) Run(ctx context.Context, workers int) error { + defer c.queue.ShutDown() + + if !cache.WaitForCacheSync(ctx.Done(), c.nodeSynced, c.podSynced) { + return fmt.Errorf("sync Node assignment controller caches") + } + + for _, obj := range c.podIndexer.List() { + c.enqueuePod(obj) + } + + for i := 0; i < workers; i++ { + go c.runWorker(ctx) + } + + <-ctx.Done() + + return nil +} + +func (c *NodeAssignmentController) runWorker(ctx context.Context) { + for c.processNext(ctx) { + } +} + +func (c *NodeAssignmentController) processNext(ctx context.Context) bool { + nodeName, shutdown := c.queue.Get() + if shutdown { + return false + } + defer c.queue.Done(nodeName) + + if err := c.syncNode(ctx, nodeName); err != nil { + klog.Errorf("Failed to reconcile node-agent Pod assignment for Node %s: %v", nodeName, err) + c.queue.AddRateLimited(nodeName) + + return true + } + + c.queue.Forget(nodeName) + + return true +} + +func (c *NodeAssignmentController) syncNode(ctx context.Context, nodeName string) error { + node, err := c.nodeLister.Get(nodeName) + if apierrors.IsNotFound(err) { + return nil + } + + if err != nil { + return fmt.Errorf("get Node from cache: %w", err) + } + + podObjects, err := c.podIndexer.ByIndex(NodeAgentPodNodeNameIndex, nodeName) + if err != nil { + return fmt.Errorf("list node-agent Pods by Node: %w", err) + } + + desiredSite := NodeSiteLabel(node) + desiredGatewayPool := nodeGatewayPoolLabel(node) + + for _, obj := range podObjects { + pod, ok := obj.(*corev1.Pod) + if !ok || !isTargetNodeAgentPod(pod) || pod.DeletionTimestamp != nil { + continue + } + + effectiveSite, siteAcknowledged := pod.Annotations[unboundednetv1alpha1.EffectiveSiteAnnotation] + + effectiveGatewayPool, gatewayAcknowledged := pod.Annotations[unboundednetv1alpha1.EffectiveGatewayPoolAnnotation] + if !siteAcknowledged || !gatewayAcknowledged { + continue + } + + if effectiveSite == desiredSite && effectiveGatewayPool == desiredGatewayPool { + continue + } + + if err := c.deleteStaleAgentPod(ctx, nodeName, pod); err != nil { + return err + } + } + + return nil +} + +// deleteStaleAgentPod revalidates a cached assignment mismatch against the +// live Node and Pod before deleting. The Node and Pod informers are separate, +// so an agent may already have acknowledged a Node update that the cache has +// not observed. +func (c *NodeAssignmentController) deleteStaleAgentPod(ctx context.Context, nodeName string, cachedPod *corev1.Pod) error { + node, err := c.clientset.CoreV1().Nodes().Get(ctx, nodeName, metav1.GetOptions{}) + if apierrors.IsNotFound(err) { + return nil + } + + if err != nil { + return fmt.Errorf("get live Node %s before replacing node-agent Pod: %w", nodeName, err) + } + + pod, err := c.clientset.CoreV1().Pods(cachedPod.Namespace).Get(ctx, cachedPod.Name, metav1.GetOptions{}) + if apierrors.IsNotFound(err) { + return nil + } + + if err != nil { + return fmt.Errorf("get live node-agent Pod %s/%s: %w", cachedPod.Namespace, cachedPod.Name, err) + } + + if pod.UID != cachedPod.UID || pod.DeletionTimestamp != nil || !isTargetNodeAgentPod(pod) { + return nil + } + + effectiveSite, siteAcknowledged := pod.Annotations[unboundednetv1alpha1.EffectiveSiteAnnotation] + + effectiveGatewayPool, gatewayAcknowledged := pod.Annotations[unboundednetv1alpha1.EffectiveGatewayPoolAnnotation] + if !siteAcknowledged || !gatewayAcknowledged { + return nil + } + + desiredSite := NodeSiteLabel(node) + + desiredGatewayPool := nodeGatewayPoolLabel(node) + if effectiveSite == desiredSite && effectiveGatewayPool == desiredGatewayPool { + return nil + } + + uid, resourceVersion := pod.UID, pod.ResourceVersion + + propagation := metav1.DeletePropagationBackground + if err := c.clientset.CoreV1().Pods(pod.Namespace).Delete(ctx, pod.Name, metav1.DeleteOptions{ + Preconditions: &metav1.Preconditions{UID: &uid, ResourceVersion: &resourceVersion}, + PropagationPolicy: &propagation, + }); err != nil && !apierrors.IsNotFound(err) { + return fmt.Errorf("delete stale node-agent Pod %s/%s with UID %s: %w", pod.Namespace, pod.Name, pod.UID, err) + } + + klog.Infof("Deleted node-agent Pod %s/%s to adopt Node %s assignment change (Site %q -> %q, GatewayPool %q -> %q)", pod.Namespace, pod.Name, nodeName, effectiveSite, desiredSite, effectiveGatewayPool, desiredGatewayPool) + + return nil +} + +func (c *NodeAssignmentController) enqueuePod(obj interface{}) { + pod, ok := obj.(*corev1.Pod) + if !ok { + tombstone, tombstoneOK := obj.(cache.DeletedFinalStateUnknown) + if !tombstoneOK { + return + } + + pod, ok = tombstone.Obj.(*corev1.Pod) + if !ok { + return + } + } + + if pod.Spec.NodeName != "" { + c.queue.Add(pod.Spec.NodeName) + } +} + +func nodeGatewayPoolLabel(node *corev1.Node) string { + if node == nil || node.Labels == nil { + return "" + } + + return node.Labels[unboundednetv1alpha1.GatewayPoolLabelKey] +} + +func isTargetNodeAgentPod(pod *corev1.Pod) bool { + if pod == nil || + pod.Labels["app.kubernetes.io/name"] != nodeAgentDaemonSetName || + pod.Labels["app.kubernetes.io/component"] != "node" { + return false + } + + owner := metav1.GetControllerOf(pod) + + return owner != nil && + owner.APIVersion == appsv1.SchemeGroupVersion.String() && + owner.Kind == "DaemonSet" && + owner.Name == nodeAgentDaemonSetName +} diff --git a/internal/net/controller/node_assignment_controller_test.go b/internal/net/controller/node_assignment_controller_test.go new file mode 100644 index 0000000000..d85f6cbf5b --- /dev/null +++ b/internal/net/controller/node_assignment_controller_test.go @@ -0,0 +1,352 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package controller + +import ( + "testing" + + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + kubefake "k8s.io/client-go/kubernetes/fake" + corev1listers "k8s.io/client-go/listers/core/v1" + "k8s.io/client-go/tools/cache" + "k8s.io/client-go/util/workqueue" + + unboundedv1alpha3 "github.com/Azure/unbounded/api/machina/v1alpha3" + unboundednetv1alpha1 "github.com/Azure/unbounded/api/net/v1alpha1" +) + +func TestNodeAssignmentControllerRestartDecisions(t *testing.T) { + tests := []struct { + name string + annotations map[string]string + terminating bool + target bool + desiredSite string + desiredPool string + explicit bool + wantDeleted bool + }{ + { + name: "site mismatch", + annotations: map[string]string{ + unboundednetv1alpha1.EffectiveSiteAnnotation: "site-b", + unboundednetv1alpha1.EffectiveGatewayPoolAnnotation: "", + }, + target: true, + wantDeleted: true, + }, + { + name: "gateway promotion", + annotations: map[string]string{ + unboundednetv1alpha1.EffectiveSiteAnnotation: "site-a", + unboundednetv1alpha1.EffectiveGatewayPoolAnnotation: "", + }, + target: true, + wantDeleted: true, + }, + { + name: "gateway changed", + annotations: map[string]string{ + unboundednetv1alpha1.EffectiveSiteAnnotation: "site-a", + unboundednetv1alpha1.EffectiveGatewayPoolAnnotation: "pool-b", + }, + target: true, + wantDeleted: true, + }, + { + name: "gateway demotion", + annotations: map[string]string{ + unboundednetv1alpha1.EffectiveSiteAnnotation: "site-a", + unboundednetv1alpha1.EffectiveGatewayPoolAnnotation: "pool-a", + }, + target: true, + desiredSite: "site-a", + explicit: true, + wantDeleted: true, + }, + { + name: "site unset", + annotations: map[string]string{ + unboundednetv1alpha1.EffectiveSiteAnnotation: "site-a", + unboundednetv1alpha1.EffectiveGatewayPoolAnnotation: "", + }, + target: true, + explicit: true, + wantDeleted: true, + }, + { + name: "unchanged", + annotations: map[string]string{ + unboundednetv1alpha1.EffectiveSiteAnnotation: "site-a", + unboundednetv1alpha1.EffectiveGatewayPoolAnnotation: "pool-a", + }, + target: true, + }, + {name: "pending both", target: true}, + { + name: "pending gateway", + annotations: map[string]string{ + unboundednetv1alpha1.EffectiveSiteAnnotation: "site-a", + }, + target: true, + }, + { + name: "terminating mismatch", + annotations: map[string]string{ + unboundednetv1alpha1.EffectiveSiteAnnotation: "site-b", + unboundednetv1alpha1.EffectiveGatewayPoolAnnotation: "", + }, + target: true, + terminating: true, + }, + { + name: "unrelated pod", + annotations: map[string]string{ + unboundednetv1alpha1.EffectiveSiteAnnotation: "site-b", + unboundednetv1alpha1.EffectiveGatewayPoolAnnotation: "", + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + desiredSite := "site-a" + desiredPool := "pool-a" + + if tt.explicit { + desiredSite = tt.desiredSite + desiredPool = tt.desiredPool + } + + node := &corev1.Node{ObjectMeta: metav1.ObjectMeta{ + Name: "node-a", + Labels: map[string]string{}, + }} + if desiredSite != "" { + node.Labels[unboundedv1alpha3.MachineSiteLabelKey] = desiredSite + } + + if desiredPool != "" { + node.Labels[unboundednetv1alpha1.GatewayPoolLabelKey] = desiredPool + } + + pod := assignmentTestPod(tt.target, tt.annotations) + if tt.terminating { + now := metav1.Now() + pod.DeletionTimestamp = &now + } + + client := kubefake.NewClientset(pod.DeepCopy(), node.DeepCopy()) + controller := assignmentTestController(t, client, node, pod) + + if err := controller.syncNode(t.Context(), node.Name); err != nil { + t.Fatal(err) + } + + _, err := client.CoreV1().Pods(pod.Namespace).Get(t.Context(), pod.Name, metav1.GetOptions{}) + if tt.wantDeleted && err == nil { + t.Fatal("expected Pod deletion") + } + + if !tt.wantDeleted && err != nil { + t.Fatalf("Pod unexpectedly deleted: %v", err) + } + + if tt.wantDeleted { + options := podDeleteOptions(t, client) + if options.Preconditions == nil || options.Preconditions.UID == nil || *options.Preconditions.UID != pod.UID || + options.Preconditions.ResourceVersion == nil { + t.Fatalf("delete preconditions = %#v, want UID %s and a resourceVersion", options.Preconditions, pod.UID) + } + } + }) + } +} + +func TestNodeAssignmentControllerSkipsOldTerminatingPodAndPendingReplacement(t *testing.T) { + node := &corev1.Node{ObjectMeta: metav1.ObjectMeta{ + Name: "node-a", + Labels: map[string]string{unboundedv1alpha3.MachineSiteLabelKey: "site-b"}, + }} + oldPod := assignmentTestPod(true, map[string]string{ + unboundednetv1alpha1.EffectiveSiteAnnotation: "site-a", + unboundednetv1alpha1.EffectiveGatewayPoolAnnotation: "", + }) + now := metav1.Now() + oldPod.DeletionTimestamp = &now + newPod := assignmentTestPod(true, nil) + newPod.Name = "agent-new" + newPod.UID = "pod-new" + + client := kubefake.NewClientset(oldPod.DeepCopy(), newPod.DeepCopy()) + + controller := assignmentTestController(t, client, node, oldPod, newPod) + if err := controller.syncNode(t.Context(), node.Name); err != nil { + t.Fatal(err) + } + + for _, pod := range []*corev1.Pod{oldPod, newPod} { + if _, err := client.CoreV1().Pods(pod.Namespace).Get(t.Context(), pod.Name, metav1.GetOptions{}); err != nil { + t.Fatalf("Pod %s unexpectedly deleted: %v", pod.Name, err) + } + } +} + +func podDeleteOptions(t *testing.T, client *kubefake.Clientset) metav1.DeleteOptions { + t.Helper() + + for _, action := range client.Actions() { + if action.GetVerb() != "delete" { + continue + } + + deleteAction, ok := action.(interface{ GetDeleteOptions() metav1.DeleteOptions }) + if !ok { + t.Fatalf("delete action type = %T", action) + } + + return deleteAction.GetDeleteOptions() + } + + t.Fatal("no delete action recorded") + + return metav1.DeleteOptions{} +} + +func TestNodeAssignmentControllerRevalidatesStaleNodeCache(t *testing.T) { + ack := map[string]string{ + unboundednetv1alpha1.EffectiveSiteAnnotation: "site-b", + unboundednetv1alpha1.EffectiveGatewayPoolAnnotation: "", + } + cachedNode := &corev1.Node{ObjectMeta: metav1.ObjectMeta{ + Name: "node-a", + Labels: map[string]string{unboundedv1alpha3.MachineSiteLabelKey: "site-a"}, + }} + liveNode := cachedNode.DeepCopy() + liveNode.Labels[unboundedv1alpha3.MachineSiteLabelKey] = "site-b" + pod := assignmentTestPod(true, ack) + + client := kubefake.NewClientset(pod.DeepCopy(), liveNode) + + controller := assignmentTestController(t, client, cachedNode, pod) + if err := controller.syncNode(t.Context(), cachedNode.Name); err != nil { + t.Fatal(err) + } + + if _, err := client.CoreV1().Pods(pod.Namespace).Get(t.Context(), pod.Name, metav1.GetOptions{}); err != nil { + t.Fatalf("Pod acknowledging the live assignment was deleted: %v", err) + } + + for _, action := range client.Actions() { + if action.GetVerb() == "delete" { + t.Fatalf("unexpected delete action: %#v", action) + } + } +} + +func TestNodeAssignmentControllerRevalidatesStalePodCache(t *testing.T) { + node := &corev1.Node{ObjectMeta: metav1.ObjectMeta{ + Name: "node-a", + Labels: map[string]string{unboundedv1alpha3.MachineSiteLabelKey: "site-b"}, + }} + cachedPod := assignmentTestPod(true, map[string]string{ + unboundednetv1alpha1.EffectiveSiteAnnotation: "site-a", + unboundednetv1alpha1.EffectiveGatewayPoolAnnotation: "", + }) + livePod := cachedPod.DeepCopy() + livePod.Annotations = map[string]string{ + unboundednetv1alpha1.EffectiveSiteAnnotation: "site-b", + unboundednetv1alpha1.EffectiveGatewayPoolAnnotation: "", + } + + client := kubefake.NewClientset(livePod, node.DeepCopy()) + + controller := assignmentTestController(t, client, node, cachedPod) + if err := controller.syncNode(t.Context(), node.Name); err != nil { + t.Fatal(err) + } + + if _, err := client.CoreV1().Pods(livePod.Namespace).Get(t.Context(), livePod.Name, metav1.GetOptions{}); err != nil { + t.Fatalf("Pod with a current live acknowledgment was deleted: %v", err) + } +} + +func TestNodeAssignmentControllerSkipsDeletedLiveNode(t *testing.T) { + node := &corev1.Node{ObjectMeta: metav1.ObjectMeta{ + Name: "node-a", + Labels: map[string]string{unboundedv1alpha3.MachineSiteLabelKey: "site-b"}, + }} + pod := assignmentTestPod(true, map[string]string{ + unboundednetv1alpha1.EffectiveSiteAnnotation: "site-a", + unboundednetv1alpha1.EffectiveGatewayPoolAnnotation: "", + }) + + client := kubefake.NewClientset(pod.DeepCopy()) + + controller := assignmentTestController(t, client, node, pod) + if err := controller.syncNode(t.Context(), node.Name); err != nil { + t.Fatal(err) + } + + if _, err := client.CoreV1().Pods(pod.Namespace).Get(t.Context(), pod.Name, metav1.GetOptions{}); err != nil { + t.Fatalf("Pod deleted after its Node was deleted: %v", err) + } +} + +func assignmentTestPod(target bool, annotations map[string]string) *corev1.Pod { + controller := true + + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "system", + Name: "agent", + UID: types.UID("pod-uid"), + Annotations: annotations, + }, + Spec: corev1.PodSpec{NodeName: "node-a"}, + } + if target { + pod.Labels = map[string]string{ + "app.kubernetes.io/name": nodeAgentDaemonSetName, + "app.kubernetes.io/component": "node", + } + pod.OwnerReferences = []metav1.OwnerReference{{ + APIVersion: appsv1.SchemeGroupVersion.String(), + Kind: "DaemonSet", + Name: nodeAgentDaemonSetName, + UID: "ds-uid", + Controller: &controller, + }} + } + + return pod +} + +func assignmentTestController(t *testing.T, client *kubefake.Clientset, node *corev1.Node, pods ...*corev1.Pod) *NodeAssignmentController { + t.Helper() + + nodeIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{}) + if err := nodeIndexer.Add(node); err != nil { + t.Fatal(err) + } + + podIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{ + NodeAgentPodNodeNameIndex: NodeAgentPodNodeNameIndexFunc, + }) + for _, pod := range pods { + if err := podIndexer.Add(pod); err != nil { + t.Fatal(err) + } + } + + return &NodeAssignmentController{ + clientset: client, + nodeLister: corev1listers.NewNodeLister(nodeIndexer), + podIndexer: podIndexer, + queue: workqueue.NewTypedRateLimitingQueue(workqueue.DefaultTypedControllerRateLimiter[string]()), + } +} diff --git a/internal/operator/components/net/net.go b/internal/operator/components/net/net.go index eed018977f..c4a5e14d77 100644 --- a/internal/operator/components/net/net.go +++ b/internal/operator/components/net/net.go @@ -16,6 +16,7 @@ import ( corev1 "k8s.io/api/core/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime/schema" "sigs.k8s.io/controller-runtime/pkg/builder" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/event" @@ -113,6 +114,28 @@ func (c Component) Plan(ctx context.Context, env *component.Env, sites []unbound op.DependsOn = dependsOn } + if obj.GetKind() == "Role" && obj.GetNamespace() == env.Namespace { + policy := "" + + switch obj.GetName() { + case controllerName: + policy = "unbounded-net-node-pod-delete-restriction" + case nodeName: + policy = "unbounded-net-effective-assignment-ownership" + } + + if policy != "" { + // These grants rely on admission to constrain namespace-wide + // Pod verbs. Failed policy installation must block the grant. + for _, kind := range []string{"ValidatingAdmissionPolicy", "ValidatingAdmissionPolicyBinding"} { + op.DependsOn = append(op.DependsOn, component.ObjectRef{ + GVK: schema.GroupVersionKind{Group: "admissionregistration.k8s.io", Version: "v1", Kind: kind}, + Name: policy, + }) + } + } + } + plan.Add(op) } diff --git a/internal/operator/components/net/net_test.go b/internal/operator/components/net/net_test.go index d4a68c7d72..6c28fd50cb 100644 --- a/internal/operator/components/net/net_test.go +++ b/internal/operator/components/net/net_test.go @@ -1458,7 +1458,7 @@ Apply ServiceAccount/unbounded-system/unbounded-net-kube-proxy Apply ClusterRole/unbounded-net-controller Apply ClusterRoleBinding/unbounded-net-kube-proxy Apply ClusterRoleBinding/unbounded-net-controller -Apply Role/unbounded-system/unbounded-net-controller +Apply Role/unbounded-system/unbounded-net-controller [after ValidatingAdmissionPolicy/unbounded-net-node-pod-delete-restriction ValidatingAdmissionPolicyBinding/unbounded-net-node-pod-delete-restriction] Apply RoleBinding/unbounded-system/unbounded-net-controller Apply Role/kube-system/unbounded-net-controller Apply RoleBinding/kube-system/unbounded-net-controller @@ -1471,12 +1471,20 @@ Apply ValidatingAdmissionPolicy/unbounded-net-create-restriction Apply ValidatingAdmissionPolicyBinding/unbounded-net-create-restriction Apply ValidatingAdmissionPolicy/unbounded-net-node-field-restriction Apply ValidatingAdmissionPolicyBinding/unbounded-net-node-field-restriction +Apply ValidatingAdmissionPolicy/unbounded-net-gateway-pool-name +Apply ValidatingAdmissionPolicyBinding/unbounded-net-gateway-pool-name Apply ValidatingAdmissionPolicy/unbounded-net-public-ip-annotation-ownership Apply ValidatingAdmissionPolicyBinding/unbounded-net-public-ip-annotation-ownership +Apply ValidatingAdmissionPolicy/unbounded-net-effective-assignment-ownership +Apply ValidatingAdmissionPolicyBinding/unbounded-net-effective-assignment-ownership +Apply ValidatingAdmissionPolicy/unbounded-net-node-pod-delete-restriction +Apply ValidatingAdmissionPolicyBinding/unbounded-net-node-pod-delete-restriction Apply ClusterRole/unbounded-net-status-viewer Apply ServiceAccount/unbounded-system/unbounded-net-node Apply ClusterRole/unbounded-net-node Apply ClusterRoleBinding/unbounded-net-node +Apply Role/unbounded-system/unbounded-net-node [after ValidatingAdmissionPolicy/unbounded-net-effective-assignment-ownership ValidatingAdmissionPolicyBinding/unbounded-net-effective-assignment-ownership] +Apply RoleBinding/unbounded-system/unbounded-net-node Apply DaemonSet/unbounded-system/unbounded-net-node [overridable] [after ConfigMap/unbounded-system/unbounded-net-config] ` @@ -1511,7 +1519,7 @@ Apply ServiceAccount/unbounded-system/unbounded-net-kube-proxy Apply ClusterRole/unbounded-net-controller Apply ClusterRoleBinding/unbounded-net-kube-proxy Apply ClusterRoleBinding/unbounded-net-controller -Apply Role/unbounded-system/unbounded-net-controller +Apply Role/unbounded-system/unbounded-net-controller [after ValidatingAdmissionPolicy/unbounded-net-node-pod-delete-restriction ValidatingAdmissionPolicyBinding/unbounded-net-node-pod-delete-restriction] Apply RoleBinding/unbounded-system/unbounded-net-controller Apply Role/kube-system/unbounded-net-controller Apply RoleBinding/kube-system/unbounded-net-controller @@ -1521,12 +1529,20 @@ Apply ValidatingAdmissionPolicy/unbounded-net-create-restriction Apply ValidatingAdmissionPolicyBinding/unbounded-net-create-restriction Apply ValidatingAdmissionPolicy/unbounded-net-node-field-restriction Apply ValidatingAdmissionPolicyBinding/unbounded-net-node-field-restriction +Apply ValidatingAdmissionPolicy/unbounded-net-gateway-pool-name +Apply ValidatingAdmissionPolicyBinding/unbounded-net-gateway-pool-name Apply ValidatingAdmissionPolicy/unbounded-net-public-ip-annotation-ownership Apply ValidatingAdmissionPolicyBinding/unbounded-net-public-ip-annotation-ownership +Apply ValidatingAdmissionPolicy/unbounded-net-effective-assignment-ownership +Apply ValidatingAdmissionPolicyBinding/unbounded-net-effective-assignment-ownership +Apply ValidatingAdmissionPolicy/unbounded-net-node-pod-delete-restriction +Apply ValidatingAdmissionPolicyBinding/unbounded-net-node-pod-delete-restriction Apply ClusterRole/unbounded-net-status-viewer Apply ServiceAccount/unbounded-system/unbounded-net-node Apply ClusterRole/unbounded-net-node Apply ClusterRoleBinding/unbounded-net-node +Apply Role/unbounded-system/unbounded-net-node [after ValidatingAdmissionPolicy/unbounded-net-effective-assignment-ownership ValidatingAdmissionPolicyBinding/unbounded-net-effective-assignment-ownership] +Apply RoleBinding/unbounded-system/unbounded-net-node Apply DaemonSet/unbounded-system/unbounded-net-node [overridable] [after ConfigMap/unbounded-system/unbounded-net-config] ` @@ -1552,6 +1568,66 @@ Apply DaemonSet/unbounded-system/unbounded-net-node [overridable] [after ConfigM // own, and it is the half of the problem the backend gate does not solve: // ordering says "after the apply returned", the gate says "after it is // serving". +func TestPodPermissionGrantsRequirePolicies(t *testing.T) { + for _, policy := range []string{"unbounded-net-effective-assignment-ownership", "unbounded-net-node-pod-delete-restriction"} { + for _, kind := range []string{"ValidatingAdmissionPolicy", "ValidatingAdmissionPolicyBinding"} { + t.Run(kind+"/"+policy, func(t *testing.T) { + env := testEnv(t, servingObjects()...) + applied := map[string]bool{} + failure := errors.New("policy installation failed") + env.Client = interceptor.NewClient(fake.NewClientBuilder().WithScheme(newNetTestScheme(t)).WithObjects(servingObjects()...).Build(), interceptor.Funcs{ + Apply: func(_ context.Context, _ client.WithWatch, obj runtime.ApplyConfiguration, _ ...client.ApplyOption) error { + object := obj.(interface { + GetName() string + GetKind() string + GetNamespace() string + }) + key := object.GetKind() + "/" + object.GetName() + applied[object.GetKind()+"/"+object.GetNamespace()+"/"+object.GetName()] = true + + if key == kind+"/"+policy { + return failure + } + + return nil + }, + }) + + plan, _, err := (Component{}).Plan(t.Context(), env, site()) + if err != nil { + t.Fatal(err) + } + + result, err := env.Execute(t.Context(), plan) + if err != nil { + t.Fatal(err) + } + + role := controllerName + if policy == "unbounded-net-effective-assignment-ownership" { + role = nodeName + } + + if applied["Role/"+env.Namespace+"/"+role] { + t.Fatal("expanded Role applied despite failed restricting policy or binding") + } + + found := false + + for _, op := range result.Results { + if errors.Is(op.Err, failure) { + found = true + } + } + + if !found { + t.Fatal("policy installation failure was not surfaced") + } + }) + } + } +} + func TestExecutionOrderGolden(t *testing.T) { env := testEnv(t, servingObjects()...) @@ -1565,7 +1641,6 @@ Apply ServiceAccount/unbounded-system/unbounded-net-kube-proxy Apply ClusterRole/unbounded-net-controller Apply ClusterRoleBinding/unbounded-net-kube-proxy Apply ClusterRoleBinding/unbounded-net-controller -Apply Role/unbounded-system/unbounded-net-controller Apply RoleBinding/unbounded-system/unbounded-net-controller Apply Role/kube-system/unbounded-net-controller Apply RoleBinding/kube-system/unbounded-net-controller @@ -1573,6 +1648,7 @@ Apply ClusterRole/unbounded-net-status-viewer Apply ServiceAccount/unbounded-system/unbounded-net-node Apply ClusterRole/unbounded-net-node Apply ClusterRoleBinding/unbounded-net-node +Apply RoleBinding/unbounded-system/unbounded-net-node CreateIfAbsent ConfigMap/unbounded-system/unbounded-net-config Apply Service/unbounded-system/unbounded-net-controller Apply Deployment/unbounded-system/unbounded-net-controller @@ -1584,8 +1660,16 @@ Apply ValidatingAdmissionPolicy/unbounded-net-create-restriction Apply ValidatingAdmissionPolicyBinding/unbounded-net-create-restriction Apply ValidatingAdmissionPolicy/unbounded-net-node-field-restriction Apply ValidatingAdmissionPolicyBinding/unbounded-net-node-field-restriction +Apply ValidatingAdmissionPolicy/unbounded-net-gateway-pool-name +Apply ValidatingAdmissionPolicyBinding/unbounded-net-gateway-pool-name Apply ValidatingAdmissionPolicy/unbounded-net-public-ip-annotation-ownership Apply ValidatingAdmissionPolicyBinding/unbounded-net-public-ip-annotation-ownership +Apply ValidatingAdmissionPolicy/unbounded-net-effective-assignment-ownership +Apply ValidatingAdmissionPolicyBinding/unbounded-net-effective-assignment-ownership +Apply Role/unbounded-system/unbounded-net-node +Apply ValidatingAdmissionPolicy/unbounded-net-node-pod-delete-restriction +Apply ValidatingAdmissionPolicyBinding/unbounded-net-node-pod-delete-restriction +Apply Role/unbounded-system/unbounded-net-controller ` got, err := plan.ExecutionOrder() diff --git a/internal/operator/components/net/vap_test.go b/internal/operator/components/net/vap_test.go index 9c6efbe72d..37fc7bb3e1 100644 --- a/internal/operator/components/net/vap_test.go +++ b/internal/operator/components/net/vap_test.go @@ -5,6 +5,7 @@ package net import ( "reflect" + "strings" "testing" "github.com/google/cel-go/cel" @@ -14,6 +15,150 @@ import ( unboundedv1alpha3 "github.com/Azure/unbounded/api/machina/v1alpha3" ) +func TestNodeFieldPolicyAssignmentOwnership(t *testing.T) { + t.Parallel() + + policy := plannedNetObject(t, "ValidatingAdmissionPolicy", "unbounded-net-node-field-restriction") + + tests := []struct { + name string + username string + nodeName string + nodeUID string + key string + annotation bool + unbound bool + wantAllowed bool + }{ + {name: "own gateway label", key: "net.unbounded-cloud.io/gateway-pool"}, + {name: "own legacy Site label", key: "net.unbounded-cloud.io/site"}, + {name: "own canonical Site label", key: "unbounded-cloud.io/site"}, + {name: "foreign gateway label", nodeName: "other", key: "net.unbounded-cloud.io/gateway-pool"}, + {name: "foreign Site label", nodeName: "other", key: "unbounded-cloud.io/site"}, + {name: "foreign public key", nodeName: "other", key: "net.unbounded-cloud.io/wg-pubkey", annotation: true}, + {name: "wrong UID", nodeUID: "other-uid", key: "net.unbounded-cloud.io/wg-pubkey", annotation: true}, + {name: "unbound token", unbound: true, key: "net.unbounded-cloud.io/wg-pubkey", annotation: true}, + {name: "own public key", key: "net.unbounded-cloud.io/wg-pubkey", annotation: true, wantAllowed: true}, + {name: "own MTU", key: "net.unbounded-cloud.io/tunnel-mtu", annotation: true, wantAllowed: true}, + {name: "own discovered IP", key: discoveredPublicIPAnnotation, annotation: true, wantAllowed: true}, + {name: "own discovery expiry", key: discoveredPublicIPExpiresAtAnnotation, annotation: true, wantAllowed: true}, + {name: "own unknown annotation", key: "net.unbounded-cloud.io/arbitrary", annotation: true}, + {name: "controller gateway label", username: controllerServiceAccountUsername, key: "net.unbounded-cloud.io/gateway-pool", wantAllowed: true}, + {name: "controller Site label", username: controllerServiceAccountUsername, key: "unbounded-cloud.io/site", wantAllowed: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + username := tt.username + if username == "" { + username = nodeServiceAccountUsername + } + + name, uid := tt.nodeName, tt.nodeUID + if name == "" { + name = "node-a" + } + + if uid == "" { + uid = "node-uid" + } + + for _, operation := range []string{"add", "change", "remove"} { + t.Run(operation, func(t *testing.T) { + oldObject, object := nodeSpecAdmissionObject(nil), nodeSpecAdmissionObject(nil) + + field := "labels" + if tt.annotation { + field = "annotations" + } + + oldValues, values := map[string]string{}, map[string]string{} + if operation != "add" { + oldValues[tt.key] = "old" + } + + if operation != "remove" { + values[tt.key] = "new" + } + + oldObject["metadata"].(map[string]any)[field] = oldValues + object["metadata"].(map[string]any)[field] = values + + activation := map[string]any{ + "oldObject": oldObject, "object": object, + "request": map[string]any{"userInfo": map[string]any{ + "username": username, + "extra": map[string][]string{nodeNameExtra: {name}, nodeUIDExtra: {uid}}, + }}, + } + if tt.unbound { + activation["request"] = map[string]any{"userInfo": map[string]any{ + "username": username, + "extra": map[string][]string{}, + }} + } + + if got := policyAllows(t, policy, activation); got != tt.wantAllowed { + t.Fatalf("allowed=%v, want %v", got, tt.wantAllowed) + } + }) + } + }) + } +} + +func TestNodeFieldPolicyPreservesControllerPodCIDRs(t *testing.T) { + t.Parallel() + policy := plannedNetObject(t, "ValidatingAdmissionPolicy", "unbounded-net-node-field-restriction") + expression := policyValidationExpression(t, policy, "unbounded-net-node may not modify controller-owned pod CIDRs") + + for _, field := range []string{"podCIDR", "podCIDRs"} { + for _, username := range []string{nodeServiceAccountUsername, controllerServiceAccountUsername} { + oldObject, object := nodeSpecAdmissionObject(nil), nodeSpecAdmissionObject(nil) + + spec := object["spec"].(map[string]any) + if field == "podCIDR" { + spec[field] = "10.244.0.0/24" + } else { + spec[field] = []string{"10.244.0.0/24"} + } + + allowed := evalCEL(t, expression, map[string]any{ + "object": object, "oldObject": oldObject, + "request": map[string]any{"userInfo": map[string]any{"username": username}}, + }) + if allowed != (username == controllerServiceAccountUsername) { + t.Fatalf("%s mutation by %s allowed=%v", field, username, allowed) + } + } + } +} + +func TestGatewayPoolNamePolicy(t *testing.T) { + t.Parallel() + + policy := plannedNetObject(t, "ValidatingAdmissionPolicy", "unbounded-net-gateway-pool-name") + + rules, found, err := unstructured.NestedSlice(policy.Object, "spec", "matchConstraints", "resourceRules") + if err != nil || !found || len(rules) != 1 { + t.Fatalf("rules=%v err=%v", rules, err) + } + + rule := rules[0].(map[string]any) + assertStringSliceField(t, rule, []string{"gatewaypools"}, "resources") + assertStringSliceField(t, rule, []string{"CREATE"}, "operations") + assertStringSliceField(t, rule, []string{"net.unbounded-cloud.io"}, "apiGroups") + + for _, name := range []string{"pool.raw", strings.Repeat("a", 63), strings.Repeat("a", 64)} { + got := policyAllows(t, policy, map[string]any{"object": map[string]any{"metadata": map[string]any{"name": name}}}) + if got != (len(name) <= 63) { + t.Fatalf("name length=%d allowed=%v", len(name), got) + } + } + + binding := plannedNetObject(t, "ValidatingAdmissionPolicyBinding", "unbounded-net-gateway-pool-name") + assertStringSliceField(t, binding.Object["spec"].(map[string]any), []string{"Deny"}, "validationActions") +} + const ( wireGuardPortAnnotation = "net.unbounded-cloud.io/wireguard-port" discoveredPublicIPAnnotation = "net.unbounded-cloud.io/discovered-public-ip" @@ -93,6 +238,74 @@ func TestNodeFieldPolicyProtectsWireGuardPort(t *testing.T) { } } +func TestNodeFieldPolicyRestrictsGatewayTaint(t *testing.T) { + t.Parallel() + + policy := plannedNetObject(t, "ValidatingAdmissionPolicy", "unbounded-net-node-field-restriction") + specExpression := policyValidationExpression(t, policy, "unbounded-net may only modify pod CIDRs and the node agent's gateway taint") + taintExpression := policyValidationExpression(t, policy, "unbounded-net-node may only add or remove its exact gateway taint on its own Node") + request := func(nodeName string) map[string]any { + return map[string]any{"userInfo": map[string]any{ + "username": nodeServiceAccountUsername, + "extra": map[string][]string{ + nodeNameExtra: {nodeName}, + nodeUIDExtra: {"node-uid"}, + }, + }} + } + + oldObject := nodeSpecAdmissionObject(nil) + exact := nodeSpecAdmissionObject([]map[string]any{{ + "key": "net.unbounded-cloud.io/gateway-node", + "value": "true", + "effect": "NoSchedule", + }}) + + activation := map[string]any{"oldObject": oldObject, "object": exact, "request": request("node-a")} + if !evalCEL(t, specExpression, activation) || !evalCEL(t, taintExpression, activation) { + t.Fatal("exact own-Node gateway taint was denied") + } + + wrong := nodeSpecAdmissionObject([]map[string]any{{ + "key": "net.unbounded-cloud.io/gateway-node", + "value": "false", + "effect": "NoSchedule", + }}) + if evalCEL(t, taintExpression, map[string]any{"oldObject": oldObject, "object": wrong, "request": request("node-a")}) { + t.Fatal("wrong gateway taint value was allowed") + } + + otherOld := nodeSpecAdmissionObject([]map[string]any{{"key": "other", "effect": "NoSchedule"}}) + + otherChanged := nodeSpecAdmissionObject([]map[string]any{{"key": "changed", "effect": "NoSchedule"}}) + if evalCEL(t, taintExpression, map[string]any{"oldObject": otherOld, "object": otherChanged, "request": request("node-a")}) { + t.Fatal("change to an unrelated taint was allowed") + } + + if evalCEL(t, taintExpression, map[string]any{"oldObject": oldObject, "object": exact, "request": request("node-b")}) { + t.Fatal("cross-Node gateway taint was allowed") + } + + for _, taint := range []map[string]any{ + {"key": "net.unbounded-cloud.io/gateway-node", "value": "true", "effect": "NoExecute"}, + {"key": "net.unbounded-cloud.io/gateway-node", "value": "false", "effect": "NoSchedule"}, + {"key": "net.unbounded-cloud.io/gateway-node", "effect": "NoSchedule"}, + } { + protected := nodeSpecAdmissionObject([]map[string]any{taint}) + if !evalCEL(t, taintExpression, map[string]any{"oldObject": protected, "object": protected, "request": request("node-a")}) { + t.Fatalf("unchanged non-owned taint denied: %v", taint) + } + + if evalCEL(t, taintExpression, map[string]any{"oldObject": protected, "object": oldObject, "request": request("node-a")}) { + t.Fatalf("non-owned taint removal allowed: %v", taint) + } + + if evalCEL(t, taintExpression, map[string]any{"oldObject": protected, "object": exact, "request": request("node-a")}) { + t.Fatalf("non-owned taint replacement allowed: %v", taint) + } + } +} + func TestPublicIPAnnotationOwnershipPolicyScope(t *testing.T) { t.Parallel() @@ -365,6 +578,144 @@ func TestPublicIPAnnotationOwnershipPolicy(t *testing.T) { } } +func TestEffectiveAssignmentOwnershipPolicy(t *testing.T) { + t.Parallel() + + policy := plannedNetObject(t, "ValidatingAdmissionPolicy", "unbounded-net-effective-assignment-ownership") + ownPodExpression := policyValidationExpression(t, policy, "unbounded-net-node may only acknowledge assignment on its own bound Pod") + fieldExpression := policyValidationExpression(t, policy, "unbounded-net-node may only modify effective assignment annotations on its own Pod") + annotationExpression := policyValidationExpression(t, policy, "unbounded-net-node may only modify effective assignment annotations") + + oldObject := podAdmissionObject("agent", "pod-uid", map[string]string{"existing": "keep"}) + updated := podAdmissionObject("agent", "pod-uid", map[string]string{ + "existing": "keep", + "net.unbounded-cloud.io/effective-site": "site-a", + "net.unbounded-cloud.io/effective-gateway-pool": "", + }) + request := map[string]any{"userInfo": map[string]any{"extra": map[string][]string{ + "authentication.kubernetes.io/pod-name": {"agent"}, + "authentication.kubernetes.io/pod-uid": {"pod-uid"}, + }}} + + if !evalCEL(t, ownPodExpression, map[string]any{"oldObject": oldObject, "object": updated, "request": request}) { + t.Fatal("own bound Pod was denied") + } + + if !evalCEL(t, annotationExpression, map[string]any{"oldObject": oldObject, "object": updated, "request": request}) { + t.Fatal("effective assignment annotation update was denied") + } + + if !evalCEL(t, fieldExpression, map[string]any{"oldObject": oldObject, "object": updated, "request": request}) { + t.Fatal("unchanged Pod fields were denied") + } + + crossPodRequest := map[string]any{"userInfo": map[string]any{"extra": map[string][]string{ + "authentication.kubernetes.io/pod-name": {"other"}, + "authentication.kubernetes.io/pod-uid": {"pod-uid"}, + }}} + if evalCEL(t, ownPodExpression, map[string]any{"oldObject": oldObject, "object": updated, "request": crossPodRequest}) { + t.Fatal("cross-Pod acknowledgment was allowed") + } + + changedUnowned := podAdmissionObject("agent", "pod-uid", map[string]string{ + "existing": "changed", + "net.unbounded-cloud.io/effective-site": "site-a", + "net.unbounded-cloud.io/effective-gateway-pool": "", + }) + if evalCEL(t, annotationExpression, map[string]any{"oldObject": oldObject, "object": changedUnowned, "request": request}) { + t.Fatal("unowned annotation change was allowed") + } +} + +func TestEffectiveAssignmentPolicyOptionalPodFields(t *testing.T) { + t.Parallel() + + policy := plannedNetObject(t, "ValidatingAdmissionPolicy", "unbounded-net-effective-assignment-ownership") + expression := policyValidationExpression(t, policy, "unbounded-net-node may only modify effective assignment annotations on its own Pod") + + for _, field := range []string{"labels", "ownerReferences", "finalizers", "status", "all"} { + t.Run(field, func(t *testing.T) { + t.Parallel() + + for _, change := range []string{"unchanged", "added", "removed"} { + t.Run(change, func(t *testing.T) { + t.Parallel() + + oldObject := podAdmissionObject("agent", "pod-uid", map[string]string{"existing": "keep"}) + updated := podAdmissionObject("agent", "pod-uid", map[string]string{ + "existing": "keep", + "net.unbounded-cloud.io/effective-site": "site-a", + "net.unbounded-cloud.io/effective-gateway-pool": "", + }) + + fields := []string{field} + if field == "all" { + fields = []string{"labels", "ownerReferences", "finalizers", "status"} + } + + for _, omitted := range fields { + oldFields := oldObject + newFields := updated + + if omitted != "status" { + oldFields = oldObject["metadata"].(map[string]any) + newFields = updated["metadata"].(map[string]any) + } + + if change != "removed" { + delete(oldFields, omitted) + } + + if change != "added" { + delete(newFields, omitted) + } + } + + allowed := evalCEL(t, expression, map[string]any{"oldObject": oldObject, "object": updated}) + if allowed != (change == "unchanged") { + t.Fatalf("allowed = %v for %s optional field %s", allowed, change, field) + } + }) + } + }) + } +} + +func TestNodePodDeleteRestrictionPolicy(t *testing.T) { + t.Parallel() + + policy := plannedNetObject(t, "ValidatingAdmissionPolicy", "unbounded-net-node-pod-delete-restriction") + expression := policyValidationExpression(t, policy, "unbounded-net controller may only delete DaemonSet-owned unbounded-net-node Pods") + request := map[string]any{"userInfo": map[string]any{}} + + target := podAdmissionObject("agent", "pod-uid", nil) + target["metadata"].(map[string]any)["namespace"] = "unbounded-system" + target["metadata"].(map[string]any)["labels"] = map[string]string{ + "app.kubernetes.io/name": "unbounded-net-node", + "app.kubernetes.io/component": "node", + } + target["metadata"].(map[string]any)["ownerReferences"] = []map[string]any{{ + "apiVersion": "apps/v1", + "kind": "DaemonSet", + "name": "unbounded-net-node", + "controller": true, + }} + + if !evalCEL(t, expression, map[string]any{"oldObject": target, "object": nil, "request": request}) { + t.Fatal("target node-agent Pod deletion was denied") + } + + target["metadata"].(map[string]any)["ownerReferences"] = []map[string]any{{ + "apiVersion": "apps/v1", + "kind": "Deployment", + "name": "unbounded-net-node", + "controller": true, + }} + if evalCEL(t, expression, map[string]any{"oldObject": target, "object": nil, "request": request}) { + t.Fatal("non-DaemonSet Pod deletion was allowed") + } +} + func plannedNetObject(t *testing.T, kind, name string) *unstructured.Unstructured { t.Helper() @@ -454,6 +805,37 @@ func publicIPNodeAdmissionObject(name, uid string, annotations map[string]string return map[string]any{"metadata": metadata} } +func podAdmissionObject(name, uid string, annotations map[string]string) map[string]any { + metadata := map[string]any{ + "name": name, + "uid": uid, + "labels": map[string]string{}, + "ownerReferences": []map[string]any{}, + "finalizers": []string{}, + } + if annotations != nil { + metadata["annotations"] = annotations + } + + return map[string]any{ + "metadata": metadata, + "spec": map[string]any{}, + "status": map[string]any{}, + } +} + +func nodeSpecAdmissionObject(taints []map[string]any) map[string]any { + if taints == nil { + taints = []map[string]any{} + } + + return map[string]any{ + "metadata": map[string]any{"name": "node-a", "uid": "node-uid"}, + "spec": map[string]any{"taints": taints}, + "status": map[string]any{}, + } +} + func evalCEL(t *testing.T, expression string, activation map[string]any) bool { t.Helper()