using an explicti selector set for network policies with both namespace and pod lable selectors

npc/analyser.go

@@ -12,12 +12,14 @@ import (
 
 func (ns *ns) analysePolicy(policy *networkingv1.NetworkPolicy) (
 	rules map[string]*ruleSpec,
-	nsSelectors, podSelectors map[string]*selectorSpec,
+	nsSelectors, podSelectors, namespacedPodSelectors map[string]*selectorSpec,
 	ipBlocks map[string]*ipBlockSpec,
 	err error) {
 
 	nsSelectors = make(map[string]*selectorSpec)
 	podSelectors = make(map[string]*selectorSpec)
+	namespacedPodSelectors = make(map[string]*selectorSpec)
+
 	ipBlocks = make(map[string]*ipBlockSpec)
 	rules = make(map[string]*ruleSpec)
 	policyTypes := make([]policyType, 0)
@@ -33,7 +35,7 @@ func (ns *ns) analysePolicy(policy *networkingv1.NetworkPolicy) (
 	// If empty, matches all pods in a namespace
 	targetSelector, err := newSelectorSpec(&policy.Spec.PodSelector, nil, policyTypes, ns.name, ipset.HashIP)
 	if err != nil {
-		return nil, nil, nil, nil, err
+		return nil, nil, nil, nil, nil, err
 	}
 
 	// To prevent targetSelector being overwritten by a subsequent selector with
@@ -52,7 +54,7 @@ func (ns *ns) analysePolicy(policy *networkingv1.NetworkPolicy) (
 
 			if !allPorts {
 				if err = checkForNamedPorts(ingressRule.Ports); err != nil {
-					return nil, nil, nil, nil, fmt.Errorf("named ports in network policies is not supported yet. "+
+					return nil, nil, nil, nil, nil, fmt.Errorf("named ports in network policies is not supported yet. "+
 						"Rejecting network policy: %s from further processing. "+err.Error(), policy.Name)
 				}
 			}
@@ -73,23 +75,23 @@ func (ns *ns) analysePolicy(policy *networkingv1.NetworkPolicy) (
 
 					// NetworkPolicyPeer describes a peer to allow traffic from.
 					if peer.PodSelector != nil && peer.NamespaceSelector != nil {
-						srcSelector, err = newSelectorSpec(peer.PodSelector, peer.NamespaceSelector, nil, ns.name, ipset.HashIP)
+						srcSelector, err = newSelectorSpec(peer.PodSelector, peer.NamespaceSelector, nil, "", ipset.HashIP)
 						if err != nil {
-							return nil, nil, nil, nil, err
+							return nil, nil, nil, nil, nil, err
 						}
-						addIfNotExist(srcSelector, podSelectors)
+						addIfNotExist(srcSelector, namespacedPodSelectors)
 						srcRuleHost = srcSelector
 					} else if peer.PodSelector != nil {
 						srcSelector, err = newSelectorSpec(peer.PodSelector, nil, nil, ns.name, ipset.HashIP)
 						if err != nil {
-							return nil, nil, nil, nil, err
+							return nil, nil, nil, nil, nil, err
 						}
 						addIfNotExist(srcSelector, podSelectors)
 						srcRuleHost = srcSelector
 					} else if peer.NamespaceSelector != nil {
 						srcSelector, err = newSelectorSpec(nil, peer.NamespaceSelector, nil, "", ipset.ListSet)
 						if err != nil {
-							return nil, nil, nil, nil, err
+							return nil, nil, nil, nil, nil, err
 						}
 						nsSelectors[srcSelector.key] = srcSelector
 						srcRuleHost = srcSelector
@@ -123,7 +125,7 @@ func (ns *ns) analysePolicy(policy *networkingv1.NetworkPolicy) (
 
 			if !allPorts {
 				if err = checkForNamedPorts(egressRule.Ports); err != nil {
-					return nil, nil, nil, nil, fmt.Errorf("named ports in network policies is not supported yet. "+
+					return nil, nil, nil, nil, nil, fmt.Errorf("named ports in network policies is not supported yet. "+
 						"Rejecting network policy: %s from further processing. "+err.Error(), policy.Name)
 				}
 			}
@@ -144,24 +146,24 @@ func (ns *ns) analysePolicy(policy *networkingv1.NetworkPolicy) (
 
 					// NetworkPolicyPeer describes a peer to allow traffic to.
 					if peer.PodSelector != nil && peer.NamespaceSelector != nil {
-						dstSelector, err = newSelectorSpec(peer.PodSelector, peer.NamespaceSelector, nil, ns.name, ipset.HashIP)
+						dstSelector, err = newSelectorSpec(peer.PodSelector, peer.NamespaceSelector, nil, "", ipset.HashIP)
 						if err != nil {
-							return nil, nil, nil, nil, err
+							return nil, nil, nil, nil, nil, err
 						}
-						addIfNotExist(dstSelector, podSelectors)
+						addIfNotExist(dstSelector, namespacedPodSelectors)
 						dstRuleHost = dstSelector
 					} else if peer.PodSelector != nil {
 						dstSelector, err = newSelectorSpec(peer.PodSelector, nil, nil, ns.name, ipset.HashIP)
 						if err != nil {
-							return nil, nil, nil, nil, err
+							return nil, nil, nil, nil, nil, err
 						}
 						addIfNotExist(dstSelector, podSelectors)
 						dstRuleHost = dstSelector
 
 					} else if peer.NamespaceSelector != nil {
-						dstSelector, err = newSelectorSpec(peer.NamespaceSelector, nil, nil, "", ipset.ListSet)
+						dstSelector, err = newSelectorSpec(nil, peer.NamespaceSelector, nil, "", ipset.ListSet)
 						if err != nil {
-							return nil, nil, nil, nil, err
+							return nil, nil, nil, nil, nil, err
 						}
 						nsSelectors[dstSelector.key] = dstSelector
 						dstRuleHost = dstSelector
@@ -185,7 +187,7 @@ func (ns *ns) analysePolicy(policy *networkingv1.NetworkPolicy) (
 		}
 	}
 
-	return rules, nsSelectors, podSelectors, ipBlocks, nil
+	return rules, nsSelectors, podSelectors, namespacedPodSelectors, ipBlocks, nil
 }
 
 func addIfNotExist(s *selectorSpec, ss map[string]*selectorSpec) {

npc/controller.go

@@ -36,9 +36,10 @@ type controller struct {
 	ipt iptables.Interface
 	ips ipset.Interface
 
-	nss               map[string]*ns // ns name -> ns struct
-	nsSelectors       *selectorSet   // selector string -> nsSelector
-	defaultEgressDrop bool           // flag to track if base iptable rule to drop egress traffic is added or not
+	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) NetworkPolicyController {
@@ -50,7 +51,7 @@ func New(nodeName string, ipt iptables.Interface, ips ipset.Interface) NetworkPo
 
 	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
 }
 
@@ -67,10 +68,28 @@ func (npc *controller) onNewNsSelector(selector *selector) error {
 	return nil
 }
 
+func (npc *controller) onNewNamespacePodsSelector(selector *selector) error {
+	for _, ns := range npc.nss {
+		if ns.namespace != nil && len(ns.pods) > 0 {
+			for _, pod := range ns.pods {
+				if hasIP(pod) {
+					if selector.matches(&pod.ObjectMeta.Labels, &ns.namespace.ObjectMeta.Labels) {
+						if err := selector.addEntry(pod.ObjectMeta.UID, 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 {
-		newNs, err := newNS(name, npc.nodeName, npc.ipt, npc.ips, npc.nsSelectors)
+		newNs, err := newNS(name, npc.nodeName, npc.ipt, npc.ips, npc.nsSelectors, npc.namespacedPodSelectors)
 		if err != nil {
 			return err
 		}

npc/namespace.go

@@ -10,6 +10,8 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/uuid"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
 
 	"github.com/weaveworks/weave/common"
 	"github.com/weaveworks/weave/net/ipset"
@@ -37,30 +39,48 @@ type ns struct {
 	ingressDefaultAllowIPSet ipset.Name
 	egressDefaultAllowIPSet  ipset.Name
 
-	nsSelectors  *selectorSet
-	podSelectors *selectorSet
-	ipBlocks     *ipBlockSet
-	rules        *ruleSet
+	nsSelectors             *selectorSet // reference to global selectorSet that is shared across the `ns`. Used to represent all pods in the matching namespaces
+	podSelectors            *selectorSet // used to represent the matching pods in namespace respresented by this `ns`
+	namespacedPodsSelectors *selectorSet // reference to global selectorSet that is shared across the `ns`. Used to represent matching pods in matching namespace
+	ipBlocks                *ipBlockSet
+	rules                   *ruleSet
 }
 
-func newNS(name, nodeName string, ipt iptables.Interface, ips ipset.Interface, nsSelectors *selectorSet) (*ns, error) {
+func newNS(name, nodeName string, ipt iptables.Interface, ips ipset.Interface, nsSelectors *selectorSet, namespacedPodsSelectors *selectorSet) (*ns, error) {
 	allPods, err := newSelectorSpec(&metav1.LabelSelector{}, nil, nil, name, ipset.HashIP)
 	if err != nil {
 		return nil, err
 	}
 
+	config, err := rest.InClusterConfig()
+	if err != nil {
+		return nil, err
+	}
+
+	client, err := kubernetes.NewForConfig(config)
+	if err != nil {
+		return nil, err
+	}
+
+	namespace, err := client.CoreV1().Namespaces().Get(name, metav1.GetOptions{})
+	if err != nil {
+		return nil, err
+	}
+
 	ns := &ns{
-		ipt:         ipt,
-		ips:         ips,
-		name:        name,
-		nodeName:    nodeName,
-		pods:        make(map[types.UID]*coreapi.Pod),
-		policies:    make(map[types.UID]interface{}),
-		uid:         uuid.NewUUID(),
-		allPods:     allPods,
-		nsSelectors: nsSelectors,
-		ipBlocks:    newIPBlockSet(ips),
-		rules:       newRuleSet(ipt),
+		ipt:                     ipt,
+		ips:                     ips,
+		name:                    name,
+		namespace:               namespace,
+		nodeName:                nodeName,
+		pods:                    make(map[types.UID]*coreapi.Pod),
+		policies:                make(map[types.UID]interface{}),
+		uid:                     uuid.NewUUID(),
+		allPods:                 allPods,
+		nsSelectors:             nsSelectors,
+		namespacedPodsSelectors: namespacedPodsSelectors,
+		ipBlocks:                newIPBlockSet(ips),
+		rules:                   newRuleSet(ipt),
 	}
 
 	ns.podSelectors = newSelectorSet(ips, ns.onNewPodSelector, ns.onNewTargetPodSelector, ns.onDestroyTargetPodSelector)
@@ -185,6 +205,11 @@ func (ns *ns) addPod(obj *coreapi.Pod) error {
 		}
 	}
 
+	_, _, err = ns.namespacedPodsSelectors.addToMatching(obj.ObjectMeta.UID, &obj.ObjectMeta.Labels, &ns.namespace.ObjectMeta.Labels, obj.Status.PodIP, podComment(obj))
+	if err != nil {
+		return err
+	}
+
 	return nil
 }
 
@@ -203,6 +228,9 @@ func (ns *ns) updatePod(oldObj, newObj *coreapi.Pod) error {
 		if err := ns.ips.DelEntry(oldObj.ObjectMeta.UID, ns.egressDefaultAllowIPSet, oldObj.Status.PodIP); err != nil {
 			return err
 		}
+		if err := ns.namespacedPodsSelectors.delFromMatching(oldObj.ObjectMeta.UID, &oldObj.ObjectMeta.Labels, &ns.namespace.ObjectMeta.Labels, oldObj.Status.PodIP); err != nil {
+			return err
+		}
 
 		return ns.podSelectors.delFromMatching(oldObj.ObjectMeta.UID, &oldObj.ObjectMeta.Labels, nil, oldObj.Status.PodIP)
 	}
@@ -223,6 +251,10 @@ func (ns *ns) updatePod(oldObj, newObj *coreapi.Pod) error {
 				return err
 			}
 		}
+		_, _, err = ns.namespacedPodsSelectors.addToMatching(newObj.ObjectMeta.UID, &newObj.ObjectMeta.Labels, &ns.namespace.ObjectMeta.Labels, newObj.Status.PodIP, podComment(newObj))
+		if err != nil {
+			return err
+		}
 
 		return nil
 	}
@@ -263,6 +295,23 @@ func (ns *ns) updatePod(oldObj, newObj *coreapi.Pod) error {
 				return err
 			}
 		}
+		for _, ps := range ns.namespacedPodsSelectors.entries {
+			oldMatch := ps.matches(&oldObj.ObjectMeta.Labels, &ns.namespace.ObjectMeta.Labels)
+			newMatch := ps.matches(&newObj.ObjectMeta.Labels, &ns.namespace.ObjectMeta.Labels)
+			if oldMatch == newMatch && oldObj.Status.PodIP == newObj.Status.PodIP {
+				continue
+			}
+			if oldMatch {
+				if err := ps.delEntry(oldObj.ObjectMeta.UID, oldObj.Status.PodIP); err != nil {
+					return err
+				}
+			}
+			if newMatch {
+				if err := ps.addEntry(newObj.ObjectMeta.UID, newObj.Status.PodIP, podComment(newObj)); err != nil {
+					return err
+				}
+			}
+		}
 	}
 
 	return nil
@@ -298,14 +347,19 @@ func (ns *ns) deletePod(obj *coreapi.Pod) error {
 	if err := ns.ips.DelEntry(obj.ObjectMeta.UID, ns.egressDefaultAllowIPSet, obj.Status.PodIP); err != nil {
 		return err
 	}
-
-	return ns.podSelectors.delFromMatching(obj.ObjectMeta.UID, &obj.ObjectMeta.Labels, nil, obj.Status.PodIP)
+	if err := ns.podSelectors.delFromMatching(obj.ObjectMeta.UID, &obj.ObjectMeta.Labels, nil, obj.Status.PodIP); err != nil {
+		return err
+	}
+	if err := ns.namespacedPodsSelectors.delFromMatching(obj.ObjectMeta.UID, &obj.ObjectMeta.Labels, &ns.namespace.ObjectMeta.Labels, obj.Status.PodIP); err != nil {
+		return err
+	}
+	return nil
 }
 
 func (ns *ns) addNetworkPolicy(obj interface{}) error {
 	// Analyse policy, determine which rules and ipsets are required
 
-	uid, rules, nsSelectors, podSelectors, ipBlocks, err := ns.analyse(obj)
+	uid, rules, nsSelectors, podSelectors, namespacedPodsSelectors, ipBlocks, err := ns.analyse(obj)
 	if err != nil {
 		return err
 	}
@@ -317,6 +371,9 @@ func (ns *ns) addNetworkPolicy(obj interface{}) error {
 	if err := ns.podSelectors.provision(uid, nil, podSelectors); err != nil {
 		return err
 	}
+	if err := ns.namespacedPodsSelectors.provision(uid, nil, namespacedPodsSelectors); err != nil {
+		return err
+	}
 	if err := ns.ipBlocks.provision(uid, nil, ipBlocks); err != nil {
 		return err
 	}
@@ -325,11 +382,11 @@ func (ns *ns) addNetworkPolicy(obj interface{}) error {
 
 func (ns *ns) updateNetworkPolicy(oldObj, newObj interface{}) error {
 	// Analyse the old and the new policy so we can determine differences
-	oldUID, oldRules, oldNsSelectors, oldPodSelectors, oldIPBlocks, err := ns.analyse(oldObj)
+	oldUID, oldRules, oldNsSelectors, oldPodSelectors, oldNamespacedPodsSelectors, oldIPBlocks, err := ns.analyse(oldObj)
 	if err != nil {
 		return err
 	}
-	newUID, newRules, newNsSelectors, newPodSelectors, newIPBlocks, err := ns.analyse(newObj)
+	newUID, newRules, newNsSelectors, newPodSelectors, newNamespacedPodsSelectors, newIPBlocks, err := ns.analyse(newObj)
 	if err != nil {
 		return err
 	}
@@ -347,6 +404,9 @@ func (ns *ns) updateNetworkPolicy(oldObj, newObj interface{}) error {
 	if err := ns.podSelectors.deprovision(oldUID, oldPodSelectors, newPodSelectors); err != nil {
 		return err
 	}
+	if err := ns.namespacedPodsSelectors.deprovision(oldUID, oldNamespacedPodsSelectors, newNamespacedPodsSelectors); err != nil {
+		return err
+	}
 	if err := ns.ipBlocks.deprovision(oldUID, oldIPBlocks, newIPBlocks); err != nil {
 		return err
 	}
@@ -356,6 +416,9 @@ func (ns *ns) updateNetworkPolicy(oldObj, newObj interface{}) error {
 	if err := ns.podSelectors.provision(oldUID, oldPodSelectors, newPodSelectors); err != nil {
 		return err
 	}
+	if err := ns.namespacedPodsSelectors.provision(oldUID, oldNamespacedPodsSelectors, newNamespacedPodsSelectors); err != nil {
+		return err
+	}
 	if err := ns.ipBlocks.provision(oldUID, oldIPBlocks, newIPBlocks); err != nil {
 		return err
 	}
@@ -364,7 +427,7 @@ func (ns *ns) updateNetworkPolicy(oldObj, newObj interface{}) error {
 
 func (ns *ns) deleteNetworkPolicy(obj interface{}) error {
 	// Analyse network policy to free resources
-	uid, rules, nsSelectors, podSelectors, ipBlocks, err := ns.analyse(obj)
+	uid, rules, nsSelectors, podSelectors, namespacedPodsSelectors, ipBlocks, err := ns.analyse(obj)
 	if err != nil {
 		return err
 	}
@@ -381,6 +444,9 @@ func (ns *ns) deleteNetworkPolicy(obj interface{}) error {
 	if err := ns.podSelectors.deprovision(uid, podSelectors, nil); err != nil {
 		return err
 	}
+	if err := ns.namespacedPodsSelectors.deprovision(uid, namespacedPodsSelectors, nil); err != nil {
+		return err
+	}
 	if err := ns.ipBlocks.deprovision(uid, ipBlocks, nil); err != nil {
 		return err
 	}
@@ -546,7 +612,7 @@ func (ns *ns) defaultAllowIPSetName(pt policyType) ipset.Name {
 func (ns *ns) analyse(obj interface{}) (
 	uid types.UID,
 	rules map[string]*ruleSpec,
-	nsSelectors, podSelectors map[string]*selectorSpec,
+	nsSelectors, podSelectors, namespacedPodsSelectors map[string]*selectorSpec,
 	ipBlocks map[string]*ipBlockSpec,
 	err error) {
 
@@ -562,7 +628,7 @@ func (ns *ns) analyse(obj interface{}) (
 	ns.policies[uid] = obj
 
 	// Analyse policy, determine which rules and ipsets are required
-	rules, nsSelectors, podSelectors, ipBlocks, err = ns.analysePolicy(obj.(*networkingv1.NetworkPolicy))
+	rules, nsSelectors, podSelectors, namespacedPodsSelectors, ipBlocks, err = ns.analysePolicy(obj.(*networkingv1.NetworkPolicy))
 	if err != nil {
 		return
 	}