fix: [NPM] policy translation correctness and limits

npm/pkg/controlplane/controllers/common/cache.go

@@ -46,6 +46,7 @@ const (
 type GenericCache interface {
 	GetPod(*Input) (*NpmPod, error)
 	GetNamespaceLabel(namespace string, key string) string
+	GetNamespaceLabels(namespace string) (map[string]string, bool)
 	GetListMap() map[string]string
 	GetSetMap() map[string]string
 }
@@ -86,6 +87,15 @@ func (c *Cache) GetNamespaceLabel(namespace, labelkey string) string {
 	return ""
 }
 
+// GetNamespaceLabels distinguishes a missing namespace from one with no labels.
+func (c *Cache) GetNamespaceLabels(namespace string) (map[string]string, bool) {
+	ns, ok := c.NsMap[namespace]
+	if !ok || ns == nil {
+		return nil, false
+	}
+	return ns.LabelsMap, true
+}
+
 func (c *Cache) GetSetMap() map[string]string {
 	return c.SetMap
 }

npm/pkg/controlplane/controllers/v2/namespaceController.go

@@ -465,12 +465,12 @@ func (nsc *NamespaceController) cleanDeletedNamespace(cachedNsKey string) error
 		cachedNsObj.RemoveLabelsWithKey(nsLabelKey)
 	}
 
-	allNamespacesSet := ipsets.NewIPSetMetadata(util.KubeAllNamespacesFlag, ipsets.KeyLabelOfNamespace)
+	allNamespacesSet := ipsets.NewIPSetMetadata(util.KubeAllNamespacesFlagV2, ipsets.KeyLabelOfNamespace)
 	toBeDeletedCachedKey := []*ipsets.IPSetMetadata{ipsets.NewIPSetMetadata(cachedNsKey, ipsets.Namespace)}
 
 	// Delete the namespace from all-namespace ipset list.
 	if err = nsc.dp.RemoveFromList(allNamespacesSet, toBeDeletedCachedKey); err != nil {
-		metrics.SendErrorLogAndMetric(util.NSID, "[DeleteNamespace] Error: failed to delete namespace %s from ipset list %s with err: %v", cachedNsKey, util.KubeAllNamespacesFlag, err)
+		metrics.SendErrorLogAndMetric(util.NSID, "[DeleteNamespace] Error: failed to delete namespace %s from ipset list %s with err: %v", cachedNsKey, util.KubeAllNamespacesFlagV2, err)
 		return fmt.Errorf("failed to remove from list during clean deleted namespace %w", err)
 	}
 

npm/pkg/controlplane/controllers/v2/networkPolicyController.go

