[NPM Lite] Querying L1VH + Non-L1VH Endpoints

npm/npm.go

@@ -192,7 +192,7 @@ func (npMgr *NetworkPolicyManager) Start(config npmconfig.Config, stopCh <-chan
 	// Starts all informers manufactured by npMgr's informerFactory.
 	npMgr.InformerFactory.Start(stopCh)
 
-	// npn lite
+	// npm lite
 	if npMgr.NpmLiteToggle {
 		npMgr.PodInformerFactory.Start(stopCh)
 	}

npm/pkg/dataplane/dataplane.go

@@ -65,13 +65,13 @@ type DataPlane struct {
 	nodeName           string
 	// endpointCache stores all endpoints of the network (including off-node)
 	// Key is PodIP
-	endpointCache     *endpointCache
-	ioShim            *common.IOShim
-	updatePodCache    *updatePodCache
-	endpointQuery     *endpointQuery
-	endpointQueryL1VH *endpointQuery // windows -> filter for state 2 (attached) endpoints in l1vh
-	applyInfo         *applyInfo
-	netPolQueue       *netPolQueue
+	endpointCache              *endpointCache
+	ioShim                     *common.IOShim
+	updatePodCache             *updatePodCache
+	endpointQuery              *endpointQuery
+	endpointQueryAttachedState *endpointQuery // windows -> filter for state 2 (attached) endpoints in l1vh
+	applyInfo                  *applyInfo
+	netPolQueue                *netPolQueue
 	// removePolicyInfo tracks when a policy was removed yet had ApplyIPSet failures.
 	// This field is only relevant for Linux.
 	removePolicyInfo removePolicyInfo
@@ -90,12 +90,12 @@ func NewDataPlane(nodeName string, ioShim *common.IOShim, cfg *Config, stopChann
 		policyMgr: policies.NewPolicyManager(ioShim, cfg.PolicyManagerCfg),
 		ipsetMgr:  ipsets.NewIPSetManager(cfg.IPSetManagerCfg, ioShim),
 		// networkID is set when initializing Windows dataplane
-		networkID:         "",
-		endpointCache:     newEndpointCache(),
-		nodeName:          nodeName,
-		ioShim:            ioShim,
-		endpointQuery:     new(endpointQuery),
-		endpointQueryL1VH: new(endpointQuery),
+		networkID:                  "",
+		endpointCache:              newEndpointCache(),
+		nodeName:                   nodeName,
+		ioShim:                     ioShim,
+		endpointQuery:              new(endpointQuery),
+		endpointQueryAttachedState: new(endpointQuery),
 		applyInfo: &applyInfo{
 			inBootupPhase: true,
 		},

npm/pkg/dataplane/dataplane_windows.go

@@ -51,7 +51,7 @@ func (dp *DataPlane) initializeDataPlane() error {
 		Flags: hcn.HostComputeQueryFlagsNone,
 	}
 	// Initialize Endpoint query used to filter healthy endpoints (vNIC) of Windows pods on L1VH Node
-	dp.endpointQueryL1VH.query = hcn.HostComputeQuery{
+	dp.endpointQueryAttachedState.query = hcn.HostComputeQuery{
 		SchemaVersion: hcn.SchemaVersion{
 			Major: hcnSchemaMajorVersion,
 			Minor: hcnSchemaMinorVersion,
@@ -63,18 +63,17 @@ func (dp *DataPlane) initializeDataPlane() error {
 	filterMap := map[string]uint16{"State": hcnEndpointStateAttachedSharing}
 	filter, err := json.Marshal(filterMap)
 	if err != nil {
-		return errors.Wrap(err, "failed to marshal endpoint filter map")
+		return errors.Wrap(err, "failed to marshal endpoint filter map for attachedsharing state")
 	}
 	dp.endpointQuery.query.Filter = string(filter)
 
-	if dp.EnableNPMLite {
-		filterMapL1VH := map[string]uint16{"State": hcnEndpointStateAttached}
-		filterL1VH, errL1VH := json.Marshal(filterMapL1VH)
-		if errL1VH != nil {
-			return errors.Wrap(errL1VH, "failed to marshal endpoint filter map")
-		}
-		dp.endpointQueryL1VH.query.Filter = string(filterL1VH)
+	// Filter out any endpoints that are not in "Attached" State. All running Windows L1VH pods with networking must be in this state.
+	filterMapAttached := map[string]uint16{"State": hcnEndpointStateAttached}
+	filterAttached, err := json.Marshal(filterMapAttached)
+	if err != nil {
+		return errors.Wrap(err, "failed to marshal endpoint filter map for attched state")
 	}
+	dp.endpointQueryAttachedState.query.Filter = string(filterAttached)
 
 	// reset endpoint cache so that netpol references are removed for all endpoints while refreshing pod endpoints
 	// no need to lock endpointCache at boot up
@@ -347,31 +346,67 @@ func (dp *DataPlane) getEndpointsToApplyPolicies(netPols []*policies.NPMNetworkP
 
 func (dp *DataPlane) getLocalPodEndpoints() ([]*hcn.HostComputeEndpoint, error) {
 	klog.Info("getting local endpoints")
+
+	// Gets endpoints in state: Attached
 	timer := metrics.StartNewTimer()
-	endpoints, err := dp.ioShim.Hns.ListEndpointsQuery(dp.endpointQuery.query)
+	endpointsAttached, err := dp.ioShim.Hns.ListEndpointsQuery(dp.endpointQueryAttachedState.query)
 	metrics.RecordListEndpointsLatency(timer)
 	if err != nil {
 		metrics.IncListEndpointsFailures()
-		return nil, errors.Wrap(err, "failed to get local pod endpoints")
+		return nil, errors.Wrap(err, "failed to get local pod endpoints in state:attached")
 	}
 
-	if dp.EnableNPMLite {
-		timer = metrics.StartNewTimer()
-		endpointsAttached, errL1vh := dp.ioShim.Hns.ListEndpointsQuery(dp.endpointQueryL1VH.query)
-		metrics.RecordListEndpointsLatency(timer)
-		if errL1vh != nil {
-			metrics.IncListEndpointsFailures()
-			return nil, errors.Wrap(errL1vh, "failed to get local pod endpoints in L1VH")
-		}
-		endpoints = append(endpoints, endpointsAttached...)
+	// Gets endpoints in state: AttachedSharing
+	timer = metrics.StartNewTimer()
+	endpoints, err := dp.ioShim.Hns.ListEndpointsQuery(dp.endpointQuery.query)
+	metrics.RecordListEndpointsLatency(timer)
+	if err != nil {
+		metrics.IncListEndpointsFailures()
+		return nil, errors.Wrap(err, "failed to get local pod endpoints in state: attachedSharing")
 	}
+
+	// Filtering out any of the same endpoints between endpoints with state attached and attachedSharing
+	endpoints = removeCommonEndpoints(&endpoints, &endpointsAttached)
+
 	epPointers := make([]*hcn.HostComputeEndpoint, 0, len(endpoints))
 	for k := range endpoints {
 		epPointers = append(epPointers, &endpoints[k])
 	}
 	return epPointers, nil
 }
 
+func removeCommonEndpoints(endpoints, endpointsAttached *[]hcn.HostComputeEndpoint) []hcn.HostComputeEndpoint {
+	// Choose smaller and larger lists based on length for efficiency
+	smallerEndpointsList, largerEndpointsList := endpoints, endpointsAttached
+	if len(*endpoints) > len(*endpointsAttached) {
+		smallerEndpointsList, largerEndpointsList = endpointsAttached, endpoints
+	}
+
+	// Store IDs of smaller list in a map for quick lookup
+	idMap := make(map[string]struct{}, len(*smallerEndpointsList))
+	for i := 0; i < len(*smallerEndpointsList); i++ {
+		ep := &(*smallerEndpointsList)[i]
+		idMap[ep.Id] = struct{}{}
+	}
+
+	// Append endpoints from larger list and remove common IDs from map
+	result := []hcn.HostComputeEndpoint{}
+	for i := 0; i < len(*largerEndpointsList); i++ {
+		ep := (*largerEndpointsList)[i]
+		result = append(result, ep)
+		delete(idMap, ep.Id)
+	}
+
+	// Append remaining unique endpoints from smaller list to result
+	for i := 0; i < len(*smallerEndpointsList); i++ {
+		ep := (*smallerEndpointsList)[i]
+		if _, found := idMap[ep.Id]; found {
+			result = append(result, ep)
+		}
+	}
+	return result
+}
+
 // refreshPodEndpoints will refresh all the pod endpoints and create empty netpol references for new endpoints
 /*
 Key Assumption: a new pod event (w/ IP) cannot come before HNS knows (and can tell us) about the endpoint.

npm/pkg/dataplane/dataplane_windows_test.go

@@ -2,6 +2,7 @@ package dataplane
 
 import (
 	"fmt"
+	"reflect"
 	"sync"
 	"testing"
 	"time"
@@ -10,6 +11,7 @@ import (
 	"github.com/Azure/azure-container-networking/npm/metrics"
 	"github.com/Azure/azure-container-networking/npm/pkg/dataplane/ipsets"
 	dptestutils "github.com/Azure/azure-container-networking/npm/pkg/dataplane/testutils"
+	"github.com/Microsoft/hcsshim/hcn"
 	"github.com/pkg/errors"
 	"github.com/stretchr/testify/require"
 )
@@ -86,6 +88,72 @@ func TestMultiJobApplyInBackground(t *testing.T) {
 	testMultiJobCases(t, multiJobApplyInBackgroundTests(), time.Duration(1*time.Second))
 }
 
+func TestRemoveCommonEndpoints(t *testing.T) {
+	tests := []struct {
+		name              string
+		endpoints         []hcn.HostComputeEndpoint
+		endpointsAttached []hcn.HostComputeEndpoint
+		expected          []hcn.HostComputeEndpoint
+	}{
+		{
+			name:              "1 value same",
+			endpoints:         []hcn.HostComputeEndpoint{{Id: "456901"}, {Id: "123456"}, {Id: "560971"}},
+			endpointsAttached: []hcn.HostComputeEndpoint{{Id: "567890"}, {Id: "123456"}, {Id: "789012"}},
+			expected:          []hcn.HostComputeEndpoint{{Id: "567890"}, {Id: "123456"}, {Id: "789012"}, {Id: "456901"}, {Id: "560971"}},
+		},
+		{
+			name:              "no values same",
+			endpoints:         []hcn.HostComputeEndpoint{{Id: "456901"}, {Id: "560971"}},
+			endpointsAttached: []hcn.HostComputeEndpoint{{Id: "567890"}, {Id: "789012"}},
+			expected:          []hcn.HostComputeEndpoint{{Id: "567890"}, {Id: "789012"}, {Id: "456901"}, {Id: "560971"}},
+		},
+		{
+			name:              "1 value same",
+			endpoints:         []hcn.HostComputeEndpoint{{Id: "456901"}, {Id: "123456"}, {Id: "560971"}},
+			endpointsAttached: []hcn.HostComputeEndpoint{{Id: "567890"}, {Id: "123456"}, {Id: "789012"}},
+			expected:          []hcn.HostComputeEndpoint{{Id: "567890"}, {Id: "123456"}, {Id: "789012"}, {Id: "456901"}, {Id: "560971"}},
+		},
+		{
+			name:              "two values same",
+			endpoints:         []hcn.HostComputeEndpoint{{Id: "456901"}, {Id: "560971"}, {Id: "123456"}, {Id: "789012"}},
+			endpointsAttached: []hcn.HostComputeEndpoint{{Id: "567890"}, {Id: "789012"}, {Id: "123456"}},
+			expected:          []hcn.HostComputeEndpoint{{Id: "456901"}, {Id: "560971"}, {Id: "123456"}, {Id: "789012"}, {Id: "567890"}},
+		},
+		{
+			name:              "no values",
+			endpoints:         []hcn.HostComputeEndpoint{},
+			endpointsAttached: []hcn.HostComputeEndpoint{},
+			expected:          []hcn.HostComputeEndpoint{},
+		},
+		{
+			name:              "1 value - same",
+			endpoints:         []hcn.HostComputeEndpoint{{Id: "456901"}},
+			endpointsAttached: []hcn.HostComputeEndpoint{{Id: "456901"}},
+			expected:          []hcn.HostComputeEndpoint{{Id: "456901"}},
+		},
+		{
+			name:              "1 value - different",
+			endpoints:         []hcn.HostComputeEndpoint{{Id: "456901"}},
+			endpointsAttached: []hcn.HostComputeEndpoint{},
+			expected:          []hcn.HostComputeEndpoint{{Id: "456901"}},
+		},
+	}
+	for _, tt := range tests {
+		tt := tt
+
+		t.Run(tt.name, func(t *testing.T) {
+			result := removeCommonEndpoints(&tt.endpoints, &tt.endpointsAttached)
+			// Use reflect.DeepEqual as a backup if require.Equal doesn't work as expected
+			if !reflect.DeepEqual(tt.expected, result) {
+				t.Errorf("Test %s failed: expected %v, got %v", tt.name, tt.expected, result)
+			}
+
+			// Or, if require.Equal works fine, it will display a descriptive error message
+			require.Equal(t, tt.expected, result, "expected array equals result")
+		})
+	}
+}
+
 func testSerialCases(t *testing.T, tests []*SerialTestCase, finalSleep time.Duration) {
 	for i, tt := range tests {
 		i := i