feat: CNS manages endpoints state for delegate IPAM use case

azure-ipam/ipconfig/ipconfig.go

@@ -32,6 +32,7 @@ func CreateIPConfigReq(args *cniSkel.CmdArgs) (cns.IPConfigRequest, error) {
 		PodInterfaceID:      args.ContainerID,
 		InfraContainerID:    args.ContainerID,
 		OrchestratorContext: orchestratorContext,
+		Ifname:              args.IfName,
 	}
 
 	return req, nil

cns/NetworkContainerContract.go

@@ -390,7 +390,7 @@ type IPConfigRequest struct {
 	PodInterfaceID      string
 	InfraContainerID    string
 	OrchestratorContext json.RawMessage
-	Ifname              string
+	Ifname              string // Used by delegated IPAM
 }
 
 func (i IPConfigRequest) String() string {

cns/azure-cns.yaml

@@ -104,6 +104,8 @@ spec:
               mountPath: /var/log
             - name: cns-state
               mountPath: /var/lib/azure-network
+            - name: azure-endpoints
+              mountPath: /var/run/azure-cns/
             - name: cns-config
               mountPath: /etc/azure-cns
             - name: cni-bin
@@ -130,6 +132,10 @@ spec:
                     fieldPath: spec.nodeName 
       hostNetwork: true
       volumes:
+        - name: azure-endpoints
+          hostPath:
+            path: /var/run/azure-cns/
+            type: DirectoryOrCreate
         - name: log
           hostPath:
             path: /var/log
@@ -179,5 +185,8 @@ data:
           "NodeSyncIntervalInSeconds": 30
       },
       "ChannelMode": "CRD",
-      "InitializeFromCNI": true
+      "InitializeFromCNI": true,
+      "ManageEndpointState": false,
+      "ProgramSNATIPTables" : false
     }
+# Toggle ManageEndpointState and ProgramSNATIPTables to true for delegated IPAM use case.

cns/client/client_test.go

@@ -169,7 +169,7 @@ func TestMain(m *testing.M) {
 	logger.InitLogger(logName, 0, 0, tmpLogDir+"/")
 	config := common.ServiceConfig{}
 
-	httpRestService, err := restserver.NewHTTPRestService(&config, &fakes.WireserverClientFake{}, &fakes.NMAgentClientFake{})
+	httpRestService, err := restserver.NewHTTPRestService(&config, &fakes.WireserverClientFake{}, &fakes.NMAgentClientFake{}, nil)
 	svc = httpRestService.(*restserver.HTTPRestService)
 	svc.Name = "cns-test-server"
 	fakeNNC := v1alpha.NodeNetworkConfig{

cns/cnireconciler/podinfoprovider.go

@@ -6,6 +6,8 @@ import (
 	"github.com/Azure/azure-container-networking/cni/api"
 	"github.com/Azure/azure-container-networking/cni/client"
 	"github.com/Azure/azure-container-networking/cns"
+	"github.com/Azure/azure-container-networking/cns/restserver"
+	"github.com/Azure/azure-container-networking/store"
 	"github.com/pkg/errors"
 	"k8s.io/utils/exec"
 )
@@ -16,6 +18,27 @@ func NewCNIPodInfoProvider() (cns.PodInfoByIPProvider, error) {
 	return newCNIPodInfoProvider(exec.New())
 }
 
+func NewCNSPodInfoProvider(endpointStore store.KeyValueStore) (cns.PodInfoByIPProvider, error) {
+	return newCNSPodInfoProvider(endpointStore)
+}
+
+func newCNSPodInfoProvider(endpointStore store.KeyValueStore) (cns.PodInfoByIPProvider, error) {
+	var state map[string]*restserver.EndpointInfo
+	err := endpointStore.Read(restserver.EndpointStoreKey, &state)
+	if err != nil {
+		if errors.Is(err, store.ErrKeyNotFound) {
+			// Nothing to restore.
+			return cns.PodInfoByIPProviderFunc(func() (map[string]cns.PodInfo, error) {
+				return endpointStateToPodInfoByIP(state)
+			}), err
+		}
+		return nil, fmt.Errorf("failed to read endpoints state from store : %w", err)
+	}
+	return cns.PodInfoByIPProviderFunc(func() (map[string]cns.PodInfo, error) {
+		return endpointStateToPodInfoByIP(state)
+	}), nil
+}
+
 func newCNIPodInfoProvider(exec exec.Interface) (cns.PodInfoByIPProvider, error) {
 	cli := client.New(exec)
 	state, err := cli.GetEndpointState()
@@ -44,3 +67,34 @@ func cniStateToPodInfoByIP(state *api.AzureCNIState) (map[string]cns.PodInfo, er
 	}
 	return podInfoByIP, nil
 }
+
+func endpointStateToPodInfoByIP(state map[string]*restserver.EndpointInfo) (map[string]cns.PodInfo, error) {
+	podInfoByIP := map[string]cns.PodInfo{}
+	for containerID, endpointInfo := range state { // for each endpoint
+		for ifname, ipinfo := range endpointInfo.IfnameToIPMap { // for each IP info object of the endpoint's interfaces
+			for _, ipv4conf := range ipinfo.IPv4 { // for each IPv4 config of the endpoint's interfaces
+				if _, ok := podInfoByIP[ipv4conf.IP.String()]; ok {
+					return nil, errors.Wrap(cns.ErrDuplicateIP, ipv4conf.IP.String())
+				}
+				podInfoByIP[ipv4conf.IP.String()] = cns.NewPodInfo(
+					containerID,
+					ifname,
+					endpointInfo.PodName,
+					endpointInfo.PodNamespace,
+				)
+			}
+			for _, ipv6conf := range ipinfo.IPv6 { // for each IPv6 config of the endpoint's interfaces
+				if _, ok := podInfoByIP[ipv6conf.IP.String()]; ok {
+					return nil, errors.Wrap(cns.ErrDuplicateIP, ipv6conf.IP.String())
+				}
+				podInfoByIP[ipv6conf.IP.String()] = cns.NewPodInfo(
+					containerID,
+					ifname,
+					endpointInfo.PodName,
+					endpointInfo.PodNamespace,
+				)
+			}
+		}
+	}
+	return podInfoByIP, nil
+}

cns/cnireconciler/podinfoprovider_test.go

@@ -1,9 +1,12 @@
 package cnireconciler
 
 import (
+	"net"
 	"testing"
 
 	"github.com/Azure/azure-container-networking/cns"
+	"github.com/Azure/azure-container-networking/cns/restserver"
+	"github.com/Azure/azure-container-networking/store"
 	testutils "github.com/Azure/azure-container-networking/test/utils"
 	"github.com/stretchr/testify/assert"
 	"k8s.io/utils/exec"
@@ -59,3 +62,49 @@ func TestNewCNIPodInfoProvider(t *testing.T) {
 		})
 	}
 }
