connectivity: Add Control Plane Node Connectivity Tests

The control plane nodes create unique policy selection
contexts that allow us to test that label selection of
host and kube-apiserver entities is correct.

Signed-off-by: Nate Sweet <[email protected]>

connectivity/check/check.go

@@ -28,6 +28,7 @@ type Parameters struct {
 	ForceDeploy           bool
 	Hubble                bool
 	HubbleServer          string
+	K8sLocalHostTest      bool
 	MultiCluster          string
 	RunTests              []*regexp.Regexp
 	SkipTests             []*regexp.Regexp
@@ -61,6 +62,8 @@ type Parameters struct {
 	ExternalOtherIP       string
 	PodCIDRs              []podCIDRs
 	NodeCIDRs             []string
+	ControlPlaneCIDRs     []string
+	K8sCIDR               string
 	NodesWithoutCiliumIPs []nodesWithoutCiliumIP
 	JunitFile             string
 	JunitProperties       map[string]string

connectivity/check/context.go

@@ -61,11 +61,13 @@ type ConnectivityTest struct {
 	echoPods          map[string]Pod
 	echoExternalPods  map[string]Pod
 	clientPods        map[string]Pod
+	clientCPPod       *Pod
 	perfClientPods    map[string]Pod
 	perfServerPod     map[string]Pod
 	PerfResults       map[PerfTests]PerfResult
 	echoServices      map[string]Service
 	ingressService    map[string]Service
+	k8sService        Service
 	externalWorkloads map[string]ExternalWorkload
 
 	hostNetNSPodsByNode      map[string]Pod
@@ -98,6 +100,29 @@ type PerfResult struct {
 	Avg      float64
 }
 
+func netIPToCIDRs(netIPs []netip.Addr) (netCIDRs []netip.Prefix) {
+	for _, ip := range netIPs {
+		found := false
+		for _, cidr := range netCIDRs {
+			if cidr.Addr().Is4() == ip.Is4() && cidr.Contains(ip) {
+				found = true
+				break
+			}
+		}
+		if found {
+			continue
+		}
+
+		// Generate a /24 or /64 accordingly
+		bits := 24
+		if ip.Is6() {
+			bits = 64
+		}
+		netCIDRs = append(netCIDRs, netip.PrefixFrom(ip, bits).Masked())
+	}
+	return
+}
+
 // verbose returns the value of the user-provided verbosity flag.
 func (ct *ConnectivityTest) verbose() bool {
 	return ct.params.Verbose
@@ -336,6 +361,11 @@ func (ct *ConnectivityTest) SetupAndValidate(ctx context.Context, setupAndValida
 			return fmt.Errorf("unable to detect node CIDRs: %w", err)
 		}
 	}
+	if ct.params.K8sLocalHostTest {
+		if err := ct.detectK8sCIDR(ctx); err != nil {
+			return fmt.Errorf("unable to detect K8s CIDR: %w", err)
+		}
+	}
 	return nil
 }
 
@@ -639,6 +669,7 @@ func (ct *ConnectivityTest) detectNodeCIDRs(ctx context.Context) error {
 	}
 
 	nodeIPs := make([]netip.Addr, 0, len(nodes.Items))
+	cpIPs := make([]netip.Addr, 0, 1)
 
 	for _, node := range nodes.Items {
 		for _, addr := range node.Status.Addresses {
@@ -651,6 +682,10 @@ func (ct *ConnectivityTest) detectNodeCIDRs(ctx context.Context) error {
 				continue
 			}
 			nodeIPs = append(nodeIPs, ip)
+			if _, ok := node.Labels["node-role.kubernetes.io/control-plane"]; ok {
+				cpIPs = append(cpIPs, ip)
+
+			}
 		}
 	}
 
@@ -659,33 +694,44 @@ func (ct *ConnectivityTest) detectNodeCIDRs(ctx context.Context) error {
 	}
 
 	// collapse set of IPs in to CIDRs
-	nodeCIDRs := []netip.Prefix{}
-
-	for _, ip := range nodeIPs {
-		found := false
-		for _, cidr := range nodeCIDRs {
-			if cidr.Addr().Is4() == ip.Is4() && cidr.Contains(ip) {
-				found = true
-				break
-			}
-		}
-		if found {
-			continue
-		}
-
-		// Generate a /24 or /64 accordingly
-		bits := 24
-		if ip.Is6() {
-			bits = 64
-		}
-		nodeCIDRs = append(nodeCIDRs, netip.PrefixFrom(ip, bits).Masked())
-	}
+	nodeCIDRs := netIPToCIDRs(nodeIPs)
+	cpCIDRs := netIPToCIDRs(cpIPs)
 
 	ct.params.NodeCIDRs = make([]string, 0, len(nodeCIDRs))
 	for _, cidr := range nodeCIDRs {
 		ct.params.NodeCIDRs = append(ct.params.NodeCIDRs, cidr.String())
 	}
+
+	ct.params.ControlPlaneCIDRs = make([]string, 0, len(cpCIDRs))
+	for _, cidr := range nodeCIDRs {
+		ct.params.ControlPlaneCIDRs = append(ct.params.ControlPlaneCIDRs, cidr.String())
+	}
+
 	ct.Debugf("Detected NodeCIDRs: %v", ct.params.NodeCIDRs)
+	ct.Debugf("Detected ControlPlaneCIDRs: %v", ct.params.ControlPlaneCIDRs)
+	return nil
+}
+
+// detectK8sCIDR produces one CIDR that covers the kube-apiserver address.
+// ipv4 addresses are collapsed in to one or more /24s, and v6 to one or more /64s
+func (ct *ConnectivityTest) detectK8sCIDR(ctx context.Context) error {
+	service, err := ct.client.GetService(ctx, "default", "kubernetes", metav1.GetOptions{})
+	if err != nil {
+		return fmt.Errorf("failed to get \"kubernetes.default\" service: %w", err)
+	}
+	addr, err := netip.ParseAddr(service.Spec.ClusterIP)
+	if err != nil {
+		return fmt.Errorf("failed to parse \"kubernetes.default\" service Cluster IP: %w", err)
+	}
+
+	// Generate a /24 or /64 accordingly
+	bits := 24
+	if addr.Is6() {
+		bits = 64
+	}
+	ct.params.K8sCIDR = netip.PrefixFrom(addr, bits).Masked().String()
+	ct.k8sService = Service{Service: service, URLPath: "/version"}
+	ct.Debugf("Detected K8sCIDR: %q", ct.params.K8sCIDR)
 	return nil
 }
 
@@ -948,6 +994,10 @@ func (ct *ConnectivityTest) RandomClientPod() *Pod {
 	return nil
 }
 
+func (ct *ConnectivityTest) ControlPlaneClientPod() *Pod {
+	return ct.clientCPPod
+}
+
 func (ct *ConnectivityTest) Params() Parameters {
 	return ct.params
 }
@@ -1000,6 +1050,10 @@ func (ct *ConnectivityTest) IngressService() map[string]Service {
 	return ct.ingressService
 }
 
+func (ct *ConnectivityTest) K8sService() Service {
+	return ct.k8sService
+}
+
 func (ct *ConnectivityTest) ExternalWorkloads() map[string]ExternalWorkload {
 	return ct.externalWorkloads
 }

connectivity/check/deployment.go

@@ -15,6 +15,7 @@ import (
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	networkingv1 "k8s.io/api/networking/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
 	k8sErrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
@@ -33,6 +34,7 @@ const (
 
 	clientDeploymentName  = "client"
 	client2DeploymentName = "client2"
+	clientCPDeployment    = "client-cp"
 
 	DNSTestServerContainerName = "dns-test-server"
 
@@ -852,6 +854,35 @@ func (ct *ConnectivityTest) deploy(ctx context.Context) error {
 		}
 	}
 
+	// 3rd client scheduled on the control plane
+	if ct.params.K8sLocalHostTest {
+		ct.Logf("✨ [%s] Deploying %s deployment...", ct.clients.src.ClusterName(), clientCPDeployment)
+		clientDeployment := newDeployment(deploymentParameters{
+			Name:        clientCPDeployment,
+			Kind:        kindClientName,
+			NamedPort:   "http-8080",
+			Port:        8080,
+			Image:       ct.params.CurlImage,
+			Command:     []string{"/bin/ash", "-c", "sleep 10000000"},
+			Labels:      map[string]string{"other": "client"},
+			Annotations: ct.params.DeploymentAnnotations.Match(client2DeploymentName),
+			NodeSelector: map[string]string{
+				"node-role.kubernetes.io/control-plane": "",
+			},
+			Tolerations: []corev1.Toleration{
+				{Key: "node-role.kubernetes.io/control-plane"},
+			},
+		})
+		_, err = ct.clients.src.CreateServiceAccount(ctx, ct.params.TestNamespace, k8s.NewServiceAccount(clientCPDeployment), metav1.CreateOptions{})
+		if err != nil && !errors.IsAlreadyExists(err) {
+			return fmt.Errorf("unable to create service account %s: %s", clientCPDeployment, err)
+		}
+		_, err = ct.clients.src.CreateDeployment(ctx, ct.params.TestNamespace, clientDeployment, metav1.CreateOptions{})
+		if err != nil && !errors.IsAlreadyExists(err) {
+			return fmt.Errorf("unable to create deployment %s: %s", clientCPDeployment, err)
+		}
+	}
+
 	if !ct.params.SingleNode || ct.params.MultiCluster != "" {
 		_, err = ct.clients.dst.GetService(ctx, ct.params.TestNamespace, echoOtherNodeDeploymentName, metav1.GetOptions{})
 		if err != nil {
@@ -1158,9 +1189,16 @@ func (ct *ConnectivityTest) validateDeployment(ctx context.Context) error {
 			return err
 		}
 
-		ct.clientPods[pod.Name] = Pod{
-			K8sClient: ct.client,
-			Pod:       pod.DeepCopy(),
+		if strings.Index(pod.Name, clientCPDeployment) == -1 {
+			ct.clientPods[pod.Name] = Pod{
+				K8sClient: ct.client,
+				Pod:       pod.DeepCopy(),
+			}
+		} else {
+			ct.clientCPPod = &Pod{
+				K8sClient: ct.client,
+				Pod:       pod.DeepCopy(),
+			}
 		}
 	}
 
@@ -1222,6 +1260,9 @@ func (ct *ConnectivityTest) validateDeployment(ctx context.Context) error {
 			return err
 		}
 	}
+	if err := WaitForCoreDNS(ctx, ct, *ct.clientCPPod); err != nil {
+		return err
+	}
 
 	for _, client := range ct.clients.clients() {
 		echoPods, err := client.ListPods(ctx, ct.params.TestNamespace, metav1.ListOptions{LabelSelector: "kind=" + kindEchoName})

connectivity/check/peer.go

@@ -4,6 +4,7 @@
 package check
 
 import (
+	"fmt"
 	"net"
 	"net/url"
 	"strconv"
@@ -162,6 +163,8 @@ func (p Pod) FlowFilters() []*flow.FlowFilter {
 type Service struct {
 	// Service  is the Kubernetes service resource
 	Service *corev1.Service
+
+	URLPath string
 }
 
 // Name returns the absolute name of the service.
@@ -176,21 +179,24 @@ func (s Service) NameWithoutNamespace() string {
 
 // Scheme returns the string 'http'.
 func (s Service) Scheme() string {
-	// We only have http services for now.
+	if s.Port() == 443 {
+		return "https"
+	}
 	return "http"
+
 }
 
 // Path returns the string '/'.
 func (s Service) Path() string {
 	// No support for paths yet.
-	return ""
+	return s.URLPath
 }
 
 // Address returns the network address of the Service.
 func (s Service) Address(family features.IPFamily) string {
 	// If the cluster IP is empty (headless service case) or the IP family is set to any, return the service name
 	if s.Service.Spec.ClusterIP == "" || family == features.IPFamilyAny {
-		return s.Service.Name
+		return fmt.Sprintf("%s.%s", s.Service.Name, s.Service.Namespace)
 	}
 
 	getClusterIPForIPFamily := func(family v1.IPFamily) string {

connectivity/manifests/client-egress-to-cidr-cp-host-knp.yaml

@@ -0,0 +1,15 @@
+# This policy allows packets to all node IPs
+apiVersion: networking.k8s.io/v1
+kind: NetworkPolicy
+metadata:
+  name: client-egress-to-cidr-cp-host
+spec:
+  podSelector:
+    matchLabels:
+      kind: client
+  egress:
+    - to:
+{{- range .ControlPlaneCIDRs }}
+        - ipBlock:
+            cidr: {{.}}
+{{- end }}

connectivity/manifests/client-egress-to-cidr-k8s.yaml

@@ -0,0 +1,19 @@
+apiVersion: cilium.io/v2
+kind: CiliumNetworkPolicy
+metadata:
+  name: client-egress-to-cidr-k8s
+spec:
+  endpointSelector:
+    matchLabels:
+      kind: client
+  egress:
+  - toCIDR:
+      - {{ .K8sCIDR }}
+  - toEndpoints:
+    - matchExpressions:
+      - { key: 'k8s-app', operator: In, values: [ "kube-dns", "coredns", "node-local-dns", "nodelocaldns" ] }
+      - { key: 'io.kubernetes.pod.namespace', operator: In, values: [ "kube-system" ] }
+    toPorts:
+    - ports:
+      - port: "53"
+        protocol: ANY

connectivity/manifests/client-egress-to-entities-host.yaml

@@ -0,0 +1,11 @@
+apiVersion: cilium.io/v2
+kind: CiliumNetworkPolicy
+metadata:
+  name: client-egress-to-entities-host
+spec:
+  endpointSelector:
+    matchLabels:
+      kind: client
+  egress:
+  - toEntities:
+    - host

connectivity/manifests/client-egress-to-entities-k8s.yaml

@@ -0,0 +1,19 @@
+apiVersion: cilium.io/v2
+kind: CiliumNetworkPolicy
+metadata:
+  name: client-egress-to-entities-k8s
+spec:
+  endpointSelector:
+    matchLabels:
+      kind: client
+  egress:
+  - toEntities:
+    - kube-apiserver
+  - toEndpoints:
+    - matchExpressions:
+      - { key: 'k8s-app', operator: In, values: [ "kube-dns", "coredns", "node-local-dns", "nodelocaldns" ] }
+      - { key: 'io.kubernetes.pod.namespace', operator: In, values: [ "kube-system" ] }
+    toPorts:
+    - ports:
+      - port: "53"
+        protocol: ANY