Merge pull request #109 from chenchun/uid

Fix a race condition that resync released ip due to pod uid missmatch

Author: ChenLingPeng

pkg/ipam/schedulerplugin/floatingip_plugin.go

@@ -19,6 +19,7 @@ package schedulerplugin
 import (
 	"fmt"
 	"net"
+	"runtime"
 	"sync"
 	"time"
 
@@ -223,7 +224,19 @@ func (p *FloatingIPPlugin) GetIpam() floatingip.IPAM {
 
 func (p *FloatingIPPlugin) lockPod(name, namespace string) func() {
 	key := fmt.Sprintf("%s_%s", namespace, name)
+	start := time.Now()
 	p.podLockPool.LockKey(key)
+	elapsed := (time.Now().UnixNano() - start.UnixNano()) / 1e6
+	if elapsed > 500 {
+		var caller string
+		pc, _, no, ok := runtime.Caller(1)
+		details := runtime.FuncForPC(pc)
+		if ok && details != nil {
+			caller = fmt.Sprintf("called from %s:%d\n", details.Name(), no)
+		}
+		glog.Infof("acquire lock for %s took %d ms, started at %s, %s", key, elapsed,
+			start.Format("15:04:05.000"), caller)
+	}
 	return func() {
 		_ = p.podLockPool.UnlockKey(key)
 	}

pkg/ipam/schedulerplugin/resync.go

@@ -77,6 +77,11 @@ func (p *FloatingIPPlugin) fetchChecklist(meta *resyncMeta) error {
 			glog.Warningf("unexpected key: %s", fip.Key)
 			continue
 		}
+		if fip.PodUid == "" && fip.NodeName == "" && !keyObj.Deployment() &&
+			constant.ReleasePolicy(fip.Policy) == constant.ReleasePolicyNever {
+			// skip endless checking pod exists for never policy
+			continue
+		}
 		meta.allocatedIPs = append(meta.allocatedIPs, resyncObj{keyObj: keyObj, fip: fip.FloatingIP})
 	}
 	return nil
@@ -88,9 +93,22 @@ func (p *FloatingIPPlugin) resyncAllocatedIPs(meta *resyncMeta) {
 		key := obj.keyObj.KeyInDB
 		func() {
 			defer p.lockPod(obj.keyObj.PodName, obj.keyObj.Namespace)()
-			if p.podRunning(obj.keyObj.PodName, obj.keyObj.Namespace, obj.fip.PodUid) {
+			// we are holding the pod's lock, query again in case the ip has been reallocated.
+			fip, err := p.ipam.ByIP(obj.fip.IP)
+			if err != nil {
+				glog.Warning(err)
 				return
 			}
+			if fip.Key != obj.fip.Key {
+				// if key changed, abort
+				return
+			}
+			obj.fip = fip
+			running, reason := p.podRunning(obj.keyObj.PodName, obj.keyObj.Namespace, obj.fip.PodUid)
+			if running {
+				return
+			}
+			glog.Infof("%s is not running, %s", obj.keyObj.KeyInDB, reason)
 			if p.cloudProvider != nil && obj.fip.NodeName != "" {
 				// For tapp and sts pod, nodeName will be updated to empty after unassigning
 				glog.Infof("UnAssignIP nodeName %s, ip %s, key %s during resync", obj.fip.NodeName,
@@ -122,28 +140,37 @@ func (p *FloatingIPPlugin) resyncAllocatedIPs(meta *resyncMeta) {
 	}
 }
 
-func (p *FloatingIPPlugin) podRunning(podName, namespace, podUid string) bool {
+func (p *FloatingIPPlugin) podRunning(podName, namespace, podUid string) (bool, string) {
 	pod, err := p.PodLister.Pods(namespace).Get(podName)
-	if runningAndUidMatch(podUid, pod, err) {
-		return true
+	running, reason1 := runningAndUidMatch(podUid, pod, err)
+	if running {
+		return true, ""
 	}
 	// double check with apiserver to confirm it is not running
 	pod, err = p.Client.CoreV1().Pods(namespace).Get(podName, v1.GetOptions{})
-	return runningAndUidMatch(podUid, pod, err)
+	running, reason2 := runningAndUidMatch(podUid, pod, err)
+	if running {
+		return true, ""
+	}
+	return false, "from podlist: " + reason1 + ", from client: " + reason2
 }
 
-func runningAndUidMatch(storedUid string, pod *corev1.Pod, err error) bool {
+func runningAndUidMatch(storedUid string, pod *corev1.Pod, err error) (bool, string) {
 	if err != nil {
 		if metaErrs.IsNotFound(err) {
-			return false
+			return false, "pod not found"
 		}
 		// we cannot figure out whether pod exist or not, we'd better keep the ip
-		return true
+		return true, ""
 	}
 	if storedUid != "" && storedUid != string(pod.GetUID()) {
-		return false
+		return false, fmt.Sprintf("pod current uid %s missmatch stored uid %s", string(pod.GetUID()), storedUid)
+	}
+	if !finished(pod) {
+		return true, ""
+	} else {
+		return false, "pod finished"
 	}
-	return !finished(pod)
 }
 
 func parsePodIndex(name string) (int, error) {