[flexible-ipam] Wait for controller ready when processing CNI request

Signed-off-by: gran <gran@vmware.com>

pkg/agent/cniserver/ipam/antrea_ipam.go

@@ -17,6 +17,7 @@ package ipam
 import (
 	"fmt"
 	"net"
+	"sync"
 
 	"github.com/containernetworking/cni/pkg/invoke"
 	cnitypes "github.com/containernetworking/cni/pkg/types"
@@ -37,7 +38,8 @@ const (
 // if present. If annotation is not present, the driver will delegate functionality
 // to traditional IPAM driver.
 type AntreaIPAM struct {
-	controller *AntreaIPAMController
+	controller      *AntreaIPAMController
+	controllerMutex sync.RWMutex
 }
 
 // Global variable is needed to work around order of initialization
@@ -97,6 +99,8 @@ func generateIPConfig(ip net.IP, prefixLength int, gwIP net.IP) (*current.IPConf
 }
 
 func (d *AntreaIPAM) setController(controller *AntreaIPAMController) {
+	d.controllerMutex.Lock()
+	defer d.controllerMutex.Unlock()
 	d.controller = controller
 }
 
@@ -197,9 +201,11 @@ func (d *AntreaIPAM) Check(args *invoke.Args, k8sArgs *argtypes.K8sArgs, network
 // of today). If annotation is not present, or annotated IP Pool not found, the driver
 // will not own the request and fall back to next IPAM driver.
 func (d *AntreaIPAM) owns(k8sArgs *argtypes.K8sArgs, errFunc func(error) error) (bool, *poolallocator.IPPoolAllocator, []net.IP, *crdv1a2.IPAddressOwner, error) {
-	if d.controller == nil {
-		klog.Warningf("Antrea IPAM driver failed to initialize due to inconsistent configuration. Falling back to default IPAM")
-		return false, nil, nil, nil, nil
+	// Wait controller ready to avoid inappropriate behavior on CNI request
+	d.controllerMutex.RLock()
+	defer d.controllerMutex.RUnlock()
+	if err := d.controller.waitControllerReadyForCNIRequest(); err != nil {
+		return false, nil, nil, nil, err
 	}
 
 	// As of today, only Namespace annotation is supported

pkg/agent/cniserver/ipam/antrea_ipam_controller.go

@@ -18,8 +18,10 @@ import (
 	"fmt"
 	"net"
 	"strings"
+	"time"
 
 	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/informers"
 	coreinformers "k8s.io/client-go/informers/core/v1"
 	clientset "k8s.io/client-go/kubernetes"
@@ -105,8 +107,6 @@ func InitializeAntreaIPAMController(kubeClient clientset.Interface, crdClient cl
 		return nil, fmt.Errorf("Antrea IPAM driver failed to initialize")
 	}
 
-	antreaIPAMDriver.setController(antreaIPAMController)
-
 	return antreaIPAMController, nil
 }
 
@@ -121,10 +121,21 @@ func (c *AntreaIPAMController) Run(stopCh <-chan struct{}) {
 	if !cache.WaitForNamedCacheSync(controllerName, stopCh, c.namespaceInformer.Informer().HasSynced, c.ipPoolInformer.Informer().HasSynced, c.podInformer.HasSynced) {
 		return
 	}
+	antreaIPAMDriver.setController(c)
 
 	<-stopCh
 }
 
+func (c *AntreaIPAMController) waitControllerReadyForCNIRequest() error {
+	return wait.PollImmediate(500*time.Millisecond, 5*time.Second, func() (bool, error) {
+		if c == nil {
+			klog.Warningf("Antrea IPAM driver is not ready.")
+			return false, nil
+		}
+		return true, nil
+	})
+}
+
 func (c *AntreaIPAMController) getIPPoolsByPod(namespace, name string) ([]string, []net.IP, *crdv1a2.IPAddressOwner, error) {
 	// Find IPPool by Pod
 	var ips []net.IP