Rechecking pending Pods

cmd/reconciler/ip.go

@@ -9,7 +9,7 @@ import (
 	"github.com/k8snetworkplumbingwg/whereabouts/pkg/reconciler"
 )
 
-const defaultReconcilerTimeout = 30
+const defaultReconcilerTimeout = 60
 
 func main() {
 	kubeConfigFile := flag.String("kubeconfig", "", "the path to the Kubernetes configuration file")

cmd/reconciler/ip_test.go

@@ -254,36 +254,6 @@ var _ = Describe("Whereabouts IP reconciler", func() {
 		})
 	})
 
-	Context("a pod in pending state, without an IP in its network-status", func() {
-		const poolName = "pool1"
-
-		var pod *v1.Pod
-		var pool *v1alpha1.IPPool
-
-		BeforeEach(func() {
-			var err error
-			pod, err = k8sClientSet.CoreV1().Pods(namespace).Create(
-				context.TODO(),
-				generatePendingPod(namespace, podName),
-				metav1.CreateOptions{})
-			Expect(err).NotTo(HaveOccurred())
-
-			pool = generateIPPoolSpec(ipRange, namespace, poolName, pod.Name)
-			Expect(k8sClient.Create(context.Background(), pool)).NotTo(HaveOccurred())
-
-			reconcileLooper, err = reconciler.NewReconcileLooperWithKubeconfig(context.TODO(), kubeConfigPath, timeout)
-			Expect(err).NotTo(HaveOccurred())
-		})
-
-		AfterEach(func() {
-			Expect(k8sClient.Delete(context.Background(), pool)).NotTo(HaveOccurred())
-			Expect(k8sClientSet.CoreV1().Pods(namespace).Delete(context.TODO(), pod.GetName(), metav1.DeleteOptions{}))
-		})
-
-		It("cannot be reconciled", func() {
-			Expect(reconcileLooper.ReconcileIPPools(context.TODO())).To(BeEmpty())
-		})
-	})
 })
 
 func generateIPPoolSpec(ipRange string, namespace string, poolName string, podNames ...string) *v1alpha1.IPPool {

doc/crds/daemonset-install.yaml

@@ -47,6 +47,7 @@ rules:
   - pods
   verbs:
   - list
+  - get
 ---
 apiVersion: apps/v1
 kind: DaemonSet

pkg/reconciler/iploop.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"net"
+	"strings"
 	"time"
 
 	"github.com/k8snetworkplumbingwg/whereabouts/pkg/allocate"
@@ -102,19 +103,73 @@ func (rl *ReconcileLooper) findOrphanedIPsPerPool(ipPools []storage.IPPool) erro
 func (rl ReconcileLooper) isPodAlive(podRef string, ip string) bool {
 	for livePodRef, livePod := range rl.liveWhereaboutsPods {
 		if podRef == livePodRef {
-			livePodIPs := livePod.ips
-			logging.Debugf(
-				"pod reference %s matches allocation; Allocation IP: %s; PodIPs: %s",
-				livePodRef,
-				ip,
-				livePodIPs)
-			_, isFound := livePodIPs[ip]
-			return isFound || livePod.phase == v1.PodPending
+			isFound := isIpOnPod(&livePod, podRef, ip)
+			if !isFound && (livePod.phase == v1.PodPending) {
+				/* Sometimes pods are still coming up, and may not yet have Multus
+				 * annotation added to it yet. We don't want to check the IPs yet
+				 * so re-fetch the Pod 5x
+				 */
+				podToMatch := &livePod
+				retries := 0
+
+				logging.Debugf("Re-fetching Pending Pod: %s IP-to-match: %s", livePodRef, ip)
+
+				for retries < storage.PodRefreshRetries {
+					retries += 1
+					podToMatch = rl.refreshPod(livePodRef)
+					if podToMatch == nil {
+						logging.Debugf("Cleaning up...")
+						return false
+					} else if podToMatch.phase != v1.PodPending {
+						logging.Debugf("Pending Pod is now in phase: %s", podToMatch.phase)
+						break
+					} else {
+						isFound = isIpOnPod(podToMatch, podRef, ip)
+						// Short-circuit - Pending Pod may have IP now
+						if isFound {
+							logging.Debugf("Pod now has IP annotation while in Pending")
+							return true
+						}
+						time.Sleep(time.Duration(500) * time.Millisecond)
+					}
+				}
+				isFound = isIpOnPod(podToMatch, podRef, ip)
+			}
+
+			return isFound
 		}
 	}
 	return false
 }
 
+func (rl ReconcileLooper) refreshPod(podRef string) *podWrapper {
+	namespace, podName := splitPodRef(podRef)
+	if namespace == "" || podName == "" {
+		logging.Errorf("Invalid podRef format: %s", podRef)
+		return nil
+	}
+
+	pod, err := rl.k8sClient.GetPod(namespace, podName)
+	if err != nil {
+		logging.Errorf("Failed to refresh Pod %s: %s\n", podRef, err)
+		return nil
+	}
+
+	wrappedPod := wrapPod(*pod)
+	logging.Debugf("Got refreshed pod: %v", wrappedPod)
+	return wrappedPod
+}
+
+func splitPodRef(podRef string) (string, string) {
+	namespacedName := strings.Split(podRef, "/")
+	if len(namespacedName) != 2 {
+		logging.Errorf("Failed to split podRef %s", podRef)
+		return "", ""
+	}
+
+	return namespacedName[0], namespacedName[1]
+}
+
 func composePodRef(pod v1.Pod) string {
 	return fmt.Sprintf("%s/%s", pod.GetNamespace(), pod.GetName())
 }

pkg/reconciler/wrappedPod.go

@@ -26,7 +26,7 @@ type void struct{}
 
 func wrapPod(pod v1.Pod) *podWrapper {
 	return &podWrapper{
-		ips: getFlatIPSet(pod),
+		ips:   getFlatIPSet(pod),
 		phase: pod.Status.Phase,
 	}
 }
@@ -80,3 +80,14 @@ func getFlatIPSet(pod v1.Pod) map[string]void {
 	}
 	return ipSet
 }
