Merge eaea1e7 into c56fa7e

pkg/hostagent/agent.go

@@ -45,9 +45,10 @@ type HostAgent struct {
 	config *HostAgentConfig
 	env    Environment
 
-	indexMutex sync.Mutex
-	ipamMutex  sync.Mutex
-	snatMutex  sync.RWMutex
+	indexMutex           sync.Mutex
+	ipamMutex            sync.Mutex
+	snatPolicyLabelMutex sync.RWMutex
+	snatPolicyCacheMutex sync.RWMutex
 
 	opflexEps             map[string][]*opflexEndpoint
 	opflexServices        map[string]*opflexService
@@ -235,39 +236,39 @@ func addPodRoute(ipn types.IPNet, dev string, src string) error {
 }
 
 func (agent *HostAgent) ReadSnatPolicyLabel(key string) (map[string]ResourceType, bool) {
-	agent.snatMutex.RLock()
-	defer agent.snatMutex.RUnlock()
+	agent.snatPolicyLabelMutex.RLock()
+	defer agent.snatPolicyLabelMutex.RUnlock()
 	value, ok := agent.snatPolicyLabels[key]
 	return value, ok
 }
 
 func (agent *HostAgent) WriteSnatPolicyLabel(key string, policy string, res ResourceType) {
-	agent.snatMutex.Lock()
-	defer agent.snatMutex.Unlock()
+	agent.snatPolicyLabelMutex.Lock()
+	defer agent.snatPolicyLabelMutex.Unlock()
 	agent.snatPolicyLabels[key][policy] = res
 }
 
 func (agent *HostAgent) WriteNewSnatPolicyLabel(key string) {
-	agent.snatMutex.Lock()
-	defer agent.snatMutex.Unlock()
+	agent.snatPolicyLabelMutex.Lock()
+	defer agent.snatPolicyLabelMutex.Unlock()
 	agent.snatPolicyLabels[key] = make(map[string]ResourceType)
 }
 
 func (agent *HostAgent) DeleteSnatPolicyLabelEntry(key string, policy string) {
-	agent.snatMutex.Lock()
-	defer agent.snatMutex.Unlock()
+	agent.snatPolicyLabelMutex.Lock()
+	defer agent.snatPolicyLabelMutex.Unlock()
 	delete(agent.snatPolicyLabels[key], policy)
 }
 
 func (agent *HostAgent) DeleteSnatPolicyLabel(key string) {
-	agent.snatMutex.Lock()
-	defer agent.snatMutex.Unlock()
+	agent.snatPolicyLabelMutex.Lock()
+	defer agent.snatPolicyLabelMutex.Unlock()
 	delete(agent.snatPolicyLabels, key)
 }
 
 func (agent *HostAgent) DeleteMatchingSnatPolicyLabel(policy string) {
-	agent.snatMutex.Lock()
-	defer agent.snatMutex.Unlock()
+	agent.snatPolicyLabelMutex.Lock()
+	defer agent.snatPolicyLabelMutex.Unlock()
 	for key, v := range agent.snatPolicyLabels {
 		if _, ok := v[policy]; ok {
 			delete(agent.snatPolicyLabels[key], policy)

pkg/hostagent/snatlocalinfo.go

@@ -49,6 +49,8 @@ func (agent *HostAgent) UpdateLocalInfoCr() bool {
 	for _, ginfo := range ginfos {
 		var localInfo SnatLocalInfo
 		localInfo.snatIp = ginfo.SnatIp
+		agent.snatPolicyCacheMutex.RLock()
+		defer agent.snatPolicyCacheMutex.RUnlock()
 		if _, ok := agent.snatPolicyCache[ginfo.SnatPolicyName]; ok {
 			if len(agent.snatPolicyCache[ginfo.SnatPolicyName].Spec.DestIp) == 0 {
 				localInfo.destIps = []string{"0.0.0.0/0"}

pkg/hostagent/snats.go

@@ -207,8 +207,8 @@ func (agent *HostAgent) initSnatPolicyInformerBase(listWatch *cache.ListWatch) {
 }
 
 func (agent *HostAgent) snatPolicyAdded(obj interface{}) {
-	agent.indexMutex.Lock()
-	defer agent.indexMutex.Unlock()
+	agent.snatPolicyCacheMutex.Lock()
+	defer agent.snatPolicyCacheMutex.Unlock()
 	policyinfo := obj.(*snatpolicy.SnatPolicy)
 	agent.log.Info("Policy Info Added: ", policyinfo.ObjectMeta.Name)
 	if policyinfo.Status.State != snatpolicy.Ready {
@@ -220,8 +220,8 @@ func (agent *HostAgent) snatPolicyAdded(obj interface{}) {
 }
 
 func (agent *HostAgent) snatPolicyUpdated(oldobj interface{}, newobj interface{}) {
-	agent.indexMutex.Lock()
-	defer agent.indexMutex.Unlock()
+	agent.snatPolicyCacheMutex.Lock()
+	defer agent.snatPolicyCacheMutex.Unlock()
 	oldpolicyinfo := oldobj.(*snatpolicy.SnatPolicy)
 	newpolicyinfo := newobj.(*snatpolicy.SnatPolicy)
 	agent.log.Info("Policy Info Updated: ", newpolicyinfo.ObjectMeta.Name)
@@ -299,8 +299,8 @@ func (agent *HostAgent) snatPolicyUpdated(oldobj interface{}, newobj interface{}
 }
 
 func (agent *HostAgent) snatPolicyDeleted(obj interface{}) {
-	agent.indexMutex.Lock()
-	defer agent.indexMutex.Unlock()
+	agent.snatPolicyCacheMutex.Lock()
+	defer agent.snatPolicyCacheMutex.Unlock()
 	policyinfo := obj.(*snatpolicy.SnatPolicy)
 	agent.deletePolicy(policyinfo)
 	delete(agent.snatPolicyCache, policyinfo.ObjectMeta.Name)
@@ -328,7 +328,7 @@ func (agent *HostAgent) handleSnatUpdate(policy *snatpolicy.SnatPolicy) {
 			})
 		// list the pods and apply the policy at service target
 		for _, service := range services {
-			uids, _ := agent.getPodsMatchingObjet(service, policy.ObjectMeta.Name)
+			uids, _ := agent.getPodsMatchingObject(service, policy.ObjectMeta.Name)
 			poduids = append(poduids, uids...)
 			key, err := cache.MetaNamespaceKeyFunc(service)
 			if err == nil {
@@ -371,7 +371,7 @@ func (agent *HostAgent) handleSnatUpdate(policy *snatpolicy.SnatPolicy) {
 }
 
 func (agent *HostAgent) updateSnatPolicyLabels(obj interface{}, policyname string) (poduids []string) {
-	uids, res := agent.getPodsMatchingObjet(obj, policyname)
+	uids, res := agent.getPodsMatchingObject(obj, policyname)
 	if len(uids) > 0 {
 		key, _ := cache.MetaNamespaceKeyFunc(obj)
 		if _, ok := agent.ReadSnatPolicyLabel(key); ok {
@@ -575,6 +575,8 @@ func fileExists(filename string) bool {
 }
 
 func (agent *HostAgent) snaGlobalInfoChanged(snatobj interface{}, logger *logrus.Entry) {
+	agent.snatPolicyCacheMutex.RLock()
+	defer agent.snatPolicyCacheMutex.RUnlock()
 	snat := snatobj.(*snatglobal.SnatGlobalInfo)
 	syncSnat := false
 	updateLocalInfo := false
@@ -800,7 +802,7 @@ func (agent *HostAgent) syncSnat() bool {
 }
 
 // Get the Pods matching the Object selector
-func (agent *HostAgent) getPodsMatchingObjet(obj interface{}, policyname string) (poduids []string, res ResourceType) {
+func (agent *HostAgent) getPodsMatchingObject(obj interface{}, policyname string) (poduids []string, res ResourceType) {
 	metadata, err := meta.Accessor(obj)
 	if err != nil {
 		return
@@ -956,6 +958,8 @@ func (agent *HostAgent) getMatchingSnatPolicy(obj interface{}) (snatPolicyNames
 	namespace := metadata.GetNamespace()
 	label := metadata.GetLabels()
 	res := getResourceType(obj)
+	agent.snatPolicyCacheMutex.RLock()
+	defer agent.snatPolicyCacheMutex.RUnlock()
 	for _, item := range agent.snatPolicyCache {
 		// check for empty policy selctor
 		if reflect.DeepEqual(item.Spec.Selector, snatpolicy.PodSelector{}) {
@@ -1043,6 +1047,8 @@ func (agent *HostAgent) getMatchingSnatPolicy(obj interface{}) (snatPolicyNames
 }
 
 func (agent *HostAgent) handleObjectUpdateForSnat(obj interface{}) {
+	agent.snatPolicyCacheMutex.RLock()
+	defer agent.snatPolicyCacheMutex.RUnlock()
 	objKey, err := cache.MetaNamespaceKeyFunc(obj)
 	if err != nil {
 		return
@@ -1056,7 +1062,7 @@ func (agent *HostAgent) handleObjectUpdateForSnat(obj interface{}) {
 		polcies := agent.getMatchingSnatPolicy(obj)
 		for name, resources := range polcies {
 			for _, res := range resources {
-				poduids, _ := agent.getPodsMatchingObjet(obj, name)
+				poduids, _ := agent.getPodsMatchingObject(obj, name)
 				if len(agent.snatPolicyCache[name].Spec.Selector.Labels) == 0 {
 					agent.applyPolicy(poduids, res, name)
 				} else {
@@ -1076,7 +1082,7 @@ func (agent *HostAgent) handleObjectUpdateForSnat(obj interface{}) {
 		seen := make(map[string]bool)
 		for name, res := range plcynames {
 			if _, ok := matchnames[name]; !ok {
-				poduids, _ := agent.getPodsMatchingObjet(obj, name)
+				poduids, _ := agent.getPodsMatchingObject(obj, name)
 				for _, uid := range poduids {
 					agent.deleteSnatLocalInfo(uid, res, name)
 				}
@@ -1094,7 +1100,7 @@ func (agent *HostAgent) handleObjectUpdateForSnat(obj interface{}) {
 				continue
 			}
 			for _, res := range resources {
-				poduids, _ := agent.getPodsMatchingObjet(obj, name)
+				poduids, _ := agent.getPodsMatchingObject(obj, name)
 				agent.applyPolicy(poduids, res, name)
 				agent.WriteSnatPolicyLabel(objKey, name, res)
 				sync = true
@@ -1116,7 +1122,7 @@ func (agent *HostAgent) handleObjectDeleteForSnat(obj interface{}) {
 	sync := false
 	for name, resources := range plcynames {
 		agent.log.Infof("Handle snatpolicy as object deleted: %s,  ObjectKey: %s", name, objKey)
-		poduids, _ := agent.getPodsMatchingObjet(obj, name)
+		poduids, _ := agent.getPodsMatchingObject(obj, name)
 		for _, uid := range poduids {
 			if getResourceType(obj) == SERVICE {
 				agent.log.Debug("Service deleted update the localInfo: ", name)