[NPM Lite] Querying L1VH + Non-L1VH Endpoints

npm/cmd/start.go

@@ -137,6 +137,8 @@ func start(config npmconfig.Config, flags npmconfig.Flags) error {
 	stopChannel := wait.NeverStop
 	if config.Toggles.EnableV2NPM {
 		// update the dataplane config
+		npmV2DataplaneCfg.EnableNPMLite = config.Toggles.EnableNPMLite
+
 		npmV2DataplaneCfg.MaxBatchedACLsPerPod = config.MaxBatchedACLsPerPod
 
 		npmV2DataplaneCfg.NetPolInBackground = config.Toggles.NetPolInBackground

npm/pkg/dataplane/dataplane.go

@@ -45,6 +45,7 @@ type Config struct {
 	NetPolInBackground bool
 	MaxPendingNetPols  int
 	NetPolInterval     time.Duration
+	EnableNPMLite      bool
 	*ipsets.IPSetManagerCfg
 	*policies.PolicyManagerCfg
 }
@@ -64,12 +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
-	applyInfo      *applyInfo
-	netPolQueue    *netPolQueue
+	endpointCache     *endpointCache
+	ioShim            *common.IOShim
+	updatePodCache    *updatePodCache
+	endpointQuery     *endpointQuery
+	endpointQueryL1VH *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
@@ -88,11 +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),
+		networkID:         "",
+		endpointCache:     newEndpointCache(),
+		nodeName:          nodeName,
+		ioShim:            ioShim,
+		endpointQuery:     new(endpointQuery),
+		endpointQueryL1VH: new(endpointQuery),
 		applyInfo: &applyInfo{
 			inBootupPhase: true,
 		},
@@ -128,7 +131,6 @@ func NewDataPlane(nodeName string, ioShim *common.IOShim, cfg *Config, stopChann
 	} else {
 		metrics.SendLog(util.DaemonDataplaneID, "[DataPlane] dataplane configured to NOT add netpols in background", true)
 	}
-
 	return dp, nil
 }
 

npm/pkg/dataplane/dataplane_windows.go

@@ -2,7 +2,6 @@ package dataplane
 
 import (
 	"encoding/json"
-	"errors"
 	"fmt"
 	"strings"
 	"time"
@@ -12,6 +11,7 @@ import (
 	"github.com/Azure/azure-container-networking/npm/util"
 	npmerrors "github.com/Azure/azure-container-networking/npm/util/errors"
 	"github.com/Microsoft/hcsshim/hcn"
+	"github.com/pkg/errors"
 	"k8s.io/klog"
 )
 
@@ -50,14 +50,32 @@ 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{
+		SchemaVersion: hcn.SchemaVersion{
+			Major: hcnSchemaMajorVersion,
+			Minor: hcnSchemaMinorVersion,
+		},
+		Flags: hcn.HostComputeQueryFlagsNone,
+	}
+
 	// Filter out any endpoints that are not in "AttachedShared" State. All running Windows pods with networking must be in this state.
 	filterMap := map[string]uint16{"State": hcnEndpointStateAttachedSharing}
 	filter, err := json.Marshal(filterMap)
 	if err != nil {
-		return npmerrors.SimpleErrorWrapper("failed to marshal endpoint filter map", err)
+		return errors.Wrap(err, "failed to marshal endpoint filter map")
 	}
 	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)
+	}
+
 	// reset endpoint cache so that netpol references are removed for all endpoints while refreshing pod endpoints
 	// no need to lock endpointCache at boot up
 	dp.endpointCache.cache = make(map[string]*npmEndpoint)
@@ -334,9 +352,19 @@ func (dp *DataPlane) getLocalPodEndpoints() ([]*hcn.HostComputeEndpoint, error)
 	metrics.RecordListEndpointsLatency(timer)
 	if err != nil {
 		metrics.IncListEndpointsFailures()
-		return nil, npmerrors.SimpleErrorWrapper("failed to get local pod endpoints", err)
+		return nil, errors.Wrap(err, "failed to get local pod endpoints")
 	}
 
+	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...)
+	}
 	epPointers := make([]*hcn.HostComputeEndpoint, 0, len(endpoints))
 	for k := range endpoints {
 		epPointers = append(epPointers, &endpoints[k])