[NPM] Caching Pod information to compare against desired state

npm/namespace.go

@@ -9,21 +9,21 @@ import (
 	"github.com/Azure/azure-container-networking/npm/ipsm"
 	"github.com/Azure/azure-container-networking/npm/iptm"
 	"github.com/Azure/azure-container-networking/npm/util"
-	"k8s.io/apimachinery/pkg/types"
 
 	corev1 "k8s.io/api/core/v1"
 	networkingv1 "k8s.io/api/networking/v1"
 )
 
 type namespace struct {
-	name           string
-	labelsMap      map[string]string
-	setMap         map[string]string
-	podMap         map[types.UID]*corev1.Pod
-	rawNpMap       map[string]*networkingv1.NetworkPolicy
-	processedNpMap map[string]*networkingv1.NetworkPolicy
-	ipsMgr         *ipsm.IpsetManager
-	iptMgr         *iptm.IptablesManager
+	name            string
+	labelsMap       map[string]string // NameSpace labels
+	setMap          map[string]string
+	podMap          map[string]*npmPod // Key is PodUID
+	rawNpMap        map[string]*networkingv1.NetworkPolicy
+	processedNpMap  map[string]*networkingv1.NetworkPolicy
+	ipsMgr          *ipsm.IpsetManager
+	iptMgr          *iptm.IptablesManager
+	resourceVersion uint64 // NameSpace ResourceVersion
 }
 
 // newNS constructs a new namespace object.
@@ -32,16 +32,24 @@ func newNs(name string) (*namespace, error) {
 		name:           name,
 		labelsMap:      make(map[string]string),
 		setMap:         make(map[string]string),
-		podMap:         make(map[types.UID]*corev1.Pod),
+		podMap:         make(map[string]*npmPod),
 		rawNpMap:       make(map[string]*networkingv1.NetworkPolicy),
 		processedNpMap: make(map[string]*networkingv1.NetworkPolicy),
 		ipsMgr:         ipsm.NewIpsetManager(),
 		iptMgr:         iptm.NewIptablesManager(),
+		// resource version is converted to uint64
+		// so make sure it is initialized to "0"
+		resourceVersion: 0,
 	}
 
 	return ns, nil
 }
 
+// setResourceVersion setter func for RV
+func setResourceVersion(nsObj *namespace, rv string) {
+	nsObj.resourceVersion = util.ParseResourceVersion(rv)
+}
+
 func isSystemNs(nsObj *corev1.Namespace) bool {
 	return nsObj.ObjectMeta.Name == util.KubeSystemFlag
 }
