weave/npc/controller.go

package npc

import (
	"fmt"
	"sync"

	"github.com/pkg/errors"
	coreapi "k8s.io/api/core/v1"
	extnapi "k8s.io/api/extensions/v1beta1"
	networkingv1 "k8s.io/api/networking/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/kubernetes"

	"github.com/weaveworks/weave/common"
	"github.com/weaveworks/weave/common/chains"
	"github.com/weaveworks/weave/net/ipset"
	"github.com/weaveworks/weave/npc/iptables"
)

type NetworkPolicyController interface {
	AddNamespace(ns *coreapi.Namespace) error
	UpdateNamespace(oldObj, newObj *coreapi.Namespace) error
	DeleteNamespace(ns *coreapi.Namespace) error

	AddPod(obj *coreapi.Pod) error
	UpdatePod(oldObj, newObj *coreapi.Pod) error
	DeletePod(obj *coreapi.Pod) error

	AddNetworkPolicy(obj interface{}) error
	UpdateNetworkPolicy(oldObj, newObj interface{}) error
	DeleteNetworkPolicy(obj interface{}) error
}

type controller struct {
	sync.Mutex

	nodeName string // my node name

	ipt                    iptables.Interface
	ips                    ipset.Interface
	clientset              kubernetes.Interface
	nss                    map[string]*ns // ns name -> ns struct
	nsSelectors            *selectorSet   // selector string -> nsSelector
	namespacedPodSelectors *selectorSet
	defaultEgressDrop      bool // flag to track if base iptable rule to drop egress traffic is added or not
}

func New(nodeName string, ipt iptables.Interface, ips ipset.Interface, clientset kubernetes.Interface) NetworkPolicyController {
	c := &controller{
		nodeName:  nodeName,
		ipt:       ipt,
		ips:       ips,
		clientset: clientset,
		nss:       make(map[string]*ns)}

	doNothing := func(*selector, policyType) error { return nil }
	c.nsSelectors = newSelectorSet(ips, c.onNewNsSelector, doNothing, doNothing)
	c.namespacedPodSelectors = newSelectorSet(ips, c.onNewNamespacePodsSelector, doNothing, doNothing)
	return c
}

func (npc *controller) onNewNsSelector(selector *selector) error {
	for _, ns := range npc.nss {
		if ns.namespaceUID != "" {
			if selector.matchesNamespaceSelector(ns.namespaceLabels) {
				if err := selector.addEntry(ns.namespaceUID, string(ns.allPods.ipsetName), namespaceComment(ns)); err != nil {
					return err
				}
			}
		}
	}
	return nil
}

func (npc *controller) onNewNamespacePodsSelector(selector *selector) error {
	for _, ns := range npc.nss {
		if ns.namespaceUID != "" && len(ns.pods) > 0 {
			for _, pod := range ns.pods {
				if hasIP(pod) {
					if selector.matchesNamespacedPodSelector(pod.ObjectMeta.Labels, ns.namespaceLabels) {
						if err := selector.addEntry(uid(pod), pod.Status.PodIP, podComment(pod)); err != nil {
							return err
						}

					}
				}
			}
		}
	}
	return nil
}

func (npc *controller) withNS(name string, f func(ns *ns) error) error {
	ns, found := npc.nss[name]
	if !found {
		namespace, err := npc.clientset.CoreV1().Namespaces().Get(name, metav1.GetOptions{})
		if err != nil {
			return err
		}
		newNs, err := newNS(name, npc.nodeName, npc.ipt, npc.ips, npc.nsSelectors, npc.namespacedPodSelectors, namespace)
		if err != nil {
			return err
		}
		npc.nss[name] = newNs
		ns = newNs
	}
	if err := f(ns); err != nil {
		return err
	}
	if ns.empty() {
		if err := ns.destroy(); err != nil {
			return err
		}
		delete(npc.nss, name)
	}
	return nil
}

func (npc *controller) AddPod(obj *coreapi.Pod) error {
	npc.Lock()
	defer npc.Unlock()

	common.Log.Debugf("EVENT AddPod %s", js(obj))
	return npc.withNS(obj.ObjectMeta.Namespace, func(ns *ns) error {
		return errors.Wrap(ns.addPod(obj), "add pod")
	})
}

func (npc *controller) UpdatePod(oldObj, newObj *coreapi.Pod) error {
	npc.Lock()
	defer npc.Unlock()

	common.Log.Debugf("EVENT UpdatePod %s %s", js(oldObj), js(newObj))
	return npc.withNS(oldObj.ObjectMeta.Namespace, func(ns *ns) error {
		return errors.Wrap(ns.updatePod(oldObj, newObj), "update pod")
	})
}

