[NPM] Support pod controller and unit test to improve reliability (#836)

* Junguk cho pod controller support (#1)

* code layout to support pod controller in npm

* filter events if they do not need to handle and clean up business logic

* put lock when it needs to access shared resource
use namespace/name as key
clean up functions

* - use pod key instead of uid.
- remove unnecessary error check

* use namespace prefix in pod namespace. add log messages to know what events happens.

* move event logs with more contexts in needsync func

* Put returning an error in a right place

* Return if the RV of both pod obj are the same. Proactively start cleaning up pod when the pod is deleted.

* first version of podController UT

* add ipset management in pod controller unit test

* Make methods flexible for ipset store and restore operation

* clean up functions and variables

Co-authored-by: Junguk Cho <[email protected]>

* correct and clean functions and error messages. Return errors from appendNamedPortIpsets function to retry syncPod operation

* Check npmPod exists in cleanUpDeletedPod function. Use GetIPSetListFromLabels in syncAddedPod and cleanUpDeletedPod functions. Correct error messages, functions, etc

* Clean up podController code. Make podController UT more flexible.

* clean up appendNamedPortIpsets to improve readibility

* minor update (v1 -> corev1) for consistency

* clean up syncPod code to compare last applied states and new pod's states

* add validation for casting old pod. correct log message

* delete unneeded codes

* minor fix: correct comments and removed unneeded variables

* add pre-filter codes to avoid unnecessary reconcile process in updatePod event

* correct a comment

* check workqueue length to validate case where it does not need to reconcile in unit test

Author: JungukCho

npm/ipsm/ipsm.go

@@ -32,7 +32,7 @@ type IpsetManager struct {
 // Ipset represents one ipset entry.
 type Ipset struct {
 	name       string
-	elements   map[string]string // key = ip, value: context associated to the ip like podUid
+	elements   map[string]string // key = ip, value: context associated to the ip like podKey
 	referCount int
 }
 
@@ -297,16 +297,16 @@ func (ipsMgr *IpsetManager) DeleteSet(setName string) error {
 }
 
 // AddToSet inserts an ip to an entry in setMap, and creates/updates the corresponding ipset.
-func (ipsMgr *IpsetManager) AddToSet(setName, ip, spec, podUid string) error {
+func (ipsMgr *IpsetManager) AddToSet(setName, ip, spec, podKey string) error {
 	if ipsMgr.Exists(setName, ip, spec) {
 
-		// make sure we have updated the podUid in case it gets changed
-		cachedPodUid := ipsMgr.SetMap[setName].elements[ip]
-		if cachedPodUid != podUid {
-			log.Logf("AddToSet: PodOwner has changed for Ip: %s, setName:%s, Old podUid: %s, new PodUid: %s. Replace context with new PodOwner.",
-				ip, setName, cachedPodUid, podUid)
+		// make sure we have updated the podKey in case it gets changed
+		cachedPodKey := ipsMgr.SetMap[setName].elements[ip]
+		if cachedPodKey != podKey {
+			log.Logf("AddToSet: PodOwner has changed for Ip: %s, setName:%s, Old podKey: %s, new podKey: %s. Replace context with new PodOwner.",
+				ip, setName, cachedPodKey, podKey)
 
-			ipsMgr.SetMap[setName].elements[ip] = podUid
+			ipsMgr.SetMap[setName].elements[ip] = podKey
 		}
 
 		return nil
@@ -351,8 +351,8 @@ func (ipsMgr *IpsetManager) AddToSet(setName, ip, spec, podUid string) error {
 		return err
 	}
 
-	// Stores the podUid as the context for this ip.
-	ipsMgr.SetMap[setName].elements[ip] = podUid
+	// Stores the podKey as the context for this ip.
+	ipsMgr.SetMap[setName].elements[ip] = podKey
 
 	metrics.NumIPSetEntries.Inc()
 	metrics.IncIPSetInventory(setName)
@@ -361,7 +361,7 @@ func (ipsMgr *IpsetManager) AddToSet(setName, ip, spec, podUid string) error {
 }
 
 // DeleteFromSet removes an ip from an entry in setMap, and delete/update the corresponding ipset.
-func (ipsMgr *IpsetManager) DeleteFromSet(setName, ip, podUid string) error {
+func (ipsMgr *IpsetManager) DeleteFromSet(setName, ip, podKey string) error {
 	ipSet, exists := ipsMgr.SetMap[setName]
 	if !exists {
 		log.Logf("ipset with name %s not found", setName)
@@ -380,10 +380,10 @@ func (ipsMgr *IpsetManager) DeleteFromSet(setName, ip, podUid string) error {
 
 	if _, exists := ipsMgr.SetMap[setName].elements[ip]; exists {
 		// in case the IP belongs to a new Pod, then ignore this Delete call as this might be stale
-		cachedPodUid := ipSet.elements[ip]
-		if cachedPodUid != podUid {
-			log.Logf("DeleteFromSet: PodOwner has changed for Ip: %s, setName:%s, Old podUid: %s, new PodUid: %s. Ignore the delete as this is stale update",
-				ip, setName, cachedPodUid, podUid)
+		cachedPodKey := ipSet.elements[ip]
+		if cachedPodKey != podKey {
+			log.Logf("DeleteFromSet: PodOwner has changed for Ip: %s, setName:%s, Old podKey: %s, new podKey: %s. Ignore the delete as this is stale update",
+				ip, setName, cachedPodKey, podKey)
 
 			return nil
 		}

npm/npm.go

@@ -45,15 +45,17 @@ type NetworkPolicyManager struct {
 	sync.Mutex
 	clientset *kubernetes.Clientset
 
-	informerFactory     informers.SharedInformerFactory
-	podInformer         coreinformers.PodInformer
+	informerFactory informers.SharedInformerFactory
+	podInformer     coreinformers.PodInformer
+	podController   *podController
+
 	nsInformer          coreinformers.NamespaceInformer
 	npInformer          networkinginformers.NetworkPolicyInformer
 	nameSpaceController *nameSpaceController
 
 	NodeName                     string
 	NsMap                        map[string]*Namespace
-	PodMap                       map[string]*NpmPod                     // Key is ns-<nsname>/<podname>/<poduuid>
+	PodMap                       map[string]*NpmPod                     // Key is <nsname>/<podname>
 	RawNpMap                     map[string]*networkingv1.NetworkPolicy // Key is ns-<nsname>/<policyname>
 	ProcessedNpMap               map[string]*networkingv1.NetworkPolicy // Key is ns-<nsname>/<podSelectorHash>
 	isAzureNpmChainCreated       bool
@@ -186,6 +188,9 @@ func (npMgr *NetworkPolicyManager) Start(stopCh <-chan struct{}) error {
 		return fmt.Errorf("Network policy informer failed to sync")
 	}
 
+	// TODO: any dependency among below functions?
+	// start pod controller after synced
+	go npMgr.podController.Run(threadness, stopCh)
 	go npMgr.nameSpaceController.Run(threadness, stopCh)
 	go npMgr.reconcileChains()
 	go npMgr.backup()
@@ -261,52 +266,8 @@ func NewNetworkPolicyManager(clientset *kubernetes.Clientset, informerFactory in
 		metrics.SendErrorLogAndMetric(util.NpmID, "Error: failed to create ipset for namespace %s.", kubeSystemNs)
 	}
 
-	podInformer.Informer().AddEventHandler(
-		// Pod event handlers
-		cache.ResourceEventHandlerFuncs{
-			AddFunc: func(obj interface{}) {
-				podObj, ok := obj.(*corev1.Pod)
-				if !ok {
-					metrics.SendErrorLogAndMetric(util.NpmID, "ADD Pod: Received unexpected object type: %v", obj)
-					return
-				}
-				npMgr.Lock()
-				npMgr.AddPod(podObj)
-				npMgr.Unlock()
-			},
-			UpdateFunc: func(_, new interface{}) {
-				newPodObj, ok := new.(*corev1.Pod)
-				if !ok {
-					metrics.SendErrorLogAndMetric(util.NpmID, "UPDATE Pod: Received unexpected new object type: %v", newPodObj)
-					return
-				}
-				npMgr.Lock()
-				npMgr.UpdatePod(newPodObj)
-				npMgr.Unlock()
-			},
-			DeleteFunc: func(obj interface{}) {
-				// DeleteFunc gets the final state of the resource (if it is known).
-				// Otherwise, it gets an object of type DeletedFinalStateUnknown.
-				// This can happen if the watch is closed and misses the delete event and
-				// the controller doesn't notice the deletion until the subsequent re-list
-				podObj, ok := obj.(*corev1.Pod)
-				if !ok {
-					tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
-					if !ok {
-						metrics.SendErrorLogAndMetric(util.NpmID, "DELETE Pod: Received unexpected object type: %v", obj)
-						return
-					}
-					if podObj, ok = tombstone.Obj.(*corev1.Pod); !ok {
-						metrics.SendErrorLogAndMetric(util.NpmID, "DELETE Pod: Received unexpected object type: %v", obj)
-						return
-					}
-				}
-				npMgr.Lock()
-				npMgr.DeletePod(podObj)
-				npMgr.Unlock()
-			},
-		},
-	)
+	// create pod controller
+	npMgr.podController = NewPodController(informerFactory.Core().V1().Pods(), clientset, npMgr)
 
 	// create NameSpace controller
 	npMgr.nameSpaceController = NewNameSpaceController(nsInformer, clientset, npMgr)