Add a go routine to update NC host version from NMAgent periodically. (#714)

* Add a go routine to update NC host version from NMAgent periodically.
If orchestrator type is CRD, update pending programming IPs as well.

* Update NC version in test from 0 to -1, which will allow default IP state as Avaialable instead of pending programming.

* Add secondary IP status updation when reconcile.
Resovle conflicts manually.

Update unit test nc version value.

Update unit test nc version.

Add get nmagent default value back for integ testing purpose.
Unit test can be break by this change.

Update default new IP CNS status to available.

Assign value to host version if none exist in util.go

Addressed feedback and perform cluster integ test with 1 sec frequent nc version update.
Need to clean logNCSnapshots when send out PR.

Update nc version associate with secondary ip. Add new nmagent api test.

Add versionResponseWithoutToken.Containers log

Add containerId from our runner sub.
Add containerId from NMAgent team.

Addressed feedback and add real nmagent logic.

Add timeout when query nmagent for nc version.

* Update comments.

* Add context background with timeout function for syncing node nc version.

* Add 5 second force update CNS pending programming IP to available logic.

* Resovle merge conflict from master.

* Debug and it pass all the test.
This is the final version.

Change the way of http get request to add context. Change channel to no buffer with same goroutine.
Found always fall in ctx.Done() condition.
Add channel close for get nc version list.
Add milisecond unit for timeout.

Testing with different context version.

* Resolve merge conflict.

* Remove force update pending programming IP to available logic. Remain retry if no response from NMAgent.
Release pending programming IP when scale down.

* Remain VMVersion, HostVersion variable name as it is and use the Version inside CreateNetworkContainerRequest.

* Addressed team member feedback.

Author: csfmomo

cns/cnsclient/cnsclient_test.go

@@ -53,6 +53,7 @@ func addTestStateToRestServer(t *testing.T, secondaryIps []string) {
 	for _, secIpAddress := range secondaryIps {
 		secIpConfig := cns.SecondaryIPConfig{
 			IPAddress: secIpAddress,
+			NCVersion: -1,
 		}
 		ipId := uuid.New()
 		secondaryIPConfigs[ipId.String()] = secIpConfig
@@ -63,6 +64,9 @@ func addTestStateToRestServer(t *testing.T, secondaryIps []string) {
 		NetworkContainerid:   "testNcId1",
 		IPConfiguration:      ipConfig,
 		SecondaryIPConfigs:   secondaryIPConfigs,
+		// Set it as -1 to be same as default host version.
+		// It will allow secondary IPs status to be set as available.
+		Version: "-1",
 	}
 
 	returnCode := svc.CreateOrUpdateNetworkContainerInternal(req, fakes.NewFakeScalar(releasePercent, requestPercent, batchSize), fakes.NewFakeNodeNetworkConfigSpec(initPoolSize))
@@ -122,7 +126,7 @@ func TestMain(m *testing.M) {
 	logger.InitLogger(logName, 0, 0, tmpLogDir+"/")
 	config := common.ServiceConfig{}
 
-	httpRestService, err := restserver.NewHTTPRestService(&config, fakes.NewFakeImdsClient())
+	httpRestService, err := restserver.NewHTTPRestService(&config, fakes.NewFakeImdsClient(), fakes.NewFakeNMAgentClient())
 	svc = httpRestService.(*restserver.HTTPRestService)
 	svc.Name = "cns-test-server"
 	svc.IPAMPoolMonitor = fakes.NewIPAMPoolMonitorFake()

cns/configuration/configuration.go

@@ -6,6 +6,7 @@ import (
 	"io/ioutil"
 	"os"
 	"path/filepath"
+	"time"
 
 	"github.com/Azure/azure-container-networking/cns"
 	"github.com/Azure/azure-container-networking/cns/logger"
@@ -17,14 +18,17 @@ const (
 )
 
 type CNSConfig struct {
-	TelemetrySettings  TelemetrySettings
-	ManagedSettings    ManagedSettings
-	ChannelMode        string
-	UseHTTPS           bool
-	TLSSubjectName     string
-	TLSCertificatePath string
-	TLSPort            string
-	WireserverIP       string
+	TelemetrySettings           TelemetrySettings
+	ManagedSettings             ManagedSettings
+	ChannelMode                 string
+	UseHTTPS                    bool
+	TLSSubjectName              string
+	TLSCertificatePath          string
+	TLSPort                     string
+	TLSEndpoint                 string
+	WireserverIP                string
+	SyncHostNCVersionIntervalMs time.Duration
+	SyncHostNCTimeoutMs         time.Duration
 }
 
 type TelemetrySettings struct {
@@ -121,11 +125,13 @@ func setManagedSettingDefaults(managedSettings *ManagedSettings) {
 	}
 }
 
-// Set Default values of CNS config if not specified
+// SetCNSConfigDefaults set default values of CNS config if not specified
 func SetCNSConfigDefaults(config *CNSConfig) {
 	setTelemetrySettingDefaults(&config.TelemetrySettings)
 	setManagedSettingDefaults(&config.ManagedSettings)
 	if config.ChannelMode == "" {
 		config.ChannelMode = cns.Direct
 	}
+	config.SyncHostNCVersionIntervalMs = 1000
+	config.SyncHostNCTimeoutMs = 500
 }

cns/fakes/cnsfake.go

@@ -1,9 +1,11 @@
 package fakes
 
 import (
+	"context"
 	"encoding/json"
 	"errors"
 	"sync"
+	"time"
 
 	"github.com/Azure/azure-container-networking/cns"
 	"github.com/Azure/azure-container-networking/cns/common"
@@ -230,6 +232,11 @@ func (fake *HTTPServiceFake) SyncNodeStatus(string, string, string, json.RawMess
 	return 0, ""
 }
 
+// SyncHostNCVersion will update HostVersion in containerstatus.
+func (fake *HTTPServiceFake) SyncHostNCVersion(context.Context, string, time.Duration) {
+	return
+}
+
 func (fake *HTTPServiceFake) GetPendingProgramIPConfigs() []cns.IPConfigurationStatus {
 	ipconfigs := []cns.IPConfigurationStatus{}
 	for _, ipconfig := range fake.IPStateManager.PendingProgramIPConfigState {

cns/fakes/nmagentclientfake.go

@@ -0,0 +1,22 @@
+// Copyright 2020 Microsoft. All rights reserved.
+// MIT License
+
+package fakes
+
+// NMAgentClientTest can be used to query to VM Host info.
+type NMAgentClientTest struct {
+}
+
+// NewFakeNMAgentClient return a mock implemetation of NMAgentClient
+func NewFakeNMAgentClient() *NMAgentClientTest {
+	return &NMAgentClientTest{}
+}
+
+// GetNcVersionListWithOutToken is mock implementation to return nc version list.
+func (nmagentclient *NMAgentClientTest) GetNcVersionListWithOutToken(ncNeedUpdateList []string) map[string]int {
+	ncVersionList := make(map[string]int)
+	for _, ncID := range ncNeedUpdateList {
+		ncVersionList[ncID] = 0
+	}
+	return ncVersionList
+}

cns/nmagentclient/nmagentclient.go

@@ -5,20 +5,25 @@ import (
 	"encoding/json"
 	"encoding/xml"
 	"fmt"
+	"io/ioutil"
 	"net/http"
+	"strconv"
+	"time"
 
 	"github.com/Azure/azure-container-networking/cns/logger"
 	"github.com/Azure/azure-container-networking/common"
 )
 
 const (
 	//GetNmAgentSupportedApiURLFmt Api endpoint to get supported Apis of NMAgent
-	GetNmAgentSupportedApiURLFmt     = "http://%s/machine/plugins/?comp=nmagent&type=GetSupportedApis"
-	GetNetworkContainerVersionURLFmt = "http://%s/machine/plugins/?comp=nmagent&type=NetworkManagement/interfaces/%s/networkContainers/%s/version/authenticationToken/%s/api-version/1"
+	GetNmAgentSupportedApiURLFmt       = "http://%s/machine/plugins/?comp=nmagent&type=GetSupportedApis"
+	GetNetworkContainerVersionURLFmt   = "http://%s/machine/plugins/?comp=nmagent&type=NetworkManagement/interfaces/%s/networkContainers/%s/version/authenticationToken/%s/api-version/1"
+	GetNcVersionListWithOutTokenURLFmt = "http://%s/machine/plugins/?comp=nmagent&type=NetworkManagement/interfaces/api-version/%s"
 )
 
 //WireServerIP - wire server ip
 var WireserverIP = "168.63.129.16"
+var getNcVersionListWithOutTokenURLVersion = "2"
 
 // NMANetworkContainerResponse - NMAgent response.
 type NMANetworkContainerResponse struct {
@@ -31,6 +36,36 @@ type NMAgentSupportedApisResponseXML struct {
 	SupportedApis []string `xml:"type"`
 }
 
+type ContainerInfo struct {
+	NetworkContainerID string `json:"networkContainerId"`
+	Version            string `json:"version"`
+}
+
+type NMANetworkContainerListResponse struct {
+	ResponseCode string          `json:"httpStatusCode"`
+	Containers   []ContainerInfo `json:"networkContainers"`
+}
+
+// NMAgentClient is client to handle queries to nmagent
+type NMAgentClient struct {
+	connectionURL string
+}
+
+// NMAgentClientInterface has interface that nmagent client will handle
+type NMAgentClientInterface interface {
+	GetNcVersionListWithOutToken(ncNeedUpdateList []string) map[string]int
+}
+
+// NewNMAgentClient create a new nmagent client.
+func NewNMAgentClient(url string) (*NMAgentClient, error) {
+	if url == "" {
+		url = fmt.Sprintf(GetNcVersionListWithOutTokenURLFmt, WireserverIP, getNcVersionListWithOutTokenURLVersion)
+	}
+	return &NMAgentClient{
+		connectionURL: url,
+	}, nil
+}
+
 // JoinNetwork joins the given network
 func JoinNetwork(
 	networkID string,
@@ -149,3 +184,42 @@ func GetNmAgentSupportedApis(
 	logger.Printf("[NMAgentClient][Response] GetNmAgentSupportedApis. Response: %+v.", response)
 	return xmlDoc.SupportedApis, nil
 }
+
+// GetNcVersionListWithOutToken query nmagent for programmed container version.
+func (nmagentclient *NMAgentClient) GetNcVersionListWithOutToken(ncNeedUpdateList []string) map[string]int {
+	ncVersionList := make(map[string]int)
+	now := time.Now()
+	response, err := http.Get(nmagentclient.connectionURL)
+	latency := time.Since(now)
+	logger.Printf("[NMAgentClient][Response] GetNcVersionListWithOutToken response: %+v, latency is %d", response, latency.Milliseconds())
+
+	if response.StatusCode != http.StatusOK {
+		logger.Printf("[NMAgentClient][Response] GetNcVersionListWithOutToken failed with %d, err is %v", response.StatusCode, err)
+		return nil
+	}
+
+	var nmaNcListResponse NMANetworkContainerListResponse
+	rBytes, _ := ioutil.ReadAll(response.Body)
+	logger.Printf("Response body is %v", rBytes)
+	json.Unmarshal(rBytes, &nmaNcListResponse)
+	if nmaNcListResponse.ResponseCode != strconv.Itoa(http.StatusOK) {
+		logger.Printf("[NMAgentClient][Response] GetNcVersionListWithOutToken unmarshal failed with %s", rBytes)
+		return nil
+	}
+
+	var receivedNcVersionListInMap = make(map[string]string)
+	for _, containers := range nmaNcListResponse.Containers {
+		receivedNcVersionListInMap[containers.NetworkContainerID] = containers.Version
+	}
+	for _, ncID := range ncNeedUpdateList {
+		if version, ok := receivedNcVersionListInMap[ncID]; ok {
+			if versionInInt, err := strconv.Atoi(version); err != nil {
+				logger.Printf("[NMAgentClient][Response] GetNcVersionListWithOutToken translate version %s to int failed with %s", version, err)
+			} else {
+				ncVersionList[ncID] = versionInInt
+				logger.Printf("Containers id is %s, programmed NC version is %d", ncID, versionInInt)
+			}
+		}
+	}
+	return ncVersionList
+}

cns/requestcontroller/kubecontroller/crdtranslator.go

@@ -6,6 +6,7 @@ import (
 	"strconv"
 
 	"github.com/Azure/azure-container-networking/cns"
+	"github.com/Azure/azure-container-networking/log"
 	nnc "github.com/Azure/azure-container-networking/nodenetworkconfig/api/v1alpha"
 )
 
@@ -66,7 +67,10 @@ func CRDStatusToNCRequest(crdStatus nnc.NodeNetworkConfigStatus) (cns.CreateNetw
 				NCVersion: ncVersion,
 			}
 			ncRequest.SecondaryIPConfigs[ipAssignment.Name] = secondaryIPConfig
+			log.Debugf("Seconday IP Configs got set, name is %s, config is %v", ipAssignment.Name, secondaryIPConfig)
 		}
+		log.Printf("Set NC request info with NetworkContainerid %s, NetworkContainerType %s, NC Version %s",
+			ncRequest.NetworkContainerid, ncRequest.NetworkContainerType, ncRequest.Version)
 	}
 
 	//Only returning the first network container for now, later we will return a list

cns/restserver/api_test.go

@@ -912,7 +912,7 @@ func startService() {
 	var err error
 	// Create the service.
 	config := common.ServiceConfig{}
-	service, err = NewHTTPRestService(&config, fakes.NewFakeImdsClient())
+	service, err = NewHTTPRestService(&config, fakes.NewFakeImdsClient(), fakes.NewFakeNMAgentClient())
 	if err != nil {
 		fmt.Printf("Failed to create CNS object %v\n", err)
 		os.Exit(1)

cns/restserver/const.go

@@ -38,6 +38,7 @@ const (
 	NetworkContainerVfpProgramComplete     = 35
 	NetworkContainerVfpProgramCheckSkipped = 36
 	NmAgentSupportedApisError              = 37
+	UnsupportedNCVersion                   = 38
 	UnexpectedError                        = 99
 )
 

cns/restserver/internalapi.go

@@ -5,11 +5,14 @@ package restserver
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
 	"fmt"
 	"net/http"
 	"net/http/httptest"
 	"reflect"
+	"strconv"
+	"time"
 
 	"github.com/Azure/azure-container-networking/cns"
 	"github.com/Azure/azure-container-networking/cns/logger"
@@ -143,6 +146,64 @@ func (service *HTTPRestService) SyncNodeStatus(dncEP, infraVnet, nodeID string,
 	return
 }
 
+// SyncHostNCVersion will check NC version from NMAgent and save it as host NC version in container status.
+// If NMAgent NC version got updated, CNS will refresh the pending programming IP status.
+func (service *HTTPRestService) SyncHostNCVersion(ctx context.Context, channelMode string, syncHostNCTimeoutMilliSec time.Duration) {
+	var hostVersionNeedUpdateNcList []string
+	service.RLock()
+	for _, containerstatus := range service.state.ContainerStatus {
+		// Will open a separate PR to convert all the NC version related variable to int. Change from string to int is a pain.
+		hostVersion, err := strconv.Atoi(containerstatus.HostVersion)
+		if err != nil {
+			log.Errorf("Received err when change containerstatus.HostVersion %s to int, err msg %v", containerstatus.HostVersion, err)
+			continue
+		}
+		dncNcVersion, err := strconv.Atoi(containerstatus.CreateNetworkContainerRequest.Version)
+		if err != nil {
+			log.Errorf("Received err when change nc version %s in containerstatus to int, err msg %v", containerstatus.CreateNetworkContainerRequest.Version, err)
+			continue
+		}
+		// host NC version is the NC version from NMAgent, if it's smaller than NC version from DNC, then append it to indicate it needs update.
+		if hostVersion < dncNcVersion {
+			hostVersionNeedUpdateNcList = append(hostVersionNeedUpdateNcList, containerstatus.ID)
+		} else if hostVersion > dncNcVersion {
+			log.Errorf("NC version from NMAgent is larger than DNC, NC version from NMAgent is %d, NC version from DNC is %d", hostVersion, dncNcVersion)
+		}
+	}
+	service.RUnlock()
+	if len(hostVersionNeedUpdateNcList) > 0 {
+		ncVersionChannel := make(chan map[string]int)
+		ctxWithTimeout, _ := context.WithTimeout(ctx, syncHostNCTimeoutMilliSec*time.Millisecond)
+		go func() {
+			ncVersionChannel <- service.nmagentClient.GetNcVersionListWithOutToken(hostVersionNeedUpdateNcList)
+			close(ncVersionChannel)
+		}()
+		select {
+		case newHostNCVersionList := <-ncVersionChannel:
+			if newHostNCVersionList == nil {
+				logger.Errorf("Can't get vfp programmed NC version list from url without token")
+			} else {
+				service.Lock()
+				for ncID, newHostNCVersion := range newHostNCVersionList {
+					// Check whether it exist in service state and get the related nc info
+					if ncInfo, exist := service.state.ContainerStatus[ncID]; !exist {
+						logger.Errorf("Can't find NC with ID %s in service state, stop updating this host NC version", ncID)
+					} else {
+						if channelMode == cns.CRD {
+							service.MarkIpsAsAvailableUntransacted(ncInfo.ID, newHostNCVersion)
+						}
+						ncInfo.HostVersion = strconv.Itoa(newHostNCVersion)
+						service.state.ContainerStatus[ncID] = ncInfo
+					}
+				}
+				service.Unlock()
+			}
+		case <-ctxWithTimeout.Done():
+			logger.Errorf("Timeout when getting vfp programmed NC version list from url without token")
+		}
+	}
+}
+
 // This API will be called by CNS RequestController on CRD update.
 func (service *HTTPRestService) ReconcileNCState(ncRequest *cns.CreateNetworkContainerRequest, podInfoByIp map[string]cns.KubernetesPodInfo, scalar nnc.Scaler, spec nnc.NodeNetworkConfigSpec) int {
 	// check if ncRequest is null, then return as there is no CRD state yet