func (npc *controller) DeletePod(obj *coreapi.Pod) error {
	npc.Lock()
	defer npc.Unlock()

	common.Log.Debugf("EVENT DeletePod %s", js(obj))
	return npc.withNS(obj.ObjectMeta.Namespace, func(ns *ns) error {
		return errors.Wrap(ns.deletePod(obj), "delete pod")
	})
}

func (npc *controller) AddNetworkPolicy(obj interface{}) error {
	npc.Lock()
	defer npc.Unlock()

	// lazily add default rule to drop egress traffic only when network policies are applied
	if !npc.defaultEgressDrop {
		egressNetworkPolicy, err := isEgressNetworkPolicy(obj)
		if err != nil {
			return err
		}
		if egressNetworkPolicy {
			npc.defaultEgressDrop = true
			if err := npc.ipt.Append(TableFilter, chains.EgressChain,
				"-m", "mark", "!", "--mark", EgressMark, "-j", "DROP"); err != nil {
				npc.defaultEgressDrop = false
				return fmt.Errorf("Failed to add iptable rule to drop egress traffic from the pods by default due to %s", err.Error())
			}
		}
	}

	nsName, err := nsName(obj)
	if err != nil {
		return err
	}
	common.Log.Infof("EVENT AddNetworkPolicy %s", js(obj))
	return npc.withNS(nsName, func(ns *ns) error {
		return errors.Wrap(ns.addNetworkPolicy(obj), "add network policy")
	})
}

func (npc *controller) UpdateNetworkPolicy(oldObj, newObj interface{}) error {
	npc.Lock()
	defer npc.Unlock()

	nsName, err := nsName(oldObj)
	if err != nil {
		return err
	}

	common.Log.Infof("EVENT UpdateNetworkPolicy %s %s", js(oldObj), js(newObj))
	return npc.withNS(nsName, func(ns *ns) error {
		return errors.Wrap(ns.updateNetworkPolicy(oldObj, newObj), "update network policy")
	})
}

func (npc *controller) DeleteNetworkPolicy(obj interface{}) error {
	npc.Lock()
	defer npc.Unlock()

	nsName, err := nsName(obj)
	if err != nil {
		return err
	}

	common.Log.Infof("EVENT DeleteNetworkPolicy %s", js(obj))
	return npc.withNS(nsName, func(ns *ns) error {
		return errors.Wrap(ns.deleteNetworkPolicy(obj), "delete network policy")
	})
}

func (npc *controller) AddNamespace(obj *coreapi.Namespace) error {
	npc.Lock()
	defer npc.Unlock()

	common.Log.Infof("EVENT AddNamespace %s", js(obj))
	return npc.withNS(obj.ObjectMeta.Name, func(ns *ns) error {
		return errors.Wrap(ns.addNamespace(obj), "add namespace")
	})
}

func (npc *controller) UpdateNamespace(oldObj, newObj *coreapi.Namespace) error {
	npc.Lock()
	defer npc.Unlock()

	common.Log.Infof("EVENT UpdateNamespace %s %s", js(oldObj), js(newObj))
	return npc.withNS(oldObj.ObjectMeta.Name, func(ns *ns) error {
		return errors.Wrap(ns.updateNamespace(oldObj, newObj), "update namespace")
	})
}

func (npc *controller) DeleteNamespace(obj *coreapi.Namespace) error {
	npc.Lock()
	defer npc.Unlock()

	common.Log.Infof("EVENT DeleteNamespace %s", js(obj))
	return npc.withNS(obj.ObjectMeta.Name, func(ns *ns) error {
		return errors.Wrap(ns.deleteNamespace(obj), "delete namespace")
	})
}

func nsName(obj interface{}) (string, error) {
	switch obj := obj.(type) {
	case *networkingv1.NetworkPolicy:
		return obj.ObjectMeta.Namespace, nil
	case *extnapi.NetworkPolicy:
		return obj.ObjectMeta.Namespace, nil
	}

	return "", errInvalidNetworkPolicyObjType
}

func isEgressNetworkPolicy(obj interface{}) (bool, error) {
	if policy, ok := obj.(*networkingv1.NetworkPolicy); ok {
		if len(policy.Spec.PolicyTypes) > 0 {
			for _, policyType := range policy.Spec.PolicyTypes {
				if policyType == networkingv1.PolicyTypeEgress {
					return true, nil
				}
			}
		}
		if policy.Spec.Egress != nil {
			return true, nil
		}
		return false, nil
	}
	return false, errInvalidNetworkPolicyObjType
}

func uid(pod *coreapi.Pod) ipset.UID {
	return ipset.UID(pod.UID)
}

func nsuid(ns *coreapi.Namespace) ipset.UID {
	return ipset.UID(ns.UID)
}