+
+func TestNewCNSPodInfoProvider(t *testing.T) {
+	goodStore := store.NewMockStore("")
+	goodEndpointState := make(map[string]*restserver.EndpointInfo)
+	endpointInfo := &restserver.EndpointInfo{PodName: "goldpinger-deploy-bbbf9fd7c-z8v4l", PodNamespace: "default", IfnameToIPMap: make(map[string]*restserver.IPInfo)}
+	endpointInfo.IfnameToIPMap["eth0"] = &restserver.IPInfo{IPv4: []net.IPNet{{IP: net.IPv4(10, 241, 0, 65), Mask: net.IPv4Mask(255, 255, 255, 0)}}}
+
+	goodEndpointState["0a4917617e15d24dc495e407d8eb5c88e4406e58fa209e4eb75a2c2fb7045eea"] = endpointInfo
+	err := goodStore.Write(restserver.EndpointStoreKey, goodEndpointState)
+	if err != nil {
+		t.Fatalf("Error writing to store: %v", err)
+	}
+	tests := []struct {
+		name    string
+		store   store.KeyValueStore
+		want    map[string]cns.PodInfo
+		wantErr bool
+	}{
+		{
+			name:    "good",
+			store:   goodStore,
+			want:    map[string]cns.PodInfo{"10.241.0.65": cns.NewPodInfo("0a4917617e15d24dc495e407d8eb5c88e4406e58fa209e4eb75a2c2fb7045eea", "eth0", "goldpinger-deploy-bbbf9fd7c-z8v4l", "default")},
+			wantErr: false,
+		},
+		{
+			name:    "empty store",
+			store:   store.NewMockStore(""),
+			want:    map[string]cns.PodInfo{},
+			wantErr: true,
+		},
+	}
+
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := newCNSPodInfoProvider(tt.store)
+			if tt.wantErr {
+				assert.Error(t, err)
+				return
+			}
+			assert.NoError(t, err)
+			podInfoByIP, _ := got.PodInfoByIP()
+			assert.Equal(t, tt.want, podInfoByIP)
+		})
+	}
+}

cns/configuration/configuration.go

@@ -37,6 +37,7 @@ type CNSConfig struct {
 	KeyVaultSettings            KeyVaultSettings
 	MSISettings                 MSISettings
 	ProgramSNATIPTables         bool
+	ManageEndpointState         bool
 }
 
 type TelemetrySettings struct {

cns/logger/cnslogger.go

@@ -85,6 +85,17 @@ func (c *CNSLogger) Debugf(format string, args ...any) {
 	c.sendTraceInternal(msg)
 }
 
+func (c *CNSLogger) Warnf(format string, args ...any) {
+	c.logger.Warnf(format, args...)
+
+	if c.th == nil || c.DisableTraceLogging {
+		return
+	}
+
+	msg := fmt.Sprintf(format, args...)
+	c.sendTraceInternal(msg)
+}
+
 func (c *CNSLogger) Errorf(format string, args ...any) {
 	c.logger.Errorf(format, args...)
 

cns/logger/log.go

@@ -37,6 +37,10 @@ func Debugf(format string, args ...any) {
 	Log.Debugf(format, args...)
 }
 
+func Warnf(format string, args ...any) {
+	Log.Warnf(format, args...)
+}
+
 func LogEvent(event aitelemetry.Event) {
 	Log.LogEvent(event)
 }

cns/restserver/api_test.go

@@ -1078,7 +1078,7 @@ func startService() error {
 	}
 
 	nmagentClient := &fakes.NMAgentClientFake{}
-	service, err = NewHTTPRestService(&config, &fakes.WireserverClientFake{}, nmagentClient)
+	service, err = NewHTTPRestService(&config, &fakes.WireserverClientFake{}, nmagentClient, nil)
 	if err != nil {
 		return err
 	}

cns/restserver/const.go

@@ -2,9 +2,10 @@ package restserver
 
 const (
 	// Key against which CNS state is persisted.
-	storeKey = "ContainerNetworkService"
-	attach   = "Attach"
-	detach   = "Detach"
+	storeKey         = "ContainerNetworkService"
+	EndpointStoreKey = "Endpoints"
+	attach           = "Attach"
+	detach           = "Detach"
 	// Rest service state identifier for named lock
 	stateJoinedNetworks = "JoinedNetworks"
 	dncApiVersion       = "?api-version=2018-03-01"