Azure CNS fix reconcile bug

cns/NetworkContainerContract.go

@@ -219,6 +219,11 @@ type IPConfigRequest struct {
 	OrchestratorContext json.RawMessage
 }
 
+func (i IPConfigRequest) String() string {
+	return fmt.Sprintf("[IPConfigRequest: DesiredIPAddress %s, OrchestratorContext %s]",
+		i.DesiredIPAddress, string(i.OrchestratorContext))
+}
+
 // IPConfigResponse is used in CNS IPAM mode as a response to CNI ADD
 type IPConfigResponse struct {
 	PodIpInfo PodIpInfo

cns/cnsclient/cnsclient_test.go

@@ -137,9 +137,15 @@ func TestMain(m *testing.M) {
 	}
 
 	if httpRestService != nil {
+		err = httpRestService.Init(&config)
+		if err != nil {
+			logger.Errorf("Failed to initialize HttpService, err:%v.\n", err)
+			return
+		}
+
 		err = httpRestService.Start(&config)
 		if err != nil {
-			logger.Errorf("Failed to start CNS, err:%v.\n", err)
+			logger.Errorf("Failed to start HttpService, err:%v.\n", err)
 			return
 		}
 	}

cns/common/service.go

@@ -24,6 +24,7 @@ type Service struct {
 
 // ServiceAPI defines base interface.
 type ServiceAPI interface {
+	Init(*ServiceConfig) error
 	Start(*ServiceConfig) error
 	Stop()
 	GetOption(string) interface{}

cns/fakes/cnsfake.go

@@ -306,4 +306,9 @@ func (fake *HTTPServiceFake) Start(*common.ServiceConfig) error {
 	return nil
 }
 
+
+func (fake *HTTPServiceFake) Init(*common.ServiceConfig) error {
+	return nil
+}
+
 func (fake *HTTPServiceFake) Stop() {}

cns/fakes/requestcontrollerfake.go

@@ -62,10 +62,18 @@ func (rc *RequestControllerFake) CarveIPConfigsAndAddToStatusAndCNS(numberOfIPCo
 	return cnsIPConfigs
 }
 
+func (rc *RequestControllerFake) InitRequestController() error {
+	return nil
+}
+
 func (rc *RequestControllerFake) StartRequestController(exitChan <-chan struct{}) error {
 	return nil
 }
 
+func (rc *RequestControllerFake) IsStarted() bool {
+	return true
+}
+
 func (rc *RequestControllerFake) UpdateCRDSpec(cntxt context.Context, desiredSpec nnc.NodeNetworkConfigSpec) error {
 	rc.cachedCRD.Spec = desiredSpec
 

cns/ipampoolmonitor/ipampoolmonitor.go

@@ -16,21 +16,23 @@ type CNSIPAMPoolMonitor struct {
 	pendingRelease bool
 
 	cachedNNC   nnc.NodeNetworkConfig
+	updatingIpsNotInUseCount int
 	scalarUnits nnc.Scaler
 
-	cns            cns.HTTPService
+	httpService    cns.HTTPService
 	rc             requestcontroller.RequestController
 	MinimumFreeIps int64
 	MaximumFreeIps int64
 
 	mu sync.RWMutex
 }
 
-func NewCNSIPAMPoolMonitor(cns cns.HTTPService, rc requestcontroller.RequestController) *CNSIPAMPoolMonitor {
+func NewCNSIPAMPoolMonitor(httpService cns.HTTPService, rc requestcontroller.RequestController) *CNSIPAMPoolMonitor {
+	logger.Printf("NewCNSIPAMPoolMonitor: Create IPAM Pool Monitor")
 	return &CNSIPAMPoolMonitor{
-		pendingRelease: false,
-		cns:            cns,
-		rc:             rc,
+		pendingRelease:				false,
+		httpService:        		httpService,
+		rc:             			rc,
 	}
 }
 
@@ -52,7 +54,7 @@ func (pm *CNSIPAMPoolMonitor) Start(ctx context.Context, poolMonitorRefreshMilli
 	for {
 		select {
 		case <-ctx.Done():
-			return fmt.Errorf("CNS IPAM Pool Monitor received cancellation signal")
+			return fmt.Errorf("[ipam-pool-monitor] CNS IPAM Pool Monitor received cancellation signal")
 		case <-ticker.C:
 			err := pm.Reconcile()
 			if err != nil {
@@ -63,14 +65,14 @@ func (pm *CNSIPAMPoolMonitor) Start(ctx context.Context, poolMonitorRefreshMilli
 }
 
 func (pm *CNSIPAMPoolMonitor) Reconcile() error {
-	cnsPodIPConfigCount := len(pm.cns.GetPodIPConfigState())
-	pendingProgramCount := len(pm.cns.GetPendingProgramIPConfigs()) // TODO: add pending program count to real cns
-	allocatedPodIPCount := len(pm.cns.GetAllocatedIPConfigs())
-	pendingReleaseIPCount := len(pm.cns.GetPendingReleaseIPConfigs())
-	availableIPConfigCount := len(pm.cns.GetAvailableIPConfigs()) // TODO: add pending allocation count to real cns
+	cnsPodIPConfigCount := len(pm.httpService.GetPodIPConfigState())
+	pendingProgramCount := len(pm.httpService.GetPendingProgramIPConfigs()) // TODO: add pending program count to real cns
+	allocatedPodIPCount := len(pm.httpService.GetAllocatedIPConfigs())
+	pendingReleaseIPCount := len(pm.httpService.GetPendingReleaseIPConfigs())
+	availableIPConfigCount := len(pm.httpService.GetAvailableIPConfigs()) // TODO: add pending allocation count to real cns
 	freeIPConfigCount := pm.cachedNNC.Spec.RequestedIPCount - int64(allocatedPodIPCount)
 
-	msg := fmt.Sprintf("Pool Size: %v, Goal Size: %v, BatchSize: %v, MinFree: %v, MaxFree:%v, Allocated: %v, Available: %v, Pending Release: %v, Free: %v, Pending Program: %v",
+	msg := fmt.Sprintf("[ipam-pool-monitor] Pool Size: %v, Goal Size: %v, BatchSize: %v, MinFree: %v, MaxFree:%v, Allocated: %v, Available: %v, Pending Release: %v, Free: %v, Pending Program: %v",
 		cnsPodIPConfigCount, pm.cachedNNC.Spec.RequestedIPCount, pm.scalarUnits.BatchSize, pm.MinimumFreeIps, pm.MaximumFreeIps, allocatedPodIPCount, availableIPConfigCount, pendingReleaseIPCount, freeIPConfigCount, pendingProgramCount)
 
 	switch {
@@ -82,7 +84,7 @@ func (pm *CNSIPAMPoolMonitor) Reconcile() error {
 	// pod count is decreasing
 	case freeIPConfigCount > pm.MaximumFreeIps:
 		logger.Printf("[ipam-pool-monitor] Decreasing pool size...%s ", msg)
-		return pm.decreasePoolSize()
+		return pm.decreasePoolSize(pendingReleaseIPCount)
 
 	// CRD has reconciled CNS state, and target spec is now the same size as the state
 	// free to remove the IP's from the CRD
@@ -100,8 +102,8 @@ func (pm *CNSIPAMPoolMonitor) Reconcile() error {
 }
 
 func (pm *CNSIPAMPoolMonitor) increasePoolSize() error {
-	pm.mu.Lock()
 	defer pm.mu.Unlock()
+	pm.mu.Lock()
 
 	var err error
 	var tempNNCSpec nnc.NodeNetworkConfigSpec
@@ -111,58 +113,79 @@ func (pm *CNSIPAMPoolMonitor) increasePoolSize() error {
 	}
 
 	tempNNCSpec.RequestedIPCount += pm.scalarUnits.BatchSize
-	logger.Printf("[ipam-pool-monitor] Increasing pool size, Current Pool Size: %v, Updated Requested IP Count: %v, Pods with IP's:%v, ToBeDeleted Count: %v", len(pm.cns.GetPodIPConfigState()), tempNNCSpec.RequestedIPCount, len(pm.cns.GetAllocatedIPConfigs()), len(tempNNCSpec.IPsNotInUse))
+	logger.Printf("[ipam-pool-monitor] Increasing pool size, Current Pool Size: %v, Updated Requested IP Count: %v, Pods with IP's:%v, ToBeDeleted Count: %v", len(pm.httpService.GetPodIPConfigState()), tempNNCSpec.RequestedIPCount, len(pm.httpService.GetAllocatedIPConfigs()), len(tempNNCSpec.IPsNotInUse))
 
 	err = pm.rc.UpdateCRDSpec(context.Background(), tempNNCSpec)
 	if err != nil {
 		// caller will retry to update the CRD again
 		return err
 	}
 
+	logger.Printf("[ipam-pool-monitor] Increasing pool size: UpdateCRDSpec succeeded for spec %+v", tempNNCSpec)
 	// save the updated state to cachedSpec
 	pm.cachedNNC.Spec = tempNNCSpec
 	return nil
 }
 
-func (pm *CNSIPAMPoolMonitor) decreasePoolSize() error {
-	pm.mu.Lock()
+func (pm *CNSIPAMPoolMonitor) decreasePoolSize(existingPendingReleaseIPCount int) error {
 	defer pm.mu.Unlock()
+	pm.mu.Lock()
 
 	// mark n number of IP's as pending
-	pendingIpAddresses, err := pm.cns.MarkIPAsPendingRelease(int(pm.scalarUnits.BatchSize))
-	if err != nil {
-		return err
-	}
+	var err error
+	var newIpsMarkedAsPending bool
+	var pendingIpAddresses map[string]cns.IPConfigurationStatus
+	if pm.updatingIpsNotInUseCount == 0 ||
+		pm.updatingIpsNotInUseCount < existingPendingReleaseIPCount {
+		logger.Printf("[ipam-pool-monitor] Marking IPs as PendingRelease, ipsToBeReleasedCount %d", int(pm.scalarUnits.BatchSize))
+		pendingIpAddresses, err = pm.httpService.MarkIPAsPendingRelease(int(pm.scalarUnits.BatchSize))
+		if err != nil {
+			return err
+		}
 
-	totalIpsSetForRelease := len(pendingIpAddresses)
-	logger.Printf("[ipam-pool-monitor] Releasing IPCount in this batch %d", totalIpsSetForRelease)
+		newIpsMarkedAsPending = true
+	}
 
 	var tempNNCSpec nnc.NodeNetworkConfigSpec
 	tempNNCSpec, err = pm.createNNCSpecForCRD(false)
 	if err != nil {
 		return err
 	}
 
-	tempNNCSpec.RequestedIPCount -= int64(totalIpsSetForRelease)
-	logger.Printf("[ipam-pool-monitor] Decreasing pool size, Current Pool Size: %v, Requested IP Count: %v, Pods with IP's: %v, ToBeDeleted Count: %v", len(pm.cns.GetPodIPConfigState()), tempNNCSpec.RequestedIPCount, len(pm.cns.GetAllocatedIPConfigs()), len(tempNNCSpec.IPsNotInUse))
+	if newIpsMarkedAsPending {
+		// cache the updatingPendingRelease so that we dont re-set new IPs to PendingRelease in case UpdateCRD call fails
+		pm.updatingIpsNotInUseCount = len(tempNNCSpec.IPsNotInUse)
+	}
+
+	logger.Printf("[ipam-pool-monitor] Releasing IPCount in this batch %d, updatingPendingIpsNotInUse count %d", len(pendingIpAddresses), pm.updatingIpsNotInUseCount)
+
+	tempNNCSpec.RequestedIPCount -= int64(len(pendingIpAddresses))
+	logger.Printf("[ipam-pool-monitor] Decreasing pool size, Current Pool Size: %v, Requested IP Count: %v, Pods with IP's: %v, ToBeDeleted Count: %v", len(pm.httpService.GetPodIPConfigState()), tempNNCSpec.RequestedIPCount, len(pm.httpService.GetAllocatedIPConfigs()), len(tempNNCSpec.IPsNotInUse))
 
 	err = pm.rc.UpdateCRDSpec(context.Background(), tempNNCSpec)
 	if err != nil {
 		// caller will retry to update the CRD again
 		return err
 	}
 
+	logger.Printf("[ipam-pool-monitor] Decreasing pool size: UpdateCRDSpec succeeded for spec %+v", tempNNCSpec)
+
 	// save the updated state to cachedSpec
 	pm.cachedNNC.Spec = tempNNCSpec
 	pm.pendingRelease = true
+
+	// clear the updatingPendingIpsNotInUse, as we have Updated the CRD
+	logger.Printf("[ipam-pool-monitor] cleaning the updatingPendingIpsNotInUse, existing length %d", pm.updatingIpsNotInUseCount)
+	pm.updatingIpsNotInUseCount = 0
+
 	return nil
 }
 
 // if cns pending ip release map is empty, request controller has already reconciled the CNS state,
 // so we can remove it from our cache and remove the IP's from the CRD
 func (pm *CNSIPAMPoolMonitor) cleanPendingRelease() error {
-	pm.mu.Lock()
 	defer pm.mu.Unlock()
+	pm.mu.Lock()
 
 	var err error
 	var tempNNCSpec nnc.NodeNetworkConfigSpec
@@ -177,6 +200,9 @@ func (pm *CNSIPAMPoolMonitor) cleanPendingRelease() error {
 		return err
 	}
 
+	logger.Printf("[ipam-pool-monitor] cleanPendingRelease: UpdateCRDSpec succeeded for spec %+v", tempNNCSpec)
+
+
 	// save the updated state to cachedSpec
 	pm.cachedNNC.Spec = tempNNCSpec
 	pm.pendingRelease = false
@@ -197,7 +223,7 @@ func (pm *CNSIPAMPoolMonitor) createNNCSpecForCRD(resetNotInUseList bool) (nnc.N
 		spec.IPsNotInUse = make([]string, 0)
 	} else {
 		// Get All Pending IPs from CNS and populate it again.
-		pendingIps := pm.cns.GetPendingReleaseIPConfigs()
+		pendingIps := pm.httpService.GetPendingReleaseIPConfigs()
 		for _, pendingIp := range pendingIps {
 			spec.IPsNotInUse = append(spec.IPsNotInUse, pendingIp.ID)
 		}
@@ -208,14 +234,18 @@ func (pm *CNSIPAMPoolMonitor) createNNCSpecForCRD(resetNotInUseList bool) (nnc.N
 
 // UpdatePoolLimitsTransacted called by request controller on reconcile to set the batch size limits
 func (pm *CNSIPAMPoolMonitor) Update(scalar nnc.Scaler, spec nnc.NodeNetworkConfigSpec) error {
-	pm.mu.Lock()
 	defer pm.mu.Unlock()
+	pm.mu.Lock()
+
 	pm.scalarUnits = scalar
 
 	pm.MinimumFreeIps = int64(float64(pm.scalarUnits.BatchSize) * (float64(pm.scalarUnits.RequestThresholdPercent) / 100))
 	pm.MaximumFreeIps = int64(float64(pm.scalarUnits.BatchSize) * (float64(pm.scalarUnits.ReleaseThresholdPercent) / 100))
 
 	pm.cachedNNC.Spec = spec
 
+	logger.Printf("[ipam-pool-monitor] Update spec %+v, pm.MinimumFreeIps %d, pm.MaximumFreeIps %d",
+		pm.cachedNNC.Spec, pm.MinimumFreeIps, pm.MaximumFreeIps)
+
 	return nil
 }

cns/requestcontroller/kubecontroller/crdreconciler.go

@@ -41,13 +41,25 @@ func (r *CrdReconciler) Reconcile(request reconcile.Request) (reconcile.Result,
 	}
 
 	logger.Printf("[cns-rc] CRD Spec: %v", nodeNetConfig.Spec)
-	logger.Printf("[cns-rc] CRD Status: %v", nodeNetConfig.Status)
+
 
 	// If there are no network containers, don't hand it off to CNS
 	if len(nodeNetConfig.Status.NetworkContainers) == 0 {
+		logger.Errorf("[cns-rc] Empty NetworkContainers")
 		return reconcile.Result{}, nil
 	}
 
+	networkContainer := nodeNetConfig.Status.NetworkContainers[0]
+	logger.Printf("[cns-rc] CRD Status: NcId: [%s], Version: [%d],  podSubnet: [%s], Subnet CIDR: [%s], " +
+		"Gateway Addr: [%s], Primary IP: [%s], SecondaryIpsCount: [%d]",
+		networkContainer.ID,
+		networkContainer.Version,
+		networkContainer.SubnetName,
+		networkContainer.SubnetAddressSpace,
+		networkContainer.DefaultGateway,
+		networkContainer.PrimaryIP,
+		len(networkContainer.IPAssignments))
+
 	// Otherwise, create NC request and hand it off to CNS
 	ncRequest, err = CRDStatusToNCRequest(nodeNetConfig.Status)
 	if err != nil {

cns/requestcontroller/kubecontroller/crdrequestcontroller.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"os"
+	"sync"
 
 	"github.com/Azure/azure-container-networking/cns"
 	"github.com/Azure/azure-container-networking/cns/cnsclient"
@@ -42,6 +43,9 @@ type crdRequestController struct {
 	CNSClient       cnsclient.APIClient
 	nodeName        string //name of node running this program
 	Reconciler      *CrdReconciler
+	initialized     bool
+	Started         bool
+	lock            sync.Mutex
 }
 
 // GetKubeConfig precedence
@@ -139,20 +143,38 @@ func NewCrdRequestController(restService *restserver.HTTPRestService, kubeconfig
 	return &crdRequestController, nil
 }
 
+// InitRequestController will initialize/reconcile the CNS state
+func (crdRC *crdRequestController) InitRequestController() error {
+	logger.Printf("InitRequestController")
+
+	defer crdRC.lock.Unlock()
+	crdRC.lock.Lock()
+
+	if err := crdRC.initCNS(); err != nil {
+		logger.Errorf("[cns-rc] Error initializing cns state: %v", err)
+		return err
+	}
+
+	crdRC.initialized = true
+	return nil
+}
+
 // StartRequestController starts the Reconciler loop which watches for CRD status updates
 // Blocks until SIGINT or SIGTERM is received
 // Notifies exitChan when kill signal received
 func (crdRC *crdRequestController) StartRequestController(exitChan <-chan struct{}) error {
-	var (
-		err error
-	)
+	logger.Printf("StartRequestController")
 
-	logger.Printf("Initializing CNS state")
-	if err = crdRC.initCNS(); err != nil {
-		logger.Errorf("[cns-rc] Error initializing cns state: %v", err)
-		return err
+	crdRC.lock.Lock()
+	if crdRC.initialized != true {
+		crdRC.lock.Unlock()
+		return fmt.Errorf("Failed to start requestController, state is not initialized [%v]", crdRC)
 	}
 
+	// Setting the started state
+	crdRC.Started = true
+	crdRC.lock.Unlock()
+
 	logger.Printf("Starting reconcile loop")
 	if err := crdRC.mgr.Start(exitChan); err != nil {
 		if crdRC.isNotDefined(err) {
@@ -166,6 +188,13 @@ func (crdRC *crdRequestController) StartRequestController(exitChan <-chan struct
 	return nil
 }
 
+// return if RequestController is started
+func (crdRC *crdRequestController) IsStarted() bool {
+	defer crdRC.lock.Unlock()
+	crdRC.lock.Lock()
+	return crdRC.Started
+}
+
 // InitCNS initializes cns by passing pods and a createnetworkcontainerrequest
 func (crdRC *crdRequestController) initCNS() error {
 	var (
@@ -242,7 +271,7 @@ func (crdRC *crdRequestController) initCNS() error {
 }
 
 // UpdateCRDSpec updates the CRD spec
-func (crdRC *crdRequestController) UpdateCRDSpec(cntxt context.Context, crdSpec nnc.NodeNetworkConfigSpec) error {
+func (crdRC *crdRequestController) UpdateCRDSpec(cntxt context.Context, crdSpec nnc.NodeNetworkConfigSpec)  error {
 	nodeNetworkConfig, err := crdRC.getNodeNetConfig(cntxt, crdRC.nodeName, k8sNamespace)
 	if err != nil {
 		logger.Errorf("[cns-rc] Error getting CRD when updating spec %v", err)

cns/requestcontroller/requestcontrollerintreface.go

@@ -8,6 +8,8 @@ import (
 
 // RequestController interface for cns to interact with the request controller
 type RequestController interface {
+	InitRequestController() error
 	StartRequestController(exitChan <-chan struct{}) error
 	UpdateCRDSpec(cntxt context.Context, crdSpec nnc.NodeNetworkConfigSpec) error
+	IsStarted() bool
 }