Add pending program status for IPs in CNS.

cns/NetworkContainerContract.go

@@ -58,9 +58,10 @@ const (
 
 // IPConfig States for CNS IPAM
 const (
-	Available      = "Available"
-	Allocated      = "Allocated"
-	PendingRelease = "PendingRelease"
+	Available          = "Available"
+	Allocated          = "Allocated"
+	PendingRelease     = "PendingRelease"
+	PendingProgramming = "PendingProgramming"
 )
 
 // ChannelMode :- CNS channel modes

cns/api.go

@@ -37,6 +37,7 @@ type HTTPService interface {
 	SendNCSnapShotPeriodically(int, chan bool)
 	SetNodeOrchestrator(*SetOrchestratorTypeRequest)
 	SyncNodeStatus(string, string, string, json.RawMessage) (int, string)
+	GetPendingProgramIPConfigs() []IPConfigurationStatus
 	GetAvailableIPConfigs() []IPConfigurationStatus
 	GetAllocatedIPConfigs() []IPConfigurationStatus
 	GetPendingReleaseIPConfigs() []IPConfigurationStatus

cns/fakes/cnsfake.go

@@ -58,6 +58,7 @@ func (stack *StringStack) Pop() (string, error) {
 }
 
 type IPStateManager struct {
+	PendingProgramIPConfigState map[string]cns.IPConfigurationStatus
 	AvailableIPConfigState      map[string]cns.IPConfigurationStatus
 	AllocatedIPConfigState      map[string]cns.IPConfigurationStatus
 	PendingReleaseIPConfigState map[string]cns.IPConfigurationStatus
@@ -67,6 +68,7 @@ type IPStateManager struct {
 
 func NewIPStateManager() IPStateManager {
 	return IPStateManager{
+		PendingProgramIPConfigState: make(map[string]cns.IPConfigurationStatus),
 		AvailableIPConfigState:      make(map[string]cns.IPConfigurationStatus),
 		AllocatedIPConfigState:      make(map[string]cns.IPConfigurationStatus),
 		PendingReleaseIPConfigState: make(map[string]cns.IPConfigurationStatus),
@@ -81,6 +83,8 @@ func (ipm *IPStateManager) AddIPConfigs(ipconfigs []cns.IPConfigurationStatus) {
 	for i := 0; i < len(ipconfigs); i++ {
 
 		switch {
+		case ipconfigs[i].State == cns.PendingProgramming:
+			ipm.PendingProgramIPConfigState[ipconfigs[i].ID] = ipconfigs[i]
 		case ipconfigs[i].State == cns.Available:
 			ipm.AvailableIPConfigState[ipconfigs[i].ID] = ipconfigs[i]
 			ipm.AvailableIPIDStack.Push(ipconfigs[i].ID)
@@ -221,6 +225,15 @@ func (fake *HTTPServiceFake) SyncNodeStatus(string, string, string, json.RawMess
 	return 0, ""
 }
 
+func (fake *HTTPServiceFake) GetPendingProgramIPConfigs() []cns.IPConfigurationStatus {
+	ipconfigs := []cns.IPConfigurationStatus{}
+	for _, ipconfig := range fake.IPStateManager.PendingProgramIPConfigState {
+		ipconfigs = append(ipconfigs, ipconfig)
+	}
+
+	return ipconfigs
+}
+
 func (fake *HTTPServiceFake) GetAvailableIPConfigs() []cns.IPConfigurationStatus {
 	ipconfigs := []cns.IPConfigurationStatus{}
 	for _, ipconfig := range fake.IPStateManager.AvailableIPConfigState {

cns/fakes/imdsclientfake.go

@@ -21,7 +21,7 @@ func NewFakeImdsClient() *ImdsClientTest {
 	return &ImdsClientTest{}
 }
 
-// GetNetworkContainerInfoFromHost- Mock implementation to return Container version info.
+// GetNetworkContainerInfoFromHost - Mock implementation to return Container version info.
 func (imdsClient *ImdsClientTest) GetNetworkContainerInfoFromHost(networkContainerID string, primaryAddress string, authToken string, apiVersion string) (*imdsclient.ContainerVersion, error) {
 
 	ret := &imdsclient.ContainerVersion{}
@@ -47,3 +47,11 @@ func (imdsClient *ImdsClientTest) GetPrimaryInterfaceInfoFromMemory() (*imdsclie
 
 	return imdsClient.GetPrimaryInterfaceInfoFromHost()
 }
+
+// GetNetworkContainerInfoFromHostWithoutToken - Mock implementation to return host NMAgent NC version
+// Set it as 0 which is the same as default initial NC version for testing purpose
+func (imdsClient *ImdsClientTest) GetNetworkContainerInfoFromHostWithoutToken() int {
+	logger.Printf("[Azure CNS] get the NC version from NMAgent")
+
+	return 0
+}

cns/imdsclient/api.go

@@ -74,4 +74,5 @@ type ImdsClientInterface interface {
 	GetNetworkContainerInfoFromHost(networkContainerID string, primaryAddress string, authToken string, apiVersion string) (*ContainerVersion, error)
 	GetPrimaryInterfaceInfoFromHost() (*InterfaceInfo, error)
 	GetPrimaryInterfaceInfoFromMemory() (*InterfaceInfo, error)
+	GetNetworkContainerInfoFromHostWithoutToken() int
 }

cns/imdsclient/imdsclient.go

@@ -7,6 +7,7 @@ import (
 	"encoding/json"
 	"encoding/xml"
 	"fmt"
+	"math"
 	"net/http"
 	"strings"
 
@@ -132,3 +133,11 @@ func (imdsClient *ImdsClient) GetPrimaryInterfaceInfoFromMemory() (*InterfaceInf
 
 	return iface, err
 }
+
+// GetNetworkContainerInfoFromHostWithoutToken is a temp implementation which will be removed once background thread
+// updating host version is ready. Return max integer value to regress current AKS scenario
+func (imdsClient *ImdsClient) GetNetworkContainerInfoFromHostWithoutToken() int {
+	logger.Printf("[Azure CNS] GetNMagentVersionFromNMAgent")
+
+	return math.MaxInt64
+}

cns/ipampoolmonitor/ipampoolmonitor.go

@@ -64,12 +64,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
 	freeIPConfigCount := pm.cachedNNC.Spec.RequestedIPCount - int64(allocatedPodIPCount)
 
-	logger.Printf("[ipam-pool-monitor] Checking pool for resize, Pool Size: %v, Goal Size: %v, BatchSize: %v, MinFree: %v, MaxFree:%v, Allocated: %v, Available: %v, Pending Release: %v, Free: %v", cnsPodIPConfigCount, pm.cachedNNC.Spec.RequestedIPCount, pm.scalarUnits.BatchSize, pm.MinimumFreeIps, pm.MaximumFreeIps, allocatedPodIPCount, availableIPConfigCount, pendingReleaseIPCount, freeIPConfigCount)
+	logger.Printf("[ipam-pool-monitor] Checking pool for resize, 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 {
 	// pod count is increasing

cns/restserver/internalapi_test.go

@@ -35,8 +35,17 @@ func TestCreateOrUpdateNetworkContainerInternal(t *testing.T) {
 
 	setEnv(t)
 	setOrchestratorTypeInternal(cns.KubernetesCRD)
+	// NC version set as 0 which is the default initial value.
+	validateCreateOrUpdateNCInternal(t, 2, "0")
+}
+
+func TestCreateOrUpdateNCWithLargerVersionComparedToNMAgent(t *testing.T) {
+	restartService()
 
-	validateCreateOrUpdateNCInternal(t, 2)
+	setEnv(t)
+	setOrchestratorTypeInternal(cns.KubernetesCRD)
+	// NC version set as 1 which is larger than NC version get from mock nmagent.
+	validateCreateOrUpdateNCInternal(t, 2, "1")
 }
 
 func TestReconcileNCWithEmptyState(t *testing.T) {
@@ -69,7 +78,7 @@ func TestReconcileNCWithExistingState(t *testing.T) {
 		secondaryIPConfigs[ipId.String()] = secIpConfig
 		startingIndex++
 	}
-	req := generateNetworkContainerRequest(secondaryIPConfigs, "reconcileNc1")
+	req := generateNetworkContainerRequest(secondaryIPConfigs, "reconcileNc1", "0")
 
 	expectedAllocatedPods := make(map[string]cns.KubernetesPodInfo)
 	expectedAllocatedPods["10.0.0.6"] = cns.KubernetesPodInfo{
@@ -106,7 +115,7 @@ func TestReconcileNCWithSystemPods(t *testing.T) {
 		secondaryIPConfigs[ipId.String()] = secIpConfig
 		startingIndex++
 	}
-	req := generateNetworkContainerRequest(secondaryIPConfigs, uuid.New().String())
+	req := generateNetworkContainerRequest(secondaryIPConfigs, uuid.New().String(), "0")
 
 	expectedAllocatedPods := make(map[string]cns.KubernetesPodInfo)
 	expectedAllocatedPods["10.0.0.6"] = cns.KubernetesPodInfo{
@@ -135,7 +144,7 @@ func setOrchestratorTypeInternal(orchestratorType string) {
 	svc.state.OrchestratorType = orchestratorType
 }
 
-func validateCreateOrUpdateNCInternal(t *testing.T, secondaryIpCount int) {
+func validateCreateOrUpdateNCInternal(t *testing.T, secondaryIpCount int, ncVersion string) {
 	secondaryIPConfigs := make(map[string]cns.SecondaryIPConfig)
 	ncId := "testNc1"
 
@@ -148,7 +157,7 @@ func validateCreateOrUpdateNCInternal(t *testing.T, secondaryIpCount int) {
 		startingIndex++
 	}
 
-	createAndValidateNCRequest(t, secondaryIPConfigs, ncId)
+	createAndValidateNCRequest(t, secondaryIPConfigs, ncId, ncVersion)
 
 	// now Validate Update, add more secondaryIpConfig and it should handle the update
 	fmt.Println("Validate Scaleup")
@@ -160,7 +169,7 @@ func validateCreateOrUpdateNCInternal(t *testing.T, secondaryIpCount int) {
 		startingIndex++
 	}
 
-	createAndValidateNCRequest(t, secondaryIPConfigs, ncId)
+	createAndValidateNCRequest(t, secondaryIPConfigs, ncId, ncVersion)
 
 	// now Scale down, delete 3 ipaddresses from secondaryIpConfig req
 	fmt.Println("Validate Scale down")
@@ -174,19 +183,19 @@ func validateCreateOrUpdateNCInternal(t *testing.T, secondaryIpCount int) {
 		}
 	}
 
-	createAndValidateNCRequest(t, secondaryIPConfigs, ncId)
+	createAndValidateNCRequest(t, secondaryIPConfigs, ncId, ncVersion)
 
 	// Cleanup all SecondaryIps
 	fmt.Println("Validate no SecondaryIpconfigs")
 	for ipid := range secondaryIPConfigs {
 		delete(secondaryIPConfigs, ipid)
 	}
 
-	createAndValidateNCRequest(t, secondaryIPConfigs, ncId)
+	createAndValidateNCRequest(t, secondaryIPConfigs, ncId, ncVersion)
 }
 
-func createAndValidateNCRequest(t *testing.T, secondaryIPConfigs map[string]cns.SecondaryIPConfig, ncId string) {
-	req := generateNetworkContainerRequest(secondaryIPConfigs, ncId)
+func createAndValidateNCRequest(t *testing.T, secondaryIPConfigs map[string]cns.SecondaryIPConfig, ncId, ncVersion string) {
+	req := generateNetworkContainerRequest(secondaryIPConfigs, ncId, ncVersion)
 	returnCode := svc.CreateOrUpdateNetworkContainerInternal(req, fakes.NewFakeScalar(releasePercent, requestPercent, batchSize), fakes.NewFakeNodeNetworkConfigSpec(initPoolSize))
 	if returnCode != 0 {
 		t.Fatalf("Failed to createNetworkContainerRequest, req: %+v, err: %d", req, returnCode)
@@ -216,6 +225,14 @@ func validateNetworkRequest(t *testing.T, req cns.CreateNetworkContainerRequest)
 		t.Fatalf("Failed as Secondary IP count doesnt match in PodIpConfig state, expected:%d, actual %d", len(req.SecondaryIPConfigs), len(svc.PodIPConfigState))
 	}
 
+	var expectedIPStatus string
+	// 0 is the default NMAgent version return from fake GetNetworkContainerInfoFromHost
+	if containerStatus.VMVersion > "0" {
+		expectedIPStatus = cns.PendingProgramming
+	} else {
+		expectedIPStatus = cns.Available
+	}
+	t.Logf("VMVersion is %s, HostVersion is %s", containerStatus.VMVersion, containerStatus.HostVersion)
 	var alreadyValidated = make(map[string]string)
 	for ipid, ipStatus := range svc.PodIPConfigState {
 		if ipaddress, found := alreadyValidated[ipid]; !found {
@@ -240,9 +257,9 @@ func validateNetworkRequest(t *testing.T, req cns.CreateNetworkContainerRequest)
 					} else {
 						t.Fatalf("Failed to find podContext for allocated ip: %+v, podinfo :%+v", ipStatus, podInfo)
 					}
-				} else if ipStatus.State != cns.Available {
+				} else if ipStatus.State != expectedIPStatus {
 					// Todo: Validate for pendingRelease as well
-					t.Fatalf("IPId: %s State is not Available, ipStatus: %+v", ipid, ipStatus)
+					t.Fatalf("IPId: %s State is not as expected, ipStatus is : %+v, expected status is %+v", ipid, ipStatus.State, expectedIPStatus)
 				}
 
 				alreadyValidated[ipid] = ipStatus.IPAddress
@@ -256,7 +273,7 @@ func validateNetworkRequest(t *testing.T, req cns.CreateNetworkContainerRequest)
 	}
 }
 
-func generateNetworkContainerRequest(secondaryIps map[string]cns.SecondaryIPConfig, ncId string) cns.CreateNetworkContainerRequest {
+func generateNetworkContainerRequest(secondaryIps map[string]cns.SecondaryIPConfig, ncId string, ncVersion string) cns.CreateNetworkContainerRequest {
 	var ipConfig cns.IPConfiguration
 	ipConfig.DNSServers = dnsservers
 	ipConfig.GatewayIPAddress = gatewayIp
@@ -269,6 +286,7 @@ func generateNetworkContainerRequest(secondaryIps map[string]cns.SecondaryIPConf
 		NetworkContainerType: dockerContainerType,
 		NetworkContainerid:   ncId,
 		IPConfiguration:      ipConfig,
+		Version:              ncVersion,
 	}
 
 	req.SecondaryIPConfigs = make(map[string]cns.SecondaryIPConfig)

cns/restserver/ipam.go

@@ -118,6 +118,15 @@ func (service *HTTPRestService) GetPodIPConfigState() map[string]cns.IPConfigura
 	return service.PodIPConfigState
 }
 
+// GetPendingProgramIPConfigs returns list of IPs which are in pending program status
+func (service *HTTPRestService) GetPendingProgramIPConfigs() []cns.IPConfigurationStatus {
+	service.RLock()
+	defer service.RUnlock()
+	return filterIPConfigMap(service.PodIPConfigState, func(ipconfig cns.IPConfigurationStatus) bool {
+		return ipconfig.State == cns.PendingProgramming
+	})
+}
+
 func (service *HTTPRestService) GetAllocatedIPConfigs() []cns.IPConfigurationStatus {
 	service.RLock()
 	defer service.RUnlock()

cns/restserver/ipam_test.go

@@ -144,7 +144,7 @@ func UpdatePodIpConfigState(t *testing.T, svc *HTTPRestService, ipconfigs map[st
 		secondaryIPConfigs[ipId] = secIpConfig
 	}
 
-	createAndValidateNCRequest(t, secondaryIPConfigs, testNCID)
+	createAndValidateNCRequest(t, secondaryIPConfigs, testNCID, "0")
 
 	// update ipconfigs to expected state
 	for ipId, ipconfig := range ipconfigs {

cns/restserver/util.go

@@ -115,6 +115,9 @@ func (service *HTTPRestService) saveNetworkContainerGoalState(req cns.CreateNetw
 	if ok {
 		hostVersion = existingNCStatus.HostVersion
 		existingSecondaryIPConfigs = existingNCStatus.CreateNetworkContainerRequest.SecondaryIPConfigs
+	} else {
+		// Host version is the NC version from NMAgent, set it -1 to indicate no result from NMAgent yet.
+		hostVersion = "-1"
 	}
 
 	service.state.ContainerStatus[req.NetworkContainerid] =
@@ -166,7 +169,7 @@ func (service *HTTPRestService) saveNetworkContainerGoalState(req cns.CreateNetw
 
 		case cns.KubernetesCRD:
 			// Validate and Update the SecondaryIpConfig state
-			returnCode, returnMesage := service.updateIpConfigsStateUntransacted(req, existingSecondaryIPConfigs)
+			returnCode, returnMesage := service.updateIpConfigsStateUntransacted(req, existingSecondaryIPConfigs, hostVersion)
 			if returnCode != 0 {
 				return returnCode, returnMesage
 			}
@@ -182,21 +185,13 @@ func (service *HTTPRestService) saveNetworkContainerGoalState(req cns.CreateNetw
 		return UnsupportedNetworkContainerType, errMsg
 	}
 
-	service.state.ContainerStatus[req.NetworkContainerid] =
-		containerstatus{
-			ID:                            req.NetworkContainerid,
-			VMVersion:                     req.Version,
-			CreateNetworkContainerRequest: req,
-			HostVersion:                   hostVersion}
-
 	service.saveState()
 	return 0, ""
 }
 
-// This func will compute the deltaIpConfigState which needs to be updated (Added or Deleted)
-// from the inmemory map
+// This func will compute the deltaIpConfigState which needs to be updated (Added or Deleted) from the inmemory map
 // Note: Also this func is an untransacted API as the caller will take a Service lock
-func (service *HTTPRestService) updateIpConfigsStateUntransacted(req cns.CreateNetworkContainerRequest, existingSecondaryIPConfigs map[string]cns.SecondaryIPConfig) (int, string) {
+func (service *HTTPRestService) updateIpConfigsStateUntransacted(req cns.CreateNetworkContainerRequest, existingSecondaryIPConfigs map[string]cns.SecondaryIPConfig, hostVersion string) (int, string) {
 	// parse the existingSecondaryIpConfigState to find the deleted Ips
 	newIPConfigs := req.SecondaryIPConfigs
 	var tobeDeletedIpConfigs = make(map[string]cns.SecondaryIPConfig)
@@ -235,31 +230,39 @@ func (service *HTTPRestService) updateIpConfigsStateUntransacted(req cns.CreateN
 		}
 	}
 
-	// Add the newIpConfigs, ignore if ip state is already in the map
-	service.addIPConfigStateUntransacted(req.NetworkContainerid, newIPConfigs)
+	newNCVersion, _ := strconv.Atoi(req.Version)
+	nmagentNCVersion, _ := strconv.Atoi(hostVersion)
+
+	// TODO, remove this override when background thread which update nmagent version is ready.
+	nmagentNCVersion = service.imdsClient.GetNetworkContainerInfoFromHostWithoutToken()
+
+	if nmagentNCVersion >= newNCVersion {
+		service.addIPConfigStateUntransacted(cns.Available, req.NetworkContainerid, newIPConfigs)
+	} else {
+		service.addIPConfigStateUntransacted(cns.PendingProgramming, req.NetworkContainerid, newIPConfigs)
+	}
 
 	return 0, ""
 }
 
 // addIPConfigStateUntransacted adds the IPConfigs to the PodIpConfigState map with Available state
 // If the IP is already added then it will be an idempotent call. Also note, caller will
 // acquire/release the service lock.
-func (service *HTTPRestService) addIPConfigStateUntransacted(ncId string, ipconfigs map[string]cns.SecondaryIPConfig) {
+func (service *HTTPRestService) addIPConfigStateUntransacted(newIPCNSStatus, ncId string, ipconfigs map[string]cns.SecondaryIPConfig) {
 	// add ipconfigs to state
 	for ipId, ipconfig := range ipconfigs {
-		// if this IPConfig already exists in the map, then ignore as this is an idempotent state
 		if _, exists := service.PodIPConfigState[ipId]; exists {
 			continue
 		}
-
 		// add the new State
 		ipconfigStatus := cns.IPConfigurationStatus{
 			NCID:                ncId,
 			ID:                  ipId,
 			IPAddress:           ipconfig.IPAddress,
-			State:               cns.Available,
+			State:               newIPCNSStatus,
 			OrchestratorContext: nil,
 		}
+		logger.Printf("[Azure-Cns] Add IP %s as %s", ipconfig.IPAddress, newIPCNSStatus)
 
 		service.PodIPConfigState[ipId] = ipconfigStatus