@@ -56,10 +64,22 @@ func isInvalidNamespaceUpdate(oldNsObj, newNsObj *corev1.Namespace) (isInvalidUp
 }
 
 func (ns *namespace) policyExists(npObj *networkingv1.NetworkPolicy) bool {
-	if np, exists := ns.rawNpMap[npObj.ObjectMeta.Name]; exists {
-		if isSamePolicy(np, npObj) {
-			return true
-		}
+	np, exists := ns.rawNpMap[npObj.ObjectMeta.Name]
+	if !exists {
+		return false
+	}
+
+	if !util.CompareResourceVersions(np.ObjectMeta.ResourceVersion, npObj.ObjectMeta.ResourceVersion) {
+		log.Logf("Cached Network Policy has larger ResourceVersion number than new Obj. Name: %s Cached RV: %d New RV: %d\n",
+			npObj.ObjectMeta.Name,
+			np.ObjectMeta.ResourceVersion,
+			npObj.ObjectMeta.ResourceVersion,
+		)
+		return true
+	}
+
+	if isSamePolicy(np, npObj) {
+		return true
 	}
 
 	return false
@@ -103,7 +123,7 @@ func (npMgr *NetworkPolicyManager) UninitAllNsList() error {
 func (npMgr *NetworkPolicyManager) AddNamespace(nsObj *corev1.Namespace) error {
 	var err error
 
-	nsName, nsLabel := "ns-"+nsObj.ObjectMeta.Name, nsObj.ObjectMeta.Labels
+	nsName, nsLabel := util.GetNSNameWithPrefix(nsObj.ObjectMeta.Name), nsObj.ObjectMeta.Labels
 	log.Logf("NAMESPACE CREATING: [%s/%v]", nsName, nsLabel)
 
 	ipsMgr := npMgr.nsMap[util.KubeAllNamespacesFlag].ipsMgr
@@ -121,14 +141,14 @@ func (npMgr *NetworkPolicyManager) AddNamespace(nsObj *corev1.Namespace) error {
 	// Add the namespace to its label's ipset list.
 	nsLabels := nsObj.ObjectMeta.Labels
 	for nsLabelKey, nsLabelVal := range nsLabels {
-		labelKey := "ns-" + nsLabelKey
+		labelKey := util.GetNSNameWithPrefix(nsLabelKey)
 		log.Logf("Adding namespace %s to ipset list %s", nsName, labelKey)
 		if err = ipsMgr.AddToList(labelKey, nsName); err != nil {
 			log.Errorf("Error: failed to add namespace %s to ipset list %s", nsName, labelKey)
 			return err
 		}
 
-		label := "ns-" + nsLabelKey + ":" + nsLabelVal
+		label := util.GetNSNameWithPrefix(nsLabelKey + ":" + nsLabelVal)
 		log.Logf("Adding namespace %s to ipset list %s", nsName, label)
 		if err = ipsMgr.AddToList(label, nsName); err != nil {
 			log.Errorf("Error: failed to add namespace %s to ipset list %s", nsName, label)
@@ -140,6 +160,7 @@ func (npMgr *NetworkPolicyManager) AddNamespace(nsObj *corev1.Namespace) error {
 	if err != nil {
 		log.Errorf("Error: failed to create namespace %s", nsName)
 	}
+	setResourceVersion(ns, nsObj.GetObjectMeta().GetResourceVersion())
 
 	// Append all labels to the cache NS obj
 	ns.labelsMap = util.AppendMap(ns.labelsMap, nsLabel)
@@ -155,8 +176,8 @@ func (npMgr *NetworkPolicyManager) UpdateNamespace(oldNsObj *corev1.Namespace, n
 	}
 
 	var err error
-	oldNsNs, oldNsLabel := "ns-"+oldNsObj.ObjectMeta.Name, oldNsObj.ObjectMeta.Labels
-	newNsNs, newNsLabel := "ns-"+newNsObj.ObjectMeta.Name, newNsObj.ObjectMeta.Labels
+	oldNsNs, oldNsLabel := util.GetNSNameWithPrefix(oldNsObj.ObjectMeta.Name), oldNsObj.ObjectMeta.Labels
+	newNsNs, newNsLabel := util.GetNSNameWithPrefix(newNsObj.ObjectMeta.Name), newNsObj.ObjectMeta.Labels
 	log.Logf(
 		"NAMESPACE UPDATING:\n old namespace: [%s/%v]\n new namespace: [%s/%v]",
 		oldNsNs, oldNsLabel, newNsNs, newNsLabel,
@@ -189,6 +210,16 @@ func (npMgr *NetworkPolicyManager) UpdateNamespace(oldNsObj *corev1.Namespace, n
 		return nil
 	}
 
+	newRv := util.ParseResourceVersion(newNsObj.ObjectMeta.ResourceVersion)
+	if !util.CompareUintResourceVersions(curNsObj.resourceVersion, newRv) {
+		log.Logf("Cached NameSpace has larger ResourceVersion number than new Obj. NameSpace: %s Cached RV: %d New RV:\n",
+			oldNsNs,
+			curNsObj.resourceVersion,
+			newRv,
+		)
+		return nil
+	}
+
 	//if no change in labels then return
 	if reflect.DeepEqual(curNsObj.labelsMap, newNsLabel) {
 		log.Logf(
@@ -199,45 +230,32 @@ func (npMgr *NetworkPolicyManager) UpdateNamespace(oldNsObj *corev1.Namespace, n
 	}
 
 	//If the Namespace is not deleted, delete removed labels and create new labels
-	toAddNsLabels, toDeleteNsLabels := util.CompareMapDiff(curNsObj.labelsMap, newNsLabel)
+	addToIPSets, deleteFromIPSets := util.GetIPSetListCompareLabels(curNsObj.labelsMap, newNsLabel)
 
 	// Delete the namespace from its label's ipset list.
 	ipsMgr := npMgr.nsMap[util.KubeAllNamespacesFlag].ipsMgr
-	for nsLabelKey, nsLabelVal := range toDeleteNsLabels {
-		labelKey := "ns-" + nsLabelKey
+	for _, nsLabelVal := range deleteFromIPSets {
+		labelKey := util.GetNSNameWithPrefix(nsLabelVal)
 		log.Logf("Deleting namespace %s from ipset list %s", oldNsNs, labelKey)
 		if err = ipsMgr.DeleteFromList(labelKey, oldNsNs); err != nil {
 			log.Errorf("Error: failed to delete namespace %s from ipset list %s", oldNsNs, labelKey)
 			return err
 		}
-
-		label := "ns-" + nsLabelKey + ":" + nsLabelVal
-		log.Logf("Deleting namespace %s from ipset list %s", oldNsNs, label)
-		if err = ipsMgr.DeleteFromList(label, oldNsNs); err != nil {
-			log.Errorf("Error: failed to delete namespace %s from ipset list %s", oldNsNs, label)
-			return err
-		}
 	}
 
 	// Add the namespace to its label's ipset list.
-	for nsLabelKey, nsLabelVal := range toAddNsLabels {
-		labelKey := "ns-" + nsLabelKey
+	for _, nsLabelVal := range addToIPSets {
+		labelKey := util.GetNSNameWithPrefix(nsLabelVal)
 		log.Logf("Adding namespace %s to ipset list %s", oldNsNs, labelKey)
 		if err = ipsMgr.AddToList(labelKey, oldNsNs); err != nil {
 			log.Errorf("Error: failed to add namespace %s to ipset list %s", oldNsNs, labelKey)
 			return err
 		}
-
-		label := "ns-" + nsLabelKey + ":" + nsLabelVal
-		log.Logf("Adding namespace %s to ipset list %s", oldNsNs, label)
-		if err = ipsMgr.AddToList(label, oldNsNs); err != nil {
-			log.Errorf("Error: failed to add namespace %s to ipset list %s", oldNsNs, label)
-			return err
-		}
 	}
 
 	// Append all labels to the cache NS obj
 	curNsObj.labelsMap = util.ClearAndAppendMap(curNsObj.labelsMap, newNsLabel)
+	setResourceVersion(curNsObj, newNsObj.GetObjectMeta().GetResourceVersion())
 	npMgr.nsMap[newNsNs] = curNsObj
 
 	return nil
@@ -247,26 +265,27 @@ func (npMgr *NetworkPolicyManager) UpdateNamespace(oldNsObj *corev1.Namespace, n
 func (npMgr *NetworkPolicyManager) DeleteNamespace(nsObj *corev1.Namespace) error {
 	var err error
 
-	nsName, nsLabel := "ns-"+nsObj.ObjectMeta.Name, nsObj.ObjectMeta.Labels
+	nsName, nsLabel := util.GetNSNameWithPrefix(nsObj.ObjectMeta.Name), nsObj.ObjectMeta.Labels
 	log.Logf("NAMESPACE DELETING: [%s/%v]", nsName, nsLabel)
 
-	_, exists := npMgr.nsMap[nsName]
+	cachedNsObj, exists := npMgr.nsMap[nsName]
 	if !exists {
 		return nil
 	}
 
+	log.Logf("NAMESPACE DELETING cached labels: [%s/%v]", nsName, cachedNsObj.labelsMap)
 	// Delete the namespace from its label's ipset list.
 	ipsMgr := npMgr.nsMap[util.KubeAllNamespacesFlag].ipsMgr
-	nsLabels := nsObj.ObjectMeta.Labels
+	nsLabels := cachedNsObj.labelsMap
 	for nsLabelKey, nsLabelVal := range nsLabels {
-		labelKey := "ns-" + nsLabelKey
+		labelKey := util.GetNSNameWithPrefix(nsLabelKey)
 		log.Logf("Deleting namespace %s from ipset list %s", nsName, labelKey)
 		if err = ipsMgr.DeleteFromList(labelKey, nsName); err != nil {
 			log.Errorf("Error: failed to delete namespace %s from ipset list %s", nsName, labelKey)
 			return err
 		}
 
-		label := "ns-" + nsLabelKey + ":" + nsLabelVal
+		label := util.GetNSNameWithPrefix(nsLabelKey + ":" + nsLabelVal)
 		log.Logf("Deleting namespace %s from ipset list %s", nsName, label)
 		if err = ipsMgr.DeleteFromList(label, nsName); err != nil {
 			log.Errorf("Error: failed to delete namespace %s from ipset list %s", nsName, label)

npm/namespace_test.go

@@ -16,7 +16,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
-func TestnewNs(t *testing.T) {
+func TestNewNs(t *testing.T) {
 	if _, err := newNs("test"); err != nil {
 		t.Errorf("TestnewNs failed @ newNs")
 	}
@@ -165,6 +165,7 @@ func TestAddNamespaceLabel(t *testing.T) {
 			Labels: map[string]string{
 				"app": "old-test-namespace",
 			},
+			ResourceVersion: "0",
 		},
 	}
 
@@ -175,6 +176,8 @@ func TestAddNamespaceLabel(t *testing.T) {
 				"app":    "old-test-namespace",
 				"update": "true",
 			},
+
+			ResourceVersion: "1",
 		},
 	}
 
@@ -225,6 +228,7 @@ func TestDeleteandUpdateNamespaceLabel(t *testing.T) {
 				"update": "true",
 				"group":  "test",
 			},
+			ResourceVersion: "0",
 		},
 	}
 
@@ -235,6 +239,7 @@ func TestDeleteandUpdateNamespaceLabel(t *testing.T) {
 				"app":    "old-test-namespace",
 				"update": "false",
 			},
+			ResourceVersion: "1",
 		},
 	}
 

npm/npm.go

@@ -50,7 +50,6 @@ type NetworkPolicyManager struct {
 
 	nodeName                     string
 	nsMap                        map[string]*namespace
-	podMap                       map[string]string // Key: Pod uuid, Value: PodIp
 	isAzureNpmChainCreated       bool
 	isSafeToCleanUpAzureNpmChain bool
 
@@ -118,14 +117,13 @@ func (npMgr *NetworkPolicyManager) SendClusterMetrics() {
 	for {
 		<-heartbeat
 		npMgr.Lock()
-		podCount.Value = float64(len(npMgr.podMap))
+		podCount.Value = 0
 		//Reducing one to remove all-namespaces ns obj
 		nsCount.Value = float64(len(npMgr.nsMap) - 1)
-		nwPolCount := 0
 		for _, ns := range npMgr.nsMap {
-			nwPolCount = nwPolCount + len(ns.rawNpMap)
+			nwPolicyCount.Value += float64(len(ns.rawNpMap))
+			podCount.Value += float64(len(ns.podMap))
 		}
-		nwPolicyCount.Value = float64(nwPolCount)
 		npMgr.Unlock()
 
 		metrics.SendMetric(podCount)
@@ -234,7 +232,6 @@ func NewNetworkPolicyManager(clientset *kubernetes.Clientset, informerFactory in
 		npInformer:                   npInformer,
 		nodeName:                     os.Getenv("HOSTNAME"),
 		nsMap:                        make(map[string]*namespace),
-		podMap:                       make(map[string]string),
 		isAzureNpmChainCreated:       false,
 		isSafeToCleanUpAzureNpmChain: false,
 		clusterState: telemetry.ClusterState{
@@ -251,7 +248,7 @@ func NewNetworkPolicyManager(clientset *kubernetes.Clientset, informerFactory in
 	npMgr.nsMap[util.KubeAllNamespacesFlag] = allNs
 
 	// Create ipset for the namespace.
-	kubeSystemNs := "ns-" + util.KubeSystemFlag
+	kubeSystemNs := util.GetNSNameWithPrefix(util.KubeSystemFlag)
 	if err := allNs.ipsMgr.CreateSet(kubeSystemNs, append([]string{util.IpsetNetHashFlag})); err != nil {
 		metrics.SendErrorLogAndMetric(util.NpmID, "Error: failed to create ipset for namespace %s.", kubeSystemNs)
 	}
@@ -269,19 +266,14 @@ func NewNetworkPolicyManager(clientset *kubernetes.Clientset, informerFactory in
 				npMgr.AddPod(podObj)
 				npMgr.Unlock()
 			},
-			UpdateFunc: func(old, new interface{}) {
-				oldPodObj, ok := old.(*corev1.Pod)
-				if !ok {
-					metrics.SendErrorLogAndMetric(util.NpmID, "UPDATE Pod: Received unexpected old object type: %v", oldPodObj)
-					return
-				}
+			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(oldPodObj, newPodObj)
+				npMgr.UpdatePod(newPodObj)
 				npMgr.Unlock()
 			},
 			DeleteFunc: func(obj interface{}) {

npm/nwpolicy.go

@@ -33,7 +33,7 @@ func (npMgr *NetworkPolicyManager) AddNetworkPolicy(npObj *networkingv1.NetworkP
 		err    error
 		ns     *namespace
 		exists bool
-		npNs   = "ns-" + npObj.ObjectMeta.Namespace
+		npNs   = util.GetNSNameWithPrefix(npObj.ObjectMeta.Namespace)
 		npName = npObj.ObjectMeta.Name
 		allNs  = npMgr.nsMap[util.KubeAllNamespacesFlag]
 		timer  = metrics.StartNewTimer()
@@ -153,7 +153,7 @@ func (npMgr *NetworkPolicyManager) DeleteNetworkPolicy(npObj *networkingv1.Netwo
 		allNs = npMgr.nsMap[util.KubeAllNamespacesFlag]
 	)
 
-	npNs, npName := "ns-"+npObj.ObjectMeta.Namespace, npObj.ObjectMeta.Name
+	npNs, npName := util.GetNSNameWithPrefix(npObj.ObjectMeta.Namespace), npObj.ObjectMeta.Name
 	log.Logf("NETWORK POLICY DELETING: Namespace: %s, Name:%s", npNs, npName)
 
 	var exists bool