Network policy pod watch changes

- Create pod watch for all namespaces instead of pod watch for each namespace
- Use shared informers for pod watch

Author: Ravi Sankar Penta

pkg/sdn/plugin/networkpolicy.go

@@ -37,6 +37,7 @@ type networkPolicyPlugin struct {
 	lock        sync.Mutex
 	namespaces  map[uint32]*npNamespace
 	kNamespaces map[string]kapi.Namespace
+	pods        map[ktypes.UID]kapi.Pod
 
 	kubeInformers kinternalinformers.SharedInformerFactory
 }
@@ -50,9 +51,6 @@ type npNamespace struct {
 	inUse    bool
 
 	policies map[ktypes.UID]*npPolicy
-
-	pods         map[ktypes.UID]kapi.Pod
-	stopPodWatch chan struct{}
 }
 
 // npPolicy is a parsed version of a single NetworkPolicy object
@@ -68,6 +66,7 @@ func NewNetworkPolicyPlugin() osdnPolicy {
 	return &networkPolicyPlugin{
 		namespaces:  make(map[uint32]*npNamespace),
 		kNamespaces: make(map[string]kapi.Namespace),
+		pods:        make(map[ktypes.UID]kapi.Pod),
 	}
 }
 
@@ -101,6 +100,7 @@ func (np *networkPolicyPlugin) Start(node *OsdnNode) error {
 	}
 
 	np.watchNamespaces()
+	np.watchPods()
 	go utilwait.Forever(np.watchNetworkPolicies, 0)
 	return nil
 }
@@ -252,92 +252,6 @@ func (np *networkPolicyPlugin) UnrefVNID(vnid uint32) {
 	np.syncNamespace(npns)
 }
 