+
+func isIpOnPod(livePod *podWrapper, podRef, ip string) bool {
+	livePodIPs := livePod.ips
+	logging.Debugf(
+		"pod reference %s matches allocation; Allocation IP: %s; PodIPs: %s",
+		podRef,
+		ip,
+		livePodIPs)
+	_, isFound := livePodIPs[ip]
+	return isFound
+}

pkg/storage/kubernetes/client.go

@@ -117,6 +117,15 @@ func (i *Client) ListPods(ctx context.Context) ([]v1.Pod, error) {
 	return podList.Items, nil
 }
 
+func (i *Client) GetPod(namespace, name string) (*v1.Pod, error) {
+	pod, err := i.clientSet.CoreV1().Pods(namespace).Get(context.TODO(), name, metav1.GetOptions{})
+	if err != nil {
+		return nil, err
+	}
+
+	return pod, nil
+}
+
 func (i *Client) ListOverlappingIPs(ctx context.Context) ([]whereaboutsv1alpha1.OverlappingRangeIPReservation, error) {
 	overlappingIPsList := whereaboutsv1alpha1.OverlappingRangeIPReservationList{}
 	if err := i.client.List(ctx, &overlappingIPsList, &client.ListOptions{}); err != nil {

pkg/storage/storage.go

@@ -13,7 +13,8 @@ var (
 	RequestTimeout = 10 * time.Second
 
 	// DatastoreRetries defines how many retries are attempted when updating the Pool
-	DatastoreRetries = 100
+	DatastoreRetries  = 100
+	PodRefreshRetries = 3
 )
 
 // IPPool is the interface that represents an manageable pool of allocated IPs