Merge pull request #230 from liornoy/cherry-pick-fix-assignment-4.14

OCPBUGS-26553: Cherry pick fix assignment 4.14

doc/crds/daemonset-install.yaml

@@ -69,6 +69,7 @@ rules:
   - create
   - patch
   - update
+  - get
 ---
 apiVersion: apps/v1
 kind: DaemonSet

pkg/allocate/allocate.go

@@ -11,13 +11,15 @@ import (
 
 // AssignmentError defines an IP assignment error.
 type AssignmentError struct {
-	firstIP net.IP
-	lastIP  net.IP
-	ipnet   net.IPNet
+	firstIP       net.IP
+	lastIP        net.IP
+	ipnet         net.IPNet
+	excludeRanges []string
 }
 
 func (a AssignmentError) Error() string {
-	return fmt.Sprintf("Could not allocate IP in range: ip: %v / - %v / range: %#v", a.firstIP, a.lastIP, a.ipnet)
+	return fmt.Sprintf("Could not allocate IP in range: ip: %v / - %v / range: %s / excludeRanges: %v",
+		a.firstIP, a.lastIP, a.ipnet.String(), a.excludeRanges)
 }
 
 // AssignIP assigns an IP using a range and a reserve list.
@@ -102,10 +104,14 @@ func IterateForAssignment(ipnet net.IPNet, rangeStart net.IP, rangeEnd net.IP, r
 	for _, r := range reserveList {
 		reserved[r.IP.String()] = true
 	}
+
 	// Build excluded list, "192.168.2.229/30", "192.168.1.229/30".
 	excluded := []*net.IPNet{}
 	for _, v := range excludeRanges {
-		_, subnet, _ := net.ParseCIDR(v)
+		subnet, err := parseExcludedRange(v)
+		if err != nil {
+			return net.IP{}, reserveList, fmt.Errorf("could not parse exclude range, err: %q", err)
+		}
 		excluded = append(excluded, subnet)
 	}
 
@@ -129,7 +135,7 @@ func IterateForAssignment(ipnet net.IPNet, rangeStart net.IP, rangeEnd net.IP, r
 	}
 
 	// No IP address for assignment found, return an error.
-	return net.IP{}, reserveList, AssignmentError{firstIP, lastIP, ipnet}
+	return net.IP{}, reserveList, AssignmentError{firstIP, lastIP, ipnet, excludeRanges}
 }
 
 // skipExcludedSubnets iterates through all subnets and checks if ip is part of them. If i is part of one of the subnets,
@@ -144,3 +150,28 @@ func skipExcludedSubnets(ip net.IP, excluded []*net.IPNet) net.IP {
 	}
 	return nil
 }
+
+// parseExcludedRange parses a provided string to a net.IPNet.
+// If the provided string is a valid CIDR, return the net.IPNet for that CIDR.
+// If the provided string is a valid IP address, add the /32 or /128 prefix to form the CIDR and return the net.IPNet.
+// Otherwise, return the error.
+func parseExcludedRange(s string) (*net.IPNet, error) {
+	// Try parsing CIDRs.
+	_, subnet, err := net.ParseCIDR(s)
+	if err == nil {
+		return subnet, nil
+	}
+	// The user might have given a single IP address, try parsing that - if it does not parse, return the error that
+	// we got earlier.
+	ip := net.ParseIP(s)
+	if ip == nil {
+		return nil, err
+	}
+	// If the address parses, check if it's IPv4 or IPv6 and add the correct prefix.
+	if ip.To4() != nil {
+		_, subnet, err = net.ParseCIDR(fmt.Sprintf("%s/32", s))
+	} else {
+		_, subnet, err = net.ParseCIDR(fmt.Sprintf("%s/128", s))
+	}
+	return subnet, err
+}

pkg/allocate/allocate_test.go

@@ -113,6 +113,33 @@ var _ = Describe("Allocation operations", func() {
 
 	})
 