-// watchPods watches Pod changes in npns until stopPodWatch is triggered. pods
-// and stopPodWatch are passed in as arguments rather than being read from npns
-// because it's possible another thread will already have cancelled the watch
-// (and changed the npns fields) before this function runs.
-func (np *networkPolicyPlugin) watchPods(npns *npNamespace, pods map[ktypes.UID]kapi.Pod, stopPodWatch chan struct{}) {
-	RunNamespacedPodEventQueue(np.node.kClient.Core().RESTClient(), npns.name, stopPodWatch, func(delta cache.Delta) error {
-		pod := delta.Object.(*kapi.Pod)
-		glog.V(5).Infof("Watch %s event for Pod %s/%s", delta.Type, pod.Namespace, pod.Name)
-
-		// We don't want to grab np.namespacesLock for every Pod.Status change...
-		// But it's safe to look up oldPod without locking here because no other
-		// threads modify this map.
-		oldPod, podExisted := pods[pod.UID]
-		if pod.Status.PodIP == "" {
-			delta.Type = cache.Deleted
-		}
-		switch delta.Type {
-		case cache.Sync, cache.Added, cache.Updated:
-			if podExisted && oldPod.Status.PodIP == pod.Status.PodIP && reflect.DeepEqual(oldPod.Labels, pod.Labels) {
-				return nil
-			}
-		case cache.Deleted:
-			if !podExisted {
-				return nil
-			}
-		}
-
-		glog.V(5).Infof("Re-checking policies after pod %s", delta.Type)
-		np.lock.Lock()
-		defer np.lock.Unlock()
-
-		// RunNamespacedPodEventQueue() will call this function at least once more
-		// after the watch is stopped, so verify that our watch is still running
-		// before changing anything.
-		if stopPodWatch != npns.stopPodWatch {
-			return nil
-		}
-
-		if delta.Type == cache.Deleted {
-			delete(pods, pod.UID)
-		} else {
-			pods[pod.UID] = *pod
-		}
-
-		changed := false
-		for _, npp := range npns.policies {
-			if npp.watchesPods {
-				if np.updateNetworkPolicy(npns, &npp.policy) {
-					changed = true
-				}
-			}
-		}
-		if changed {
-			np.syncNamespace(npns)
-		}
-
-		return nil
-	})
-}
-
-func (np *networkPolicyPlugin) podWatchUntilStopped(npns *npNamespace) {
-	pods := npns.pods
-	stop := npns.stopPodWatch
-	go utilwait.Until(func() { np.watchPods(npns, pods, stop) }, 0, stop)
-}
-
-func (np *networkPolicyPlugin) updatePodWatch(npns *npNamespace) {
-	watchesPods := false
-	for _, npp := range npns.policies {
-		if npp.watchesPods {
-			watchesPods = true
-			break
-		}
-	}
-
-	if watchesPods && (npns.stopPodWatch == nil) {
-		npns.pods = make(map[ktypes.UID]kapi.Pod)
-		npns.stopPodWatch = make(chan struct{})
-		np.podWatchUntilStopped(npns)
-	} else if !watchesPods && (npns.stopPodWatch != nil) {
-		close(npns.stopPodWatch)
-		npns.stopPodWatch = nil
-		npns.pods = nil
-	}
-}
-
 func (np *networkPolicyPlugin) selectNamespaces(lsel *metav1.LabelSelector) []uint32 {
 	vnids := []uint32{}
 	sel, err := metav1.LabelSelectorAsSelector(lsel)
@@ -364,7 +278,7 @@ func (np *networkPolicyPlugin) selectPods(npns *npNamespace, lsel *metav1.LabelS
 		glog.Errorf("ValidateNetworkPolicy() failure! Invalid PodSelector: %v", err)
 		return ips
 	}
-	for _, pod := range npns.pods {
+	for _, pod := range np.pods {
 		if sel.Matches(labels.Set(pod.Labels)) {
 			ips = append(ips, pod.Status.PodIP)
 		}
@@ -464,7 +378,6 @@ func (np *networkPolicyPlugin) updateNetworkPolicy(npns *npNamespace, policy *ex
 
 	oldNPP, existed := npns.policies[policy.UID]
 	npns.policies[policy.UID] = npp
-	np.updatePodWatch(npns)
 
 	changed := !existed || !reflect.DeepEqual(oldNPP.flows, npp.flows)
 	if !changed {
@@ -540,6 +453,81 @@ func namespaceIsIsolated(ns *kapi.Namespace) bool {
 	}
 }
 
+func (np *networkPolicyPlugin) watchPods() {
+	podInformer := np.kubeInformers.Core().InternalVersion().Pods()
+	podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc: func(obj interface{}) {
+			pod := obj.(*kapi.Pod)
+			np.handleAddOrUpdatePod(pod, watch.Added)
+		},
+		UpdateFunc: func(_, cur interface{}) {
+			pod := cur.(*kapi.Pod)
+			np.handleAddOrUpdatePod(pod, watch.Modified)
+		},
+		DeleteFunc: func(obj interface{}) {
+			obj, err := getDeletedObjFromInformer(obj, Pods)
+			if err != nil {
+				glog.Error(err)
+				return
+			}
+			pod := obj.(*kapi.Pod)
+			np.handleDeletePod(pod)
+		},
+	})
+}
+
+func (np *networkPolicyPlugin) handleAddOrUpdatePod(pod *kapi.Pod, eventType watch.EventType) {
+	glog.V(5).Infof("Watch %s event for Pod '%s/%s'", eventType, pod.Namespace, pod.Name)
+
+	// We don't want to grab np.Lock for every Pod.Status change...
+	// But it's safe to look up oldPod without locking here because no other
+	// threads modify this map.
+	oldPod, podExisted := np.pods[pod.UID]
+	if pod.Status.PodIP == "" {
+		return
+	} else if podExisted && oldPod.Status.PodIP == pod.Status.PodIP && reflect.DeepEqual(oldPod.Labels, pod.Labels) {
+		return
+	}
+
+	np.lock.Lock()
+	defer np.lock.Unlock()
+
+	np.pods[pod.UID] = *pod
+	np.refreshNetworkPolicies()
+}
+
+func (np *networkPolicyPlugin) handleDeletePod(pod *kapi.Pod) {
+	glog.V(5).Infof("Watch %s event for Pod '%s/%s'", watch.Deleted, pod.Namespace, pod.Name)
+
+	_, podExisted := np.pods[pod.UID]
+	if !podExisted || (pod.Status.PodIP == "") {
+		return
+	}
+
+	np.lock.Lock()
+	defer np.lock.Unlock()
+
+	delete(np.pods, pod.UID)
+	np.refreshNetworkPolicies()
+}
+
+func (np *networkPolicyPlugin) refreshNetworkPolicies() {
+	for _, npns := range np.namespaces {
+		changed := false
+		for _, npp := range npns.policies {
+			if npp.watchesPods {
+				if np.updateNetworkPolicy(npns, &npp.policy) {
+					changed = true
+					break
+				}
+			}
+		}
+		if changed {
+			np.syncNamespace(npns)
+		}
+	}
+}
+
 func (np *networkPolicyPlugin) watchNamespaces() {
 	nsInformer := np.kubeInformers.Core().InternalVersion().Namespaces()
 	nsInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{