Skip to content

K
k3s
Unverified Commit 773c2a41 authored by Manuel Buil's avatar Manuel Buil Committed by GitHub
Browse files
Options

Merge pull request #5079 from manuelbuil/michalsPR

netpol: Use kube-router as a library
parents 4f36c82f 4fed9f40
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 36 additions and 358 deletions
README.md
View file @ 773c2a41
......@@ -33,7 +33,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 load balancer 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
View file @ 773c2a41
......@@ -64,11 +64,11 @@ replace (
require (
github.com/Microsoft/hcsshim v0.9.2
github.com/cloudnativelabs/kube-router v1.3.2
github.com/containerd/cgroups v1.0.1
github.com/containerd/containerd v1.6.0-beta.2.0.20211117185425-a776a27af54a
github.com/containerd/fuse-overlayfs-snapshotter v1.0.4
github.com/containerd/stargz-snapshotter v0.10.1
github.com/coreos/go-iptables v0.5.0
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
......@@ -127,7 +127,6 @@ require (
k8s.io/controller-manager v0.23.3 // indirect
k8s.io/cri-api v0.23.3
k8s.io/klog v1.0.0
k8s.io/klog/v2 v2.30.0
k8s.io/kubectl v0.23.3
k8s.io/kubernetes v1.23.3
k8s.io/utils v0.0.0-20211116205334-6203023598ed
......
pkg/agent/netpol/namespace.go deleted 100644 → 0
View file @ 4f36c82f
// 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
View file @ 773c2a41
// 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 deleted 100644 → 0
View file @ 4f36c82f
// 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 deleted 100644 → 0
View file @ 4f36c82f
// 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_UnitbuildIPSetRestore(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 deleted 100644 → 0
View file @ 4f36c82f
// 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 deleted 100644 → 0
View file @ 4f36c82f
// 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")
}
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment