Merge branch 'master' into statelessCNI-ACI

Author: behzad-mir

.pipelines/pipeline.yaml

@@ -261,6 +261,17 @@ stages:
         k8sVersion: ""
         dependsOn: "containerize"
 
+    # Cilium Nodesubnet E2E tests
+    - template: singletenancy/cilium-nodesubnet/cilium-nodesubnet-e2e-job-template.yaml
+      parameters:
+        name: "cilium_nodesubnet_e2e"
+        displayName: Cilium NodeSubnet
+        clusterType: nodesubnet-byocni-nokubeproxy-up
+        clusterName: "cilndsubnete2e"
+        vmSize: Standard_B2s
+        k8sVersion: ""
+        dependsOn: "containerize"
+
     # Cilium Overlay E2E tests
     - template: singletenancy/cilium-overlay/cilium-overlay-e2e-job-template.yaml
       parameters:
@@ -405,6 +416,7 @@ stages:
         - azure_overlay_stateless_e2e
         - aks_swift_e2e
         - cilium_e2e
+        - cilium_nodesubnet_e2e
         - cilium_overlay_e2e
         - cilium_h_overlay_e2e
         - aks_ubuntu_22_linux_e2e
@@ -425,6 +437,10 @@ stages:
               cilium_e2e:
                 name: cilium_e2e
                 clusterName: "ciliume2e"
+                region: $(REGION_AKS_CLUSTER_TEST)                
+              cilium_nodesubnet_e2e:
+                name: cilium_nodesubnet_e2e
+                clusterName: "cilndsubnete2e"
                 region: $(REGION_AKS_CLUSTER_TEST)
               cilium_overlay_e2e:
                 name: cilium_overlay_e2e

.pipelines/singletenancy/cilium-nodesubnet/cilium-nodesubnet-e2e-job-template.yaml

@@ -71,9 +71,9 @@ stages:
           os: ${{ parameters.os }}
           datapath: true
           dns: true
+          cni: cilium
           portforward: true
           service: true
-          hostport: true
           dependsOn: ${{ parameters.name }}
 
       - job: failedE2ELogs

.pipelines/singletenancy/cilium-nodesubnet/cilium-nodesubnet-e2e-step-template.yaml

@@ -58,7 +58,7 @@ steps:
         kubectl cluster-info
         kubectl get po -owide -A
         echo "install Cilium ${CILIUM_VERSION_TAG}"
-        export DIR=${CILIUM_VERSION_TAG%.*}
+        export DIR=$(echo ${CILIUM_VERSION_TAG#v} | cut -d. -f1,2)
         echo "installing files from ${DIR}"
         echo "deploy Cilium ConfigMap"
         kubectl apply -f test/integration/manifests/cilium/v${DIR}/cilium-config/cilium-config.yaml

Makefile

@@ -436,7 +436,8 @@ cni-image: ## build cni container image.
 		TARGET=$(OS) \
 		OS=$(OS) \
 		ARCH=$(ARCH) \
-		OS_VERSION=$(OS_VERSION)
+		OS_VERSION=$(OS_VERSION) \
+		EXTRA_BUILD_ARGS='--build-arg CNI_AI_PATH=$(CNI_AI_PATH) --build-arg CNI_AI_ID=$(CNI_AI_ID)'
 
 cni-image-push: ## push cni container image.
 	$(MAKE) container-push \

cni/Dockerfile

@@ -12,10 +12,12 @@ FROM --platform=linux/${ARCH} mcr.microsoft.com/cbl-mariner/base/core@sha256:a49
 FROM go AS azure-vnet
 ARG OS
 ARG VERSION
+ARG CNI_AI_PATH
+ARG CNI_AI_ID
 WORKDIR /azure-container-networking
 COPY . .
 RUN GOOS=$OS CGO_ENABLED=0 go build -a -o /go/bin/azure-vnet -trimpath -ldflags "-X main.version="$VERSION"" -gcflags="-dwarflocationlists=true" cni/network/plugin/main.go
-RUN GOOS=$OS CGO_ENABLED=0 go build -a -o /go/bin/azure-vnet-telemetry -trimpath -ldflags "-X main.version="$VERSION"" -gcflags="-dwarflocationlists=true" cni/telemetry/service/telemetrymain.go
+RUN GOOS=$OS CGO_ENABLED=0 go build -a -o /go/bin/azure-vnet-telemetry -trimpath -ldflags "-X main.version="$VERSION" -X "$CNI_AI_PATH"="$CNI_AI_ID"" -gcflags="-dwarflocationlists=true" cni/telemetry/service/telemetrymain.go
 RUN GOOS=$OS CGO_ENABLED=0 go build -a -o /go/bin/azure-vnet-ipam -trimpath -ldflags "-X main.version="$VERSION"" -gcflags="-dwarflocationlists=true" cni/ipam/plugin/main.go
 RUN GOOS=$OS CGO_ENABLED=0 go build -a -o /go/bin/azure-vnet-stateless -trimpath -ldflags "-X main.version="$VERSION"" -gcflags="-dwarflocationlists=true" cni/network/stateless/main.go
 

cni/telemetry/service/telemetrymain.go

@@ -167,8 +167,8 @@ func main() {
 		GetEnvRetryWaitTimeInSecs:    config.GetEnvRetryWaitTimeInSecs,
 	}
 
-	if tb.CreateAITelemetryHandle(aiConfig, config.DisableAll, config.DisableTrace, config.DisableMetric) != nil {
-		logger.Error("AI Handle creation error", zap.Error(err))
+	if err := tb.CreateAITelemetryHandle(aiConfig, config.DisableAll, config.DisableTrace, config.DisableMetric); err != nil { // nolint
+		logger.Error("AI Handle creation error:", zap.Error(err))
 	}
 	logger.Info("Report to host interval", zap.Duration("seconds", config.ReportToHostIntervalInSeconds))
 	tb.PushData(context.Background())

cns/fakes/nmagentclientfake.go

@@ -14,9 +14,10 @@ import (
 
 // NMAgentClientFake can be used to query to VM Host info.
 type NMAgentClientFake struct {
-	SupportedAPIsF    func(context.Context) ([]string, error)
-	GetNCVersionListF func(context.Context) (nmagent.NCVersionList, error)
-	GetHomeAzF        func(context.Context) (nmagent.AzResponse, error)
+	SupportedAPIsF      func(context.Context) ([]string, error)
+	GetNCVersionListF   func(context.Context) (nmagent.NCVersionList, error)
+	GetHomeAzF          func(context.Context) (nmagent.AzResponse, error)
+	GetInterfaceIPInfoF func(ctx context.Context) (nmagent.Interfaces, error)
 }
 
 func (n *NMAgentClientFake) SupportedAPIs(ctx context.Context) ([]string, error) {
@@ -30,3 +31,7 @@ func (n *NMAgentClientFake) GetNCVersionList(ctx context.Context) (nmagent.NCVer
 func (n *NMAgentClientFake) GetHomeAz(ctx context.Context) (nmagent.AzResponse, error) {
 	return n.GetHomeAzF(ctx)
 }
+
+func (n *NMAgentClientFake) GetInterfaceIPInfo(ctx context.Context) (nmagent.Interfaces, error) {
+	return n.GetInterfaceIPInfoF(ctx)
+}

cns/restserver/helper_for_nodesubnet_test.go

@@ -0,0 +1,89 @@
+package restserver
+
+import (
+	"context"
+	"net/netip"
+	"testing"
+
+	"github.com/Azure/azure-container-networking/cns"
+	"github.com/Azure/azure-container-networking/cns/common"
+	"github.com/Azure/azure-container-networking/cns/fakes"
+	"github.com/Azure/azure-container-networking/cns/nodesubnet"
+	acn "github.com/Azure/azure-container-networking/common"
+	"github.com/Azure/azure-container-networking/nmagent"
+	"github.com/Azure/azure-container-networking/store"
+)
+
+// GetRestServiceObjectForNodeSubnetTest creates a new HTTPRestService object for use in nodesubnet unit tests.
+func GetRestServiceObjectForNodeSubnetTest(t *testing.T, generator CNIConflistGenerator) *HTTPRestService {
+	config := &common.ServiceConfig{
+		Name:        "test",
+		Version:     "1.0",
+		ChannelMode: "AzureHost",
+		Store:       store.NewMockStore("test"),
+	}
+	interfaces := nmagent.Interfaces{
+		Entries: []nmagent.Interface{
+			{
+				MacAddress: nmagent.MACAddress{0x00, 0x0D, 0x3A, 0xF9, 0xDC, 0xA6},
+				IsPrimary:  true,
+				InterfaceSubnets: []nmagent.InterfaceSubnet{
+					{
+						Prefix: "10.0.0.0/24",
+						IPAddress: []nmagent.NodeIP{
+							{
+								Address:   nmagent.IPAddress(netip.AddrFrom4([4]byte{10, 0, 0, 4})),
+								IsPrimary: true,
+							},
+							{
+								Address:   nmagent.IPAddress(netip.AddrFrom4([4]byte{10, 0, 0, 52})),
+								IsPrimary: false,
+							},
+							{
+								Address:   nmagent.IPAddress(netip.AddrFrom4([4]byte{10, 0, 0, 63})),
+								IsPrimary: false,
+							},
+							{
+								Address:   nmagent.IPAddress(netip.AddrFrom4([4]byte{10, 0, 0, 45})),
+								IsPrimary: false,
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	svc, err := cns.NewService(config.Name, config.Version, config.ChannelMode, config.Store)
+	if err != nil {
+		return nil
+	}
+
+	svc.SetOption(acn.OptCnsURL, "")
+	svc.SetOption(acn.OptCnsPort, "")
+	err = svc.Initialize(config)
+	if err != nil {
+		return nil
+	}
+
+	t.Cleanup(func() { svc.Uninitialize() })
+
+	return &HTTPRestService{
+		Service:                  svc,
+		cniConflistGenerator:     generator,
+		state:                    &httpRestServiceState{},
+		PodIPConfigState:         make(map[string]cns.IPConfigurationStatus),
+		PodIPIDByPodInterfaceKey: make(map[string][]string),
+		nma: &fakes.NMAgentClientFake{
+			GetInterfaceIPInfoF: func(_ context.Context) (nmagent.Interfaces, error) {
+				return interfaces, nil
+			},
+		},
+		wscli: &fakes.WireserverClientFake{},
+	}
+}
+
+// GetNodesubnetIPFetcher gets the nodesubnetIPFetcher from the HTTPRestService.
+func (service *HTTPRestService) GetNodesubnetIPFetcher() *nodesubnet.IPFetcher {
+	return service.nodesubnetIPFetcher
+}

cns/restserver/nodesubnet.go

@@ -0,0 +1,64 @@
+package restserver
+
+import (
+	"context"
+	"net/netip"
+
+	"github.com/Azure/azure-container-networking/cns"
+	"github.com/Azure/azure-container-networking/cns/logger"
+	nodesubnet "github.com/Azure/azure-container-networking/cns/nodesubnet"
+	"github.com/Azure/azure-container-networking/cns/types"
+	"github.com/pkg/errors"
+)
+
+var _ nodesubnet.IPConsumer = &HTTPRestService{}
+
+// UpdateIPsForNodeSubnet updates the IP pool of HTTPRestService with newly fetched secondary IPs
+func (service *HTTPRestService) UpdateIPsForNodeSubnet(secondaryIPs []netip.Addr) error {
+	secondaryIPStrs := make([]string, len(secondaryIPs))
+	for i, ip := range secondaryIPs {
+		secondaryIPStrs[i] = ip.String()
+	}
+
+	networkContainerRequest := nodesubnet.CreateNodeSubnetNCRequest(secondaryIPStrs)
+
+	code, msg := service.saveNetworkContainerGoalState(*networkContainerRequest)
+	if code != types.Success {
+		return errors.Errorf("failed to save fetched ips. code: %d, message %s", code, msg)
+	}
+
+	logger.Debugf("IP change processed successfully")
+
+	// saved NC successfully. UpdateIPsForNodeSubnet is called only when IPs are fetched from NMAgent.
+	// We now have IPs to serve IPAM requests. Generate conflist to indicate CNS is ready
+	service.MustGenerateCNIConflistOnce()
+	return nil
+}
+
+// InitializeNodeSubnet prepares CNS for serving NodeSubnet requests.
+// It sets the orchestrator type to KubernetesCRD, reconciles the initial
+// CNS state from the statefile, then creates an IP fetcher.
+func (service *HTTPRestService) InitializeNodeSubnet(ctx context.Context, podInfoByIPProvider cns.PodInfoByIPProvider) error {
+	// set orchestrator type
+	orchestrator := cns.SetOrchestratorTypeRequest{
+		OrchestratorType: cns.KubernetesCRD,
+	}
+	service.SetNodeOrchestrator(&orchestrator)
+
+	if podInfoByIPProvider == nil {
+		logger.Printf("PodInfoByIPProvider is nil, this usually means no saved endpoint state. Skipping reconciliation")
+	} else if _, err := nodesubnet.ReconcileInitialCNSState(ctx, service, podInfoByIPProvider); err != nil {
+		return errors.Wrap(err, "reconcile initial CNS state")
+	}
+	// statefile (if any) is reconciled. Initialize the IP fetcher. Start the IP fetcher only after the service is started,
+	// because starting the IP fetcher will generate conflist, which should be done only once we are ready to respond to IPAM requests.
+	service.nodesubnetIPFetcher = nodesubnet.NewIPFetcher(service.nma, service, 0, 0, logger.Log)
+
+	return nil
+}
+
+// StartNodeSubnet starts the IP fetcher for NodeSubnet. This will cause secondary IPs to be fetched periodically.
+// After the first successful fetch, conflist will be generated to indicate CNS is ready.
+func (service *HTTPRestService) StartNodeSubnet(ctx context.Context) {
+	service.nodesubnetIPFetcher.Start(ctx)
+}