netpol: Use kube-router as a library

README.md

@@ -34,7 +34,7 @@ K3s bundles the following technologies together into a single cohesive distribut
 * [Metrics Server](https://github.com/kubernetes-sigs/metrics-server)
 * [Traefik](https://containo.us/traefik/) for ingress
 * [Klipper-lb](https://github.com/k3s-io/klipper-lb) as an embedded service loadbalancer provider
-* [Kube-router](https://www.kube-router.io/) for network policy
+* [Kube-router](https://www.kube-router.io/) netpol controller for network policy
 * [Helm-controller](https://github.com/k3s-io/helm-controller) to allow for CRD-driven deployment of helm manifests
 * [Kine](https://github.com/k3s-io/kine) as a datastore shim that allows etcd to be replaced with other databases
 * [Local-path-provisioner](https://github.com/rancher/local-path-provisioner) for provisioning volumes using local storage

go.mod

@@ -69,13 +69,13 @@ replace (
 )
 
 require (
+	github.com/cloudnativelabs/kube-router v1.3.2
 	github.com/containerd/cgroups v1.0.1
 	github.com/containerd/containerd v1.5.7
 	github.com/containerd/cri v1.11.1-0.20200820101445-b0cc07999aa5
 	github.com/containerd/fuse-overlayfs-snapshotter v1.0.2
 	github.com/containerd/go-cni v1.0.2 // indirect
 	github.com/containerd/imgcrypt v1.1.1 // indirect
-	github.com/coreos/go-iptables v0.4.5
 	github.com/coreos/go-systemd v0.0.0-20190719114852-fd7a80b32e1f
 	github.com/docker/docker v20.10.10+incompatible
 	github.com/erikdubbelboer/gspt v0.0.0-20190125194910-e68493906b83
@@ -129,7 +129,6 @@ require (
 	k8s.io/controller-manager v0.21.9 // indirect
 	k8s.io/cri-api v0.21.9
 	k8s.io/klog v1.0.0
-	k8s.io/klog/v2 v2.9.0
 	k8s.io/kubectl v0.21.9
 	k8s.io/kubernetes v1.21.9
 	k8s.io/utils v0.0.0-20210521133846-da695404a2bc

pkg/agent/netpol/namespace.go

-// Apache License v2.0 (copyright Cloud Native Labs & Rancher Labs)
-// - modified from https://github.com/cloudnativelabs/kube-router/blob/73b1b03b32c5755b240f6c077bb097abe3888314/pkg/controllers/netpol/namespace.go
-
-// +build !windows
-
-package netpol
-
-import (
-	"reflect"
-
-	api "k8s.io/api/core/v1"
-	"k8s.io/client-go/tools/cache"
-	"k8s.io/klog/v2"
-)
-
-func (npc *NetworkPolicyController) newNamespaceEventHandler() cache.ResourceEventHandler {
-	return cache.ResourceEventHandlerFuncs{
-		AddFunc: func(obj interface{}) {
-			npc.handleNamespaceAdd(obj.(*api.Namespace))
-		},
-		UpdateFunc: func(oldObj, newObj interface{}) {
-			npc.handleNamespaceUpdate(oldObj.(*api.Namespace), newObj.(*api.Namespace))
-		},
-		DeleteFunc: func(obj interface{}) {
-			switch obj := obj.(type) {
-			case *api.Namespace:
-				npc.handleNamespaceDelete(obj)
-				return
-			case cache.DeletedFinalStateUnknown:
-				if namespace, ok := obj.Obj.(*api.Namespace); ok {
-					npc.handleNamespaceDelete(namespace)
-					return
-				}
-			default:
-				klog.Errorf("unexpected object type: %v", obj)
-			}
-		},
-	}
-}
-
-func (npc *NetworkPolicyController) handleNamespaceAdd(obj *api.Namespace) {
-	if obj.Labels == nil {
-		return
-	}
-	klog.V(2).Infof("Received update for namespace: %s", obj.Name)
-
-	npc.RequestFullSync()
-}
-
-func (npc *NetworkPolicyController) handleNamespaceUpdate(oldObj, newObj *api.Namespace) {
-	if reflect.DeepEqual(oldObj.Labels, newObj.Labels) {
-		return
-	}
-	klog.V(2).Infof("Received update for namespace: %s", newObj.Name)
-
-	npc.RequestFullSync()
-}
-
-func (npc *NetworkPolicyController) handleNamespaceDelete(obj *api.Namespace) {
-	if obj.Labels == nil {
-		return
-	}
-	klog.V(2).Infof("Received namespace: %s delete event", obj.Name)
-
-	npc.RequestFullSync()
-}

pkg/agent/netpol/netpol.go

 // Apache License v2.0 (copyright Cloud Native Labs & Rancher Labs)
 // - modified from https://github.com/cloudnativelabs/kube-router/blob/73b1b03b32c5755b240f6c077bb097abe3888314/pkg/controllers/netpol.go
 
+//go:build !windows
 // +build !windows
 
 package netpol
 
 import (
 	"context"
+	"strings"
 	"sync"
 
-	"github.com/rancher/k3s/pkg/agent/netpol/utils"
+	"github.com/cloudnativelabs/kube-router/pkg/controllers/netpol"
+	"github.com/cloudnativelabs/kube-router/pkg/healthcheck"
+	"github.com/cloudnativelabs/kube-router/pkg/options"
+	"github.com/cloudnativelabs/kube-router/pkg/utils"
 	"github.com/rancher/k3s/pkg/daemons/config"
 	"github.com/sirupsen/logrus"
 	"k8s.io/client-go/informers"
@@ -20,8 +25,8 @@ import (
 // Run creates and starts a new instance of the kube-router network policy controller
 // The code in this function is cribbed from the upstream controller at:
 // https://github.com/cloudnativelabs/kube-router/blob/ee9f6d890d10609284098229fa1e283ab5d83b93/pkg/cmd/kube-router.go#L78
-// The NewNetworkPolicyController function has also been modified to use the k3s config.Node struct instead of KubeRouter's
-// CLI configuration, eliminate use of a WaitGroup for shutdown sequencing, and drop Prometheus metrics support.
+// It converts the k3s config.Node into kube-router configuration (only the
+// subset of options needed for netpol controller).
 func Run(ctx context.Context, nodeConfig *config.Node) error {
 	set, err := utils.NewIPSet(false)
 	if err != nil {
@@ -44,7 +49,21 @@ func Run(ctx context.Context, nodeConfig *config.Node) error {
 		return err
 	}
 
+	krConfig := options.NewKubeRouterConfig()
+	krConfig.ClusterIPCIDR = nodeConfig.AgentConfig.ServiceCIDR.String()
+	krConfig.NodePortRange = strings.ReplaceAll(nodeConfig.AgentConfig.ServiceNodePortRange.String(), "-", ":")
+	krConfig.HostnameOverride = nodeConfig.AgentConfig.NodeName
+	krConfig.MetricsEnabled = false
+	krConfig.RunFirewall = true
+	krConfig.RunRouter = false
+	krConfig.RunServiceProxy = false
+
 	stopCh := ctx.Done()
+	healthCh := make(chan *healthcheck.ControllerHeartbeat)
+
+	// We don't use this WaitGroup, but kube-router components require it.
+	var wg sync.WaitGroup
+
 	informerFactory := informers.NewSharedInformerFactory(client, 0)
 	podInformer := informerFactory.Core().V1().Pods().Informer()
 	nsInformer := informerFactory.Core().V1().Namespaces().Informer()
@@ -52,7 +71,15 @@ func Run(ctx context.Context, nodeConfig *config.Node) error {
 	informerFactory.Start(stopCh)
 	informerFactory.WaitForCacheSync(stopCh)
 
-	npc, err := NewNetworkPolicyController(client, nodeConfig, podInformer, npInformer, nsInformer, &sync.Mutex{})
+	// Start kube-router healthcheck server. Netpol requires it
+	hc, err := healthcheck.NewHealthController(krConfig)
+	if err != nil {
+		return err
+	}
+	wg.Add(1)
+	go hc.RunCheck(healthCh, stopCh, &wg)
+
+	npc, err := netpol.NewNetworkPolicyController(client, krConfig, podInformer, npInformer, nsInformer, &sync.Mutex{})
 	if err != nil {
 		return err
 	}
@@ -61,7 +88,9 @@ func Run(ctx context.Context, nodeConfig *config.Node) error {
 	nsInformer.AddEventHandler(npc.NamespaceEventHandler)
 	npInformer.AddEventHandler(npc.NetworkPolicyEventHandler)
 
-	go npc.Run(stopCh)
+	wg.Add(1)
+	logrus.Info("Starting the netpol controller")
+	go npc.Run(healthCh, stopCh, &wg)
 
 	return nil
 }

pkg/agent/netpol/utils.go

-// Apache License v2.0 (copyright Cloud Native Labs & Rancher Labs)
-// - modified from https://github.com/cloudnativelabs/kube-router/blob/73b1b03b32c5755b240f6c077bb097abe3888314/pkg/controllers/netpol/utils.go
-
-package netpol
-
-import (
-	"fmt"
-	"reflect"
-	"regexp"
-	"strconv"
-
-	api "k8s.io/api/core/v1"
-)
-
-const (
-	PodCompleted api.PodPhase = "Completed"
-)
-
-// isPodUpdateNetPolRelevant checks the attributes that we care about for building NetworkPolicies on the host and if it
-// finds a relevant change, it returns true otherwise it returns false. The things we care about for NetworkPolicies:
-// 1) Is the phase of the pod changing? (matters for catching completed, succeeded, or failed jobs)
-// 2) Is the pod IP changing? (changes how the network policy is applied to the host)
-// 3) Is the pod's host IP changing? (should be caught in the above, with the CNI kube-router runs with but we check this as well for sanity)
-// 4) Is a pod's label changing? (potentially changes which NetworkPolicies select this pod)
-func isPodUpdateNetPolRelevant(oldPod, newPod *api.Pod) bool {
-	return newPod.Status.Phase != oldPod.Status.Phase ||
-		newPod.Status.PodIP != oldPod.Status.PodIP ||
-		!reflect.DeepEqual(newPod.Status.PodIPs, oldPod.Status.PodIPs) ||
-		newPod.Status.HostIP != oldPod.Status.HostIP ||
-		!reflect.DeepEqual(newPod.Labels, oldPod.Labels)
-}
-
-func isNetPolActionable(pod *api.Pod) bool {
-	return !isFinished(pod) && pod.Status.PodIP != "" && !pod.Spec.HostNetwork
-}
-
-func isFinished(pod *api.Pod) bool {
-	switch pod.Status.Phase {
-	case api.PodFailed, api.PodSucceeded, PodCompleted:
-		return true
-	}
-	return false
-}
-
-func validateNodePortRange(nodePortOption string) (string, error) {
-	nodePortValidator := regexp.MustCompile(`^([0-9]+)[:-]([0-9]+)$`)
-	if matched := nodePortValidator.MatchString(nodePortOption); !matched {
-		return "", fmt.Errorf("failed to parse node port range given: '%s' please see specification in help text", nodePortOption)
-	}
-	matches := nodePortValidator.FindStringSubmatch(nodePortOption)
-	if len(matches) != 3 {
-		return "", fmt.Errorf("could not parse port number from range given: '%s'", nodePortOption)
-	}
-	port1, err := strconv.ParseUint(matches[1], 10, 16)
-	if err != nil {
-		return "", fmt.Errorf("could not parse first port number from range given: '%s'", nodePortOption)
-	}
-	port2, err := strconv.ParseUint(matches[2], 10, 16)
-	if err != nil {
-		return "", fmt.Errorf("could not parse second port number from range given: '%s'", nodePortOption)
-	}
-	if port1 >= port2 {
-		return "", fmt.Errorf("port 1 is greater than or equal to port 2 in range given: '%s'", nodePortOption)
-	}
-	return fmt.Sprintf("%d:%d", port1, port2), nil
-}

pkg/agent/netpol/utils/ipset_test.go

-// Apache License v2.0 (copyright Cloud Native Labs & Rancher Labs)
-// - modified from https://github.com/cloudnativelabs/kube-router/blob/73b1b03b32c5755b240f6c077bb097abe3888314/pkg/utils/ipset_test.go
-
-package utils
-
-import "testing"
-
-func Test_buildIPSetRestore(t *testing.T) {
-	type args struct {
-		ipset *IPSet
-	}
-	tests := []struct {
-		name string
-		args args
-		want string
-	}{
-		{
-			name: "simple-restore",
-			args: args{
-				ipset: &IPSet{Sets: map[string]*Set{
-					"foo": {
-						Name:    "foo",
-						Options: []string{"hash:ip", "yolo", "things", "12345"},
-						Entries: []*Entry{
-							{Options: []string{"1.2.3.4"}},
-						},
-					},
-					"google-dns-servers": {
-						Name:    "google-dns-servers",
-						Options: []string{"hash:ip", "lol"},
-						Entries: []*Entry{
-							{Options: []string{"4.4.4.4"}},
-							{Options: []string{"8.8.8.8"}},
-						},
-					},
-					// this one and the one above share the same exact options -- and therefore will reuse the same
-					// tmp ipset:
-					"more-ip-addresses": {
-						Name:    "google-dns-servers",
-						Options: []string{"hash:ip", "lol"},
-						Entries: []*Entry{
-							{Options: []string{"5.5.5.5"}},
-							{Options: []string{"6.6.6.6"}},
-						},
-					},
-				}},
-			},
-			want: "create TMP-7NOTZDOMLXBX6DAJ hash:ip yolo things 12345\n" +
-				"flush TMP-7NOTZDOMLXBX6DAJ\n" +
-				"add TMP-7NOTZDOMLXBX6DAJ 1.2.3.4\n" +
-				"create foo hash:ip yolo things 12345\n" +
-				"swap TMP-7NOTZDOMLXBX6DAJ foo\n" +
-				"flush TMP-7NOTZDOMLXBX6DAJ\n" +
-				"create TMP-XD7BSSQZELS7TP35 hash:ip lol\n" +
-				"flush TMP-XD7BSSQZELS7TP35\n" +
-				"add TMP-XD7BSSQZELS7TP35 4.4.4.4\n" +
-				"add TMP-XD7BSSQZELS7TP35 8.8.8.8\n" +
-				"create google-dns-servers hash:ip lol\n" +
-				"swap TMP-XD7BSSQZELS7TP35 google-dns-servers\n" +
-				"flush TMP-XD7BSSQZELS7TP35\n" +
-				"add TMP-XD7BSSQZELS7TP35 5.5.5.5\n" +
-				"add TMP-XD7BSSQZELS7TP35 6.6.6.6\n" +
-				"create google-dns-servers hash:ip lol\n" +
-				"swap TMP-XD7BSSQZELS7TP35 google-dns-servers\n" +
-				"flush TMP-XD7BSSQZELS7TP35\n" +
-				"destroy TMP-7NOTZDOMLXBX6DAJ\n" +
-				"destroy TMP-XD7BSSQZELS7TP35\n",
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			if got := buildIPSetRestore(tt.args.ipset); got != tt.want {
-				t.Errorf("buildIPSetRestore() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}

pkg/agent/netpol/utils/iptables.go

-// Apache License v2.0 (copyright Cloud Native Labs & Rancher Labs)
-// - modified from https://github.com/cloudnativelabs/kube-router/blob/73b1b03b32c5755b240f6c077bb097abe3888314/pkg/utils/iptables.go
-
-package utils
-
-import (
-	"bytes"
-	"fmt"
-	"os/exec"
-	"strings"
-)
-
-var hasWait bool
-
-func init() {
-	path, err := exec.LookPath("iptables-restore")
-	if err != nil {
-		return
-	}
-	args := []string{"iptables-restore", "--help"}
-	cmd := exec.Cmd{
-		Path: path,
-		Args: args,
-	}
-	cmdOutput, err := cmd.CombinedOutput()
-	if err != nil {
-		return
-	}
-	hasWait = strings.Contains(string(cmdOutput), "wait")
-}
-
-// SaveInto calls `iptables-save` for given table and stores result in a given buffer.
-func SaveInto(table string, buffer *bytes.Buffer) error {
-	path, err := exec.LookPath("iptables-save")
-	if err != nil {
-		return err
-	}
-	stderrBuffer := bytes.NewBuffer(nil)
-	args := []string{"iptables-save", "-t", table}
-	cmd := exec.Cmd{
-		Path:   path,
-		Args:   args,
-		Stdout: buffer,
-		Stderr: stderrBuffer,
-	}
-	if err := cmd.Run(); err != nil {
-		return fmt.Errorf("%v (%s)", err, stderrBuffer)
-	}
-	return nil
-}
-
-// Restore runs `iptables-restore` passing data through []byte.
-func Restore(table string, data []byte) error {
-	path, err := exec.LookPath("iptables-restore")
-	if err != nil {
-		return err
-	}
-	var args []string
-	if hasWait {
-		args = []string{"iptables-restore", "--wait", "-T", table}
-	} else {
-		args = []string{"iptables-restore", "-T", table}
-	}
-	cmd := exec.Cmd{
-		Path:  path,
-		Args:  args,
-		Stdin: bytes.NewBuffer(data),
-	}
-	b, err := cmd.CombinedOutput()
-	if err != nil {
-		return fmt.Errorf("%v (%s)", err, b)
-	}
-
-	return nil
-}

pkg/agent/netpol/utils/node.go

-// Apache License v2.0 (copyright Cloud Native Labs & Rancher Labs)
-// - modified from https://github.com/cloudnativelabs/kube-router/blob/73b1b03b32c5755b240f6c077bb097abe3888314/pkg/utils/node.go
-
-// +build !windows
-
-package utils
-
-import (
-	"context"
-	"errors"
-	"fmt"
-	"net"
-	"os"
-
-	apiv1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/client-go/kubernetes"
-)
-
-// GetNodeObject returns the node API object for the node
-func GetNodeObject(clientset kubernetes.Interface, hostnameOverride string) (*apiv1.Node, error) {
-	// assuming kube-router is running as pod, first check env NODE_NAME
-	nodeName := os.Getenv("NODE_NAME")
-	if nodeName != "" {
-		node, err := clientset.CoreV1().Nodes().Get(context.Background(), nodeName, metav1.GetOptions{})
-		if err == nil {
-			return node, nil
-		}
-	}
-
-	// if env NODE_NAME is not set then check if node is register with hostname
-	hostName, _ := os.Hostname()
-	node, err := clientset.CoreV1().Nodes().Get(context.Background(), hostName, metav1.GetOptions{})
-	if err == nil {
-		return node, nil
-	}
-
-	// if env NODE_NAME is not set and node is not registered with hostname, then use host name override
-	if hostnameOverride != "" {
-		node, err = clientset.CoreV1().Nodes().Get(context.Background(), hostnameOverride, metav1.GetOptions{})
-		if err == nil {
-			return node, nil
-		}
-	}
-
-	return nil, fmt.Errorf("failed to identify the node by NODE_NAME, hostname or --hostname-override")
-}
-
-// GetNodeIP returns the most valid external facing IP address for a node.
-// Order of preference:
-// 1. NodeInternalIP
-// 2. NodeExternalIP (Only set on cloud providers usually)
-func GetNodeIP(node *apiv1.Node) (net.IP, error) {
-	addresses := node.Status.Addresses
-	addressMap := make(map[apiv1.NodeAddressType][]apiv1.NodeAddress)
-	for i := range addresses {
-		addressMap[addresses[i].Type] = append(addressMap[addresses[i].Type], addresses[i])
-	}
-	if addresses, ok := addressMap[apiv1.NodeInternalIP]; ok {
-		return net.ParseIP(addresses[0].Address), nil
-	}
-	if addresses, ok := addressMap[apiv1.NodeExternalIP]; ok {
-		return net.ParseIP(addresses[0].Address), nil
-	}
-	return nil, errors.New("host IP unknown")
-}