@@ -186,6 +186,10 @@ func (c *NetworkPolicyController) processNextWorkItem() bool {
 		// Run the syncNetPol, passing it the namespace/name string of the
 		// network policy resource to be synced.
 		if err := c.syncNetPol(key); err != nil {
+			if errors.Is(err, errNetPolTranslationFailure) {
+				c.workqueue.Forget(obj)
+				return fmt.Errorf("error syncing '%s': %w; waiting for a policy change", key, err)
+			}
 			// Put the item back on the workqueue to handle any transient errors.
 			c.workqueue.AddRateLimited(key)
 			return fmt.Errorf("error syncing '%s': %w, requeuing", key, err)
@@ -291,18 +295,25 @@ func (c *NetworkPolicyController) syncAddAndUpdateNetPol(netPolObj *networkingv1
 	// install translated rules into kernel
 	npmNetPolObj, err := translation.TranslatePolicy(netPolObj, c.npmLiteToggle)
 	if err != nil {
-		if isUnsupportedWindowsTranslationErr(err) {
-			klog.Warningf("NetworkPolicy %s in namespace %s is not translated because it has unsupported translated features of Windows: %s",
+		if isUnsupportedTranslationErr(err, c.npmLiteToggle) {
+			klog.Warningf("NetworkPolicy %s in namespace %s is not translated because it uses a feature this datapath does not support: %s",
 				netPolObj.ObjectMeta.Name, netPolObj.ObjectMeta.Namespace, err.Error())
 
 			// We can safely suppress unsupported network policy because re-Queuing will result in same error.
 			// The exec time isn't relevant here, so consider a no-op.
 			return metrics.NoOp, nil
 		}
 
-		klog.Errorf("Failed to translate podSelector in NetworkPolicy %s in namespace %s: %s", netPolObj.ObjectMeta.Name, netPolObj.ObjectMeta.Namespace, err.Error())
-		// The exec time isn't relevant here, so consider a no-op. Returning nil to prevent re-queuing since this is not a transient error.
-		return metrics.NoOp, nil
+		// Translation depends only on the policy and fixed controller mode. Full NPM
+		// reports deterministic failures without retrying or caching an unapplied spec;
+		// the informer queues changed resource versions. Dataplane errors below remain
+		// retryable, and Lite keeps its existing error handling. The worker reports errors.
+		if !c.npmLiteToggle {
+			return metrics.NoOp, fmt.Errorf("%w %s/%s: %w",
+				errNetPolTranslationFailure, netPolObj.Namespace, netPolObj.Name, err)
+		}
+		return metrics.NoOp, fmt.Errorf("translating network policy %s/%s: %w",
+			netPolObj.Namespace, netPolObj.Name, err)
 	}
 
 	_, policyExisted := c.rawNpSpecMap[netpolKey]
@@ -358,3 +369,19 @@ func isUnsupportedWindowsTranslationErr(err error) bool {
 		errors.Is(err, translation.ErrUnsupportedSCTP) ||
 		errors.Is(err, translation.ErrUnsupportedExceptCIDR)
 }
+
+// isUnsupportedTranslationErr reports whether err is a deliberate limitation of the datapath
+// or mode NPM is running in, rather than a policy NPM failed to translate. Those limitations
+// stay suppressed with a warning; other failures must be reported without recording success.
+func isUnsupportedTranslationErr(err error, npmLiteToggle bool) bool {
+	if errors.Is(err, util.ErrInvalidCIDR) || errors.Is(err, translation.ErrInvalidIPBlockExcept) {
+		return false
+	}
+	// Full NPM supplies a typed cause; only Lite retains unclassified address errors.
+	unsupportedAddress := errors.Is(err, util.ErrUnsupportedIPFamily) ||
+		(npmLiteToggle && errors.Is(err, translation.ErrUnsupportedIPAddress))
+	return isUnsupportedWindowsTranslationErr(err) ||
+		(util.IsWindowsDP() && unsupportedAddress) ||
+		// NPM Lite only supports CIDR peers; a label-selector peer is out of scope there.
+		errors.Is(err, translation.ErrUnsupportedNonCIDR)
+}

npm/pkg/controlplane/controllers/v2/networkPolicyController_retry_test.go

@@ -0,0 +1,211 @@
+package controllers
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/Azure/azure-container-networking/npm/pkg/controlplane/translation"
+	dpmocks "github.com/Azure/azure-container-networking/npm/pkg/dataplane/mocks"
+	"github.com/Azure/azure-container-networking/npm/util"
+	"github.com/golang/mock/gomock"
+	"github.com/stretchr/testify/require"
+	networkingv1 "k8s.io/api/networking/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/util/workqueue"
+)
+
+func newNetPolQueueFixture(t *testing.T, policy *networkingv1.NetworkPolicy, dp *dpmocks.MockGenericDataplane, npmLite bool) *netPolFixture {
+	t.Helper()
+	f := newNetPolFixture(t)
+	f.netPolLister = append(f.netPolLister, policy)
+	f.kubeobjects = append(f.kubeobjects, policy)
+	f.newNetPolController(nil, dp, npmLite)
+	f.netPolController.workqueue.ShutDown()
+	// Zero-delay retries make incorrect requeueing observable without sleeps.
+	f.netPolController.workqueue = workqueue.NewTypedRateLimitingQueue[any](workqueue.NewTypedItemFastSlowRateLimiter[any](0, 0, 1))
+	t.Cleanup(f.netPolController.workqueue.ShutDown)
+	return f
+}
+
+func TestFullNPMTranslationFailureWaitsForPolicyChange(t *testing.T) {
+	oversized := netPolWithCIDR("192.0.2.0/24")
+	selector := &metav1.LabelSelector{}
+	for i := 0; i < 19; i++ {
+		selector.MatchExpressions = append(selector.MatchExpressions, metav1.LabelSelectorRequirement{
+			Key: fmt.Sprintf("key%d", i), Operator: metav1.LabelSelectorOpIn, Values: []string{"a", "b"},
+		})
+	}
+	oversized.Spec.Ingress[0].From = []networkingv1.NetworkPolicyPeer{{NamespaceSelector: selector}}
+	invalidWithExcept := netPolWithCIDR("192.0.2.0/33")
+	invalidWithExcept.Spec.Ingress[0].From[0].IPBlock.Except = []string{"192.0.2.1/32"}
+	unknownOperator := netPolWithCIDR("192.0.2.0/24")
+	unknownOperator.Spec.Ingress[0].From = []networkingv1.NetworkPolicyPeer{{
+		NamespaceSelector: &metav1.LabelSelector{MatchExpressions: []metav1.LabelSelectorRequirement{{
+			Key: "tenant", Operator: "Unknown",
+		}}},
+	}}
+
+	tests := []struct {
+		name   string
+		policy *networkingv1.NetworkPolicy
+		cause  error
+	}{
+		{"malformed CIDR", netPolWithCIDR("192.0.2.0/33"), util.ErrInvalidCIDR},
+		{"malformed CIDR with Except", invalidWithExcept, util.ErrInvalidCIDR},
+		{"selector expansion", oversized, translation.ErrTooManyFlattenedSelectors},
+		{"unsupported operator", unknownOperator, translation.ErrUnsupportedMatchExpressionOperator},
+	}
+	if !util.IsWindowsDP() {
+		tests = append(tests, struct {
+			name   string
+			policy *networkingv1.NetworkPolicy
+			cause  error
+		}{"unsupported family", netPolWithCIDR("2001:db8::/32"), util.ErrUnsupportedIPFamily})
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			test.policy.ResourceVersion = "1"
+			ctrl := gomock.NewController(t)
+			dp := dpmocks.NewMockGenericDataplane(ctrl)
+			f := newNetPolQueueFixture(t, test.policy, dp, false)
+			c := f.netPolController
+			key := getKey(test.policy, t)
+
+			_, err := c.syncAddAndUpdateNetPol(test.policy)
+			require.ErrorIs(t, err, test.cause)
+			require.ErrorIs(t, err, errNetPolTranslationFailure)
+			require.ErrorContains(t, err, key)
+			require.Empty(t, c.rawNpSpecMap)
+
+			c.addNetworkPolicy(test.policy)
+			require.Equal(t, 1, c.workqueue.Len())
+			require.True(t, c.processNextWorkItem())
+			require.Zero(t, c.workqueue.NumRequeues(key))
+			require.Zero(t, c.workqueue.Len())
+			require.Empty(t, c.rawNpSpecMap)
+
+			c.updateNetworkPolicy(test.policy, test.policy.DeepCopy())
+			require.Zero(t, c.workqueue.Len(), "an informer resync must not repeat the rejection")
+
+			corrected := test.policy.DeepCopy()
+			corrected.ResourceVersion = "2"
+			corrected.Spec = netPolWithCIDR("10.0.0.0/0").Spec
+			require.NoError(t, f.kubeInformer.Networking().V1().NetworkPolicies().Informer().GetIndexer().Update(corrected))
+			dp.EXPECT().UpdatePolicy(gomock.Any()).Return(nil).Times(1)
+			c.updateNetworkPolicy(test.policy, corrected)
+			require.Equal(t, 1, c.workqueue.Len())
+			require.True(t, c.processNextWorkItem())
+			require.Equal(t, &corrected.Spec, c.rawNpSpecMap[key])
+			require.Zero(t, c.workqueue.NumRequeues(key))
+			require.Zero(t, c.workqueue.Len())
+		})
+	}
+}
+
+func TestFullNPMDataplaneFailureStillRetries(t *testing.T) {
+	policy := netPolWithCIDR("10.0.0.0/0")
+	ctrl := gomock.NewController(t)
+	dp := dpmocks.NewMockGenericDataplane(ctrl)
+	f := newNetPolQueueFixture(t, policy, dp, false)
+	c := f.netPolController
+	key := getKey(policy, t)
+
+	gomock.InOrder(
+		dp.EXPECT().UpdatePolicy(gomock.Any()).Return(fmt.Errorf("programming policy: %w", translation.ErrUnsupportedIPAddress)),
+		dp.EXPECT().UpdatePolicy(gomock.Any()).Return(nil),
+	)
+	c.addNetworkPolicy(policy)
+	require.True(t, c.processNextWorkItem())
+	require.Equal(t, 1, c.workqueue.NumRequeues(key))
+	require.Equal(t, 1, c.workqueue.Len())
+	require.Empty(t, c.rawNpSpecMap)
+
+	require.True(t, c.processNextWorkItem())
+	require.Zero(t, c.workqueue.NumRequeues(key))
+	require.Zero(t, c.workqueue.Len())
+	require.Equal(t, &policy.Spec, c.rawNpSpecMap[key])
+}
+
+func TestFullNPMRejectedUpdateRetainsAppliedPolicyUntilDeletion(t *testing.T) {
+	policy := netPolWithCIDR("192.0.2.0/24")
+	policy.ResourceVersion = "1"
+	ctrl := gomock.NewController(t)
+	dp := dpmocks.NewMockGenericDataplane(ctrl)
+	f := newNetPolQueueFixture(t, policy, dp, false)
+	c := f.netPolController
+	key := getKey(policy, t)
+
+	dp.EXPECT().UpdatePolicy(gomock.Any()).Return(nil).Times(1)
+	c.addNetworkPolicy(policy)
+	require.True(t, c.processNextWorkItem())
+	require.Equal(t, &policy.Spec, c.rawNpSpecMap[key])
+
+	rejected := policy.DeepCopy()
+	rejected.ResourceVersion = "2"
+	rejected.Spec.Ingress[0].From[0].IPBlock.CIDR = "192.0.2.0/33"
+	indexer := f.kubeInformer.Networking().V1().NetworkPolicies().Informer().GetIndexer()
+	require.NoError(t, indexer.Update(rejected))
+	c.updateNetworkPolicy(policy, rejected)
+	require.Equal(t, 1, c.workqueue.Len())
+	require.True(t, c.processNextWorkItem())
+	require.Equal(t, &policy.Spec, c.rawNpSpecMap[key])
+	require.Zero(t, c.workqueue.NumRequeues(key))
+	require.Zero(t, c.workqueue.Len())
+
+	require.NoError(t, indexer.Delete(rejected))
+	dp.EXPECT().RemovePolicy(key).Return(nil).Times(1)
+	c.deleteNetworkPolicy(rejected)
+	require.Equal(t, 1, c.workqueue.Len())
+	require.True(t, c.processNextWorkItem())
+	require.Empty(t, c.rawNpSpecMap)
+	require.Zero(t, c.workqueue.Len())
+}
+
+func TestFullNPMRejectedCreateCanBeDeleted(t *testing.T) {
+	policy := netPolWithCIDR("invalid")
+	ctrl := gomock.NewController(t)
+	dp := dpmocks.NewMockGenericDataplane(ctrl)
+	f := newNetPolQueueFixture(t, policy, dp, false)
+	c := f.netPolController
+
+	c.addNetworkPolicy(policy)
+	require.True(t, c.processNextWorkItem())
+	require.Zero(t, c.workqueue.Len())
+	require.Empty(t, c.rawNpSpecMap)
+
+	require.NoError(t, f.kubeInformer.Networking().V1().NetworkPolicies().Informer().GetIndexer().Delete(policy))
+	c.deleteNetworkPolicy(policy)
+	require.Equal(t, 1, c.workqueue.Len())
+	require.True(t, c.processNextWorkItem())
+	require.Empty(t, c.rawNpSpecMap)
+	require.Zero(t, c.workqueue.Len())
+}
+
+func TestLiteTranslationFailureHandlingIsUnchanged(t *testing.T) {
+	policy := netPolWithCIDR("invalid")
+	ctrl := gomock.NewController(t)
+	dp := dpmocks.NewMockGenericDataplane(ctrl)
+	f := newNetPolQueueFixture(t, policy, dp, true)
+	c := f.netPolController
+	key := getKey(policy, t)
+
+	_, err := c.syncAddAndUpdateNetPol(policy)
+	if util.IsWindowsDP() {
+		require.NoError(t, err, "the legacy Lite direct-address limitation stays suppressed")
+	} else {
+		require.ErrorIs(t, err, util.ErrInvalidCIDR)
+		require.NotErrorIs(t, err, errNetPolTranslationFailure)
+	}
+
+	c.addNetworkPolicy(policy)
+	require.True(t, c.processNextWorkItem())
+	require.Empty(t, c.rawNpSpecMap)
+	if util.IsWindowsDP() {
+		require.Zero(t, c.workqueue.NumRequeues(key))
+		require.Zero(t, c.workqueue.Len())
+	} else {
+		require.Equal(t, 1, c.workqueue.NumRequeues(key))
+		require.Equal(t, 1, c.workqueue.Len())
+	}
+}