Stop ignoring local pods

This behaviour was breaking the use of pod selectors for the 'from'
part of a rule.

Move checking of local pods into namespace.go so we can maintain
LocalIpSet correctly

npc/controller.go

@@ -66,7 +66,7 @@ func (npc *controller) onNewNsSelector(selector *selector) error {
 func (npc *controller) withNS(name string, f func(ns *ns) error) error {
 	ns, found := npc.nss[name]
 	if !found {
-		newNs, err := newNS(name, npc.ipt, npc.ips, npc.nsSelectors)
+		newNs, err := newNS(name, npc.nodeName, npc.ipt, npc.ips, npc.nsSelectors)
 		if err != nil {
 			return err
 		}
@@ -85,21 +85,10 @@ func (npc *controller) withNS(name string, f func(ns *ns) error) error {
 	return nil
 }
 
-func (npc *controller) checkLocalPod(obj *coreapi.Pod, event string) bool {
-	if obj.Spec.NodeName != npc.nodeName {
-		common.Log.Debugf("%s ignored for pod %s/%s on node %s", event, obj.ObjectMeta.Namespace, obj.ObjectMeta.Name, obj.Spec.NodeName)
-		return false
-	}
-	return true
-}
-
 func (npc *controller) AddPod(obj *coreapi.Pod) error {
 	npc.Lock()
 	defer npc.Unlock()
 
-	if !npc.checkLocalPod(obj, "AddPod") {
-		return nil
-	}
 	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")
@@ -110,10 +99,6 @@ func (npc *controller) UpdatePod(oldObj, newObj *coreapi.Pod) error {
 	npc.Lock()
 	defer npc.Unlock()
 
-	// Pods don't move across nodes so we only check the new version
-	if !npc.checkLocalPod(newObj, "UpdatePod") {
-		return nil
-	}
 	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")
@@ -124,9 +109,6 @@ func (npc *controller) DeletePod(obj *coreapi.Pod) error {
 	npc.Lock()
 	defer npc.Unlock()
 
-	if !npc.checkLocalPod(obj, "DeletePod") {
-		return nil
-	}
 	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")

npc/namespace.go

@@ -19,6 +19,7 @@ type ns struct {
 	ips ipset.Interface    // interface to ipset
 
 	name      string                               // k8s Namespace name
+	nodeName  string                               // my node name
 	namespace *coreapi.Namespace                   // k8s Namespace object
 	pods      map[types.UID]*coreapi.Pod           // k8s Pod objects by UID
 	policies  map[types.UID]*extnapi.NetworkPolicy // k8s NetworkPolicy objects by UID
@@ -31,7 +32,7 @@ type ns struct {
 	rules        *ruleSet
 }
 
-func newNS(name string, ipt *iptables.IPTables, ips ipset.Interface, nsSelectors *selectorSet) (*ns, error) {
+func newNS(name, nodeName string, ipt *iptables.IPTables, ips ipset.Interface, nsSelectors *selectorSet) (*ns, error) {
 	allPods, err := newSelectorSpec(&unversioned.LabelSelector{}, name, ipset.HashIP)
 	if err != nil {
 		return nil, err
@@ -41,6 +42,7 @@ func newNS(name string, ipt *iptables.IPTables, ips ipset.Interface, nsSelectors
 		ipt:         ipt,
 		ips:         ips,
 		name:        name,
+		nodeName:    nodeName,
 		pods:        make(map[types.UID]*coreapi.Pod),
 		policies:    make(map[types.UID]*extnapi.NetworkPolicy),
 		uid:         uuid.NewUUID(),
@@ -81,6 +83,13 @@ func (ns *ns) onNewPodSelector(selector *selector) error {
 	return nil
 }
 
+func (ns *ns) checkLocalPod(obj *coreapi.Pod) bool {
+	if obj.Spec.NodeName != ns.nodeName {
+		return false
+	}
+	return true
+}
+
 func (ns *ns) addPod(obj *coreapi.Pod) error {
 	ns.pods[obj.ObjectMeta.UID] = obj
 
@@ -88,7 +97,9 @@ func (ns *ns) addPod(obj *coreapi.Pod) error {
 		return nil
 	}
 
-	ns.ips.AddEntry(LocalIpset, obj.Status.PodIP)
+	if ns.checkLocalPod(obj) {
+		ns.ips.AddEntry(LocalIpset, obj.Status.PodIP)
+	}
 	return ns.podSelectors.addToMatching(obj.ObjectMeta.Labels, obj.Status.PodIP)
 }
 
@@ -101,12 +112,16 @@ func (ns *ns) updatePod(oldObj, newObj *coreapi.Pod) error {
 	}
 
 	if hasIP(oldObj) && !hasIP(newObj) {
-		ns.ips.DelEntry(LocalIpset, oldObj.Status.PodIP)
+		if ns.checkLocalPod(oldObj) {
+			ns.ips.DelEntry(LocalIpset, oldObj.Status.PodIP)
+		}
 		return ns.podSelectors.delFromMatching(oldObj.ObjectMeta.Labels, oldObj.Status.PodIP)
 	}
 
 	if !hasIP(oldObj) && hasIP(newObj) {
-		ns.ips.AddEntry(LocalIpset, newObj.Status.PodIP)
+		if ns.checkLocalPod(newObj) {
+			ns.ips.AddEntry(LocalIpset, newObj.Status.PodIP)
+		}
 		return ns.podSelectors.addToMatching(newObj.ObjectMeta.Labels, newObj.Status.PodIP)
 	}
 
@@ -142,7 +157,9 @@ func (ns *ns) deletePod(obj *coreapi.Pod) error {
 		return nil
 	}
 
-	ns.ips.DelEntry(LocalIpset, obj.Status.PodIP)
+	if ns.checkLocalPod(obj) {
+		ns.ips.DelEntry(LocalIpset, obj.Status.PodIP)
+	}
 	return ns.podSelectors.delFromMatching(obj.ObjectMeta.Labels, obj.Status.PodIP)
 }