Adding dns network policies - Part 1

pkg/cliplugins/workload/plugin/syncer.yaml

@@ -41,6 +41,14 @@ rules:
   - "get"
   - "watch"
   - "list"
+- apiGroups:
+  - "networking.k8s.io"
+  resources:
+  - networkpolicies
+  verbs:
+  - "create"
+  - "list"
+  - "watch"
 {{- range $groupMapping := .GroupMappings}}
 - apiGroups:
   - "{{$groupMapping.APIGroup}}"

pkg/syncer/spec/dns/dns_process.go

@@ -28,6 +28,7 @@ import (
 	"k8s.io/client-go/kubernetes"
 	listersappsv1 "k8s.io/client-go/listers/apps/v1"
 	listerscorev1 "k8s.io/client-go/listers/core/v1"
+	listersnetworkingv1 "k8s.io/client-go/listers/networking/v1"
 	listersrbacv1 "k8s.io/client-go/listers/rbac/v1"
 	"k8s.io/klog/v2"
 
@@ -43,9 +44,11 @@ type DNSProcessor struct {
 	deploymentLister     listersappsv1.DeploymentLister
 	serviceLister        listerscorev1.ServiceLister
 	endpointLister       listerscorev1.EndpointsLister
+	networkPolicyLister  listersnetworkingv1.NetworkPolicyLister
 
-	syncTargetName string
 	syncTargetUID  types.UID
+	syncTargetName string
+	syncTargetKey  string
 	dnsNamespace   string // namespace containing all DNS objects
 	dnsImage       string
 }
@@ -58,8 +61,10 @@ func NewDNSProcessor(
 	deploymentLister listersappsv1.DeploymentLister,
 	serviceLister listerscorev1.ServiceLister,
 	endpointLister listerscorev1.EndpointsLister,
-	syncTargetName string,
+	networkPolicyLister listersnetworkingv1.NetworkPolicyLister,
 	syncTargetUID types.UID,
+	syncTargetName string,
+	syncTargetKey string,
 	dnsNamespace string,
 	dnsImage string) *DNSProcessor {
 
@@ -71,8 +76,10 @@ func NewDNSProcessor(
 		deploymentLister:     deploymentLister,
 		serviceLister:        serviceLister,
 		endpointLister:       endpointLister,
-		syncTargetName:       syncTargetName,
+		networkPolicyLister:  networkPolicyLister,
 		syncTargetUID:        syncTargetUID,
+		syncTargetName:       syncTargetName,
+		syncTargetKey:        syncTargetKey,
 		dnsNamespace:         dnsNamespace,
 		dnsImage:             dnsImage,
 	}
@@ -105,6 +112,9 @@ func (d *DNSProcessor) EnsureDNSUpAndReady(ctx context.Context, workspace logica
 	}
 
 	// No Endpoints resource was found: try to create all the DNS-related resources
+	if err := d.processNetworkPolicy(ctx, dnsID); err != nil {
+		return false, err
+	}
 	if err := d.processServiceAccount(ctx, dnsID); err != nil {
 		return false, err
 	}
@@ -120,6 +130,7 @@ func (d *DNSProcessor) EnsureDNSUpAndReady(ctx context.Context, workspace logica
 	if err := d.processService(ctx, dnsID); err != nil {
 		return false, err
 	}
+
 	// Since the Endpoints resource was not found, the DNS is not yet ready,
 	// even though all the required resources have been created
 	// (deployment still needs to start).
@@ -231,6 +242,25 @@ func (d *DNSProcessor) processService(ctx context.Context, name string) error {
 	return nil
 }
 
+func (d *DNSProcessor) processNetworkPolicy(ctx context.Context, name string) error {
+	logger := klog.FromContext(ctx)
+
+	_, err := d.networkPolicyLister.NetworkPolicies(d.dnsNamespace).Get(name)
+	if apierrors.IsNotFound(err) {
+		expected := MakeNetworkPolicy(name, d.dnsNamespace, d.syncTargetKey)
+		_, err = d.downstreamKubeClient.NetworkingV1().NetworkPolicies(d.dnsNamespace).Create(ctx, expected, metav1.CreateOptions{})
+		if err == nil {
+			logger.Info("NetworkPolicy created")
+		}
+	}
+	if err != nil && !apierrors.IsAlreadyExists(err) {
+		logger.Error(err, "failed to get NetworkPolicy (retrying)")
+		return err
+	}
+
+	return nil
+}
+
 func hasAtLeastOneReadyAddress(endpoints *corev1.Endpoints) bool {
 	for _, s := range endpoints.Subsets {
 		if len(s.Addresses) > 0 && s.Addresses[0].IP != "" {

pkg/syncer/spec/dns/networkpolicy_dns.yaml

@@ -0,0 +1,33 @@
+apiVersion: networking.k8s.io/v1
+kind: NetworkPolicy
+metadata:
+  name: Name
+  namespace: Namespace
+spec:
+  podSelector:
+    matchLabels:
+      app: Name
+  policyTypes:
+    - Ingress
+    - Egress
+  ingress:
+    - from:
+        - namespaceSelector:
+            matchLabels:
+              internal.workload.kcp.dev/cluster: Cluster
+      ports:
+        - protocol: TCP
+          port: 5353
+        - protocol: UDP
+          port: 5353
+  egress:
+    # Only give access to coredns in kube-system
+    - to:
+        - namespaceSelector:
+            matchLabels:
+              kubernetes.io/metadata.name: kube-system
+      ports:
+        - protocol: TCP
+          port: 53
+        - protocol: UDP
+          port: 53

pkg/syncer/spec/dns/resources.go

@@ -24,10 +24,13 @@ import (
 
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	networkingv1 "k8s.io/api/networking/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/yaml"
+
+	workloadv1alpha1 "github.com/kcp-dev/kcp/pkg/apis/workload/v1alpha1"
 )
 
 //go:embed *.yaml
@@ -39,6 +42,7 @@ var (
 	roleBindingTemplate    rbacv1.RoleBinding
 	deploymentTemplate     appsv1.Deployment
 	serviceTemplate        corev1.Service
+	networkPolicyTemplate  networkingv1.NetworkPolicy
 )
 
 func init() {
@@ -47,6 +51,7 @@ func init() {
 	loadTemplateOrDie("rolebinding_dns.yaml", &roleBindingTemplate)
 	loadTemplateOrDie("deployment_dns.yaml", &deploymentTemplate)
 	loadTemplateOrDie("service_dns.yaml", &serviceTemplate)
+	loadTemplateOrDie("networkpolicy_dns.yaml", &networkPolicyTemplate)
 }
 
 func MakeServiceAccount(name, namespace string) *corev1.ServiceAccount {
@@ -105,6 +110,17 @@ func MakeService(name, namespace string) *corev1.Service {
 	return service
 }
 
+func MakeNetworkPolicy(name, namespace, cluster string) *networkingv1.NetworkPolicy {
+	np := networkPolicyTemplate.DeepCopy()
+
+	np.Name = name
+	np.Namespace = namespace
+	np.Spec.PodSelector.MatchLabels["app"] = name
+	np.Spec.Ingress[0].From[0].NamespaceSelector.MatchLabels[workloadv1alpha1.InternalDownstreamClusterLabel] = cluster
+
+	return np
+}
+
 // load a YAML resource into a typed kubernetes object.
 func loadTemplateOrDie(filename string, obj interface{}) {
 	raw, err := dnsFiles.ReadFile(filename)

pkg/syncer/spec/spec_controller.go

@@ -43,6 +43,7 @@ import (
 	"k8s.io/client-go/kubernetes"
 	listersappsv1 "k8s.io/client-go/listers/apps/v1"
 	listerscorev1 "k8s.io/client-go/listers/core/v1"
+	listersnetworkingv1 "k8s.io/client-go/listers/networking/v1"
 	listersrbacv1 "k8s.io/client-go/listers/rbac/v1"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/util/workqueue"
@@ -91,6 +92,7 @@ func NewSpecSyncer(syncerLogger logr.Logger, syncTargetWorkspace logicalcluster.
 	deploymentLister listersappsv1.DeploymentLister,
 	serviceLister listerscorev1.ServiceLister,
 	endpointLister listerscorev1.EndpointsLister,
+	networkPolicyLister listersnetworkingv1.NetworkPolicyLister,
 	dnsNamespace string,
 	dnsImage string) (*Controller, error) {
 
@@ -225,7 +227,7 @@ func NewSpecSyncer(syncerLogger logr.Logger, syncTargetWorkspace logicalcluster.
 	}
 
 	c.dnsProcessor = dns.NewDNSProcessor(downstreamKubeClient, serviceAccountLister, roleLister, roleBindingLister, deploymentLister,
-		serviceLister, endpointLister, syncTargetName, syncTargetUID, dnsNamespace, dnsImage)
+		serviceLister, endpointLister, networkPolicyLister, syncTargetUID, syncTargetName, syncTargetKey, dnsNamespace, dnsImage)
 
 	return &c, nil
 }

pkg/syncer/spec/spec_process_test.go

@@ -1066,6 +1066,7 @@ func TestSyncerProcess(t *testing.T) {
 			deploymentLister := toInformerFactory.Apps().V1().Deployments().Lister()
 			serviceLister := toInformerFactory.Core().V1().Services().Lister()
 			endpointLister := toInformerFactory.Core().V1().Endpoints().Lister()
+			networkPolicyLister := toInformerFactory.Networking().V1().NetworkPolicies().Lister()
 
 			upstreamURL, err := url.Parse("https://kcp.dev:6443")
 			require.NoError(t, err)
@@ -1075,7 +1076,8 @@ func TestSyncerProcess(t *testing.T) {
 			}
 			controller, err := NewSpecSyncer(logger, kcpLogicalCluster, tc.syncTargetName, syncTargetKey, upstreamURL, tc.advancedSchedulingEnabled,
 				fromClusterClient, toClient, toKubeClient, fromInformers, toInformers, mockedCleaner, fakeInformers, syncTargetUID,
-				serviceAccountLister, roleLister, roleBindingLister, deploymentLister, serviceLister, endpointLister, "kcp-01c0zzvlqsi7n", "dnsimage")
+				serviceAccountLister, roleLister, roleBindingLister, deploymentLister, serviceLister, endpointLister,
+				networkPolicyLister, "kcp-01c0zzvlqsi7n", "dnsimage")
 			require.NoError(t, err)
 
 			fromInformers.Start(ctx.Done())

pkg/syncer/syncer.go

@@ -195,6 +195,7 @@ func StartSyncer(ctx context.Context, cfg *SyncerConfig, numSyncerThreads int, i
 	deploymentLister := downstreamInformerFactory.Apps().V1().Deployments().Lister()
 	serviceLister := downstreamInformerFactory.Core().V1().Services().Lister()
 	endpointLister := downstreamInformerFactory.Core().V1().Endpoints().Lister()
+	networkPolicyLister := downstreamInformerFactory.Networking().V1().NetworkPolicies().Lister()
 
 	syncerInformers, err := resourcesync.NewController(
 		logger,
@@ -231,7 +232,7 @@ func StartSyncer(ctx context.Context, cfg *SyncerConfig, numSyncerThreads int, i
 
 	specSyncer, err := spec.NewSpecSyncer(logger, cfg.SyncTargetWorkspace, cfg.SyncTargetName, syncTargetKey, upstreamURL, advancedSchedulingEnabled,
 		upstreamDynamicClusterClient, downstreamDynamicClient, downstreamKubeClient, upstreamInformers, downstreamInformers, downstreamNamespaceController, syncerInformers, syncTarget.GetUID(),
-		serviceAccountLister, roleLister, roleBindingLister, deploymentLister, serviceLister, endpointLister, syncerNamespace, cfg.DNSImage)
+		serviceAccountLister, roleLister, roleBindingLister, deploymentLister, serviceLister, endpointLister, networkPolicyLister, syncerNamespace, cfg.DNSImage)
 	if err != nil {
 		return err
 	}