+	It("can IterateForAssignment on an IPv4 address excluding a range which is a single IP", func() {
+		firstip, ipnet, err := net.ParseCIDR("192.168.0.0/29")
+		Expect(err).NotTo(HaveOccurred())
+
+		// figure out the range start.
+		calculatedrangestart := net.ParseIP(firstip.Mask(ipnet.Mask).String())
+
+		var ipres []types.IPReservation
+		exrange := []string{"192.168.0.1"}
+		newip, _, err := IterateForAssignment(*ipnet, calculatedrangestart, nil, ipres, exrange, "0xdeadbeef", "")
+		Expect(err).NotTo(HaveOccurred())
+		Expect(fmt.Sprint(newip)).To(Equal("192.168.0.2"))
+	})
+
+	It("correctly handles invalid syntax for an exclude range with IPv4", func() {
+		firstip, ipnet, err := net.ParseCIDR("192.168.0.0/29")
+		Expect(err).NotTo(HaveOccurred())
+
+		// figure out the range start.
+		calculatedrangestart := net.ParseIP(firstip.Mask(ipnet.Mask).String())
+
+		var ipres []types.IPReservation
+		exrange := []string{"192.168.0.1/123"}
+		_, _, err = IterateForAssignment(*ipnet, calculatedrangestart, nil, ipres, exrange, "0xdeadbeef", "")
+		Expect(err).To(MatchError(HavePrefix("could not parse exclude range")))
+	})
+
 	It("can IterateForAssignment on an IPv6 address excluding a range", func() {
 
 		firstip, ipnet, err := net.ParseCIDR("100::2:1/125")
@@ -128,6 +155,32 @@ var _ = Describe("Allocation operations", func() {
 
 	})
 
+	It("can IterateForAssignment on an IPv6 address excluding a range which is a single IP", func() {
+		firstip, ipnet, err := net.ParseCIDR("100::2:1/125")
+		Expect(err).NotTo(HaveOccurred())
+
+		// figure out the range start.
+		calculatedrangestart := net.ParseIP(firstip.Mask(ipnet.Mask).String())
+
+		var ipres []types.IPReservation
+		exrange := []string{"100::2:1"}
+		newip, _, _ := IterateForAssignment(*ipnet, calculatedrangestart, nil, ipres, exrange, "0xdeadbeef", "")
+		Expect(fmt.Sprint(newip)).To(Equal("100::2:2"))
+	})
+
+	It("correctly handles invalid syntax for an exclude range with IPv6", func() {
+		firstip, ipnet, err := net.ParseCIDR("100::2:1/125")
+		Expect(err).NotTo(HaveOccurred())
+
+		// figure out the range start.
+		calculatedrangestart := net.ParseIP(firstip.Mask(ipnet.Mask).String())
+
+		var ipres []types.IPReservation
+		exrange := []string{"100::2::1"}
+		_, _, err = IterateForAssignment(*ipnet, calculatedrangestart, nil, ipres, exrange, "0xdeadbeef", "")
+		Expect(err).To(MatchError(HavePrefix("could not parse exclude range")))
+	})
+
 	It("can IterateForAssignment on an IPv6 address excluding a very large range", func() {
 
 		firstip, ipnet, err := net.ParseCIDR("2001:db8::/30")

pkg/reconciler/ip_test.go

@@ -264,8 +264,8 @@ var _ = Describe("Whereabouts IP reconciler", func() {
 			Expect(err).NotTo(HaveOccurred())
 		})
 
-		It("cannot be reconciled", func() {
-			Expect(reconcileLooper.ReconcileIPPools(context.TODO())).To(BeEmpty())
+		It("can be reconciled", func() {
+			Expect(reconcileLooper.ReconcileIPPools(context.TODO())).NotTo(BeEmpty())
 		})
 	})
 })

pkg/reconciler/iploop.go

@@ -103,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

@@ -89,3 +89,14 @@ func networkStatusFromPod(pod v1.Pod) string {
 	}
 	return networkStatusAnnotationValue
 }
+
+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

@@ -107,6 +107,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) {
 	ctxWithTimeout, cancel := context.WithTimeout(ctx, storage.RequestTimeout)
 	defer cancel()

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