Adds retry mechanism for Network Policy

go-controller/pkg/factory/factory.go

@@ -32,6 +32,7 @@ import (
 	v1coreinformers "k8s.io/client-go/informers/core/v1"
 	"k8s.io/client-go/kubernetes"
 	listers "k8s.io/client-go/listers/core/v1"
+	netlisters "k8s.io/client-go/listers/networking/v1"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/klog/v2"
 )
@@ -586,6 +587,12 @@ func (wf *WatchFactory) GetNamespacesBySelector(labelSelector metav1.LabelSelect
 	return namespaceLister.List(selector)
 }
 
+// GetNetworkPolicy gets a specific network policy by the namespace/name
+func (wf *WatchFactory) GetNetworkPolicy(namespace, name string) (*knet.NetworkPolicy, error) {
+	networkPolicyLister := wf.informers[policyType].lister.(netlisters.NetworkPolicyLister)
+	return networkPolicyLister.NetworkPolicies(namespace).Get(name)
+}
+
 func (wf *WatchFactory) NodeInformer() cache.SharedIndexInformer {
 	return wf.informers[nodeType].inf
 }

go-controller/pkg/ovn/namespace.go

@@ -365,8 +365,18 @@ func (oc *Controller) deleteNamespace(ns *kapi.Namespace) {
 
 	klog.V(5).Infof("Deleting Namespace's NetworkPolicy entities")
 	for _, np := range nsInfo.networkPolicies {
-		delete(nsInfo.networkPolicies, np.name)
-		oc.destroyNetworkPolicy(np, nsInfo)
+		oc.checkAndSkipRetryPolicy(np.policy)
+		// add the full np object to the retry entry, since the namespace is going to be removed
+		// along with any mappings of nsInfo -> network policies
+		oc.initRetryPolicyWithDelete(np.policy, np)
+		isLastPolicyInNamespace := len(nsInfo.networkPolicies) == 1
+		if err := oc.destroyNetworkPolicy(np, isLastPolicyInNamespace); err != nil {
+			klog.Errorf("Failed to delete network policy: %s, error: %v", getPolicyNamespacedName(np.policy), err)
+			oc.unSkipRetryPolicy(np.policy)
+		} else {
+			oc.checkAndDeleteRetryPolicy(np.policy)
+			delete(nsInfo.networkPolicies, np.name)
+		}
 	}
 	oc.deleteGWRoutesForNamespace(ns.Name)
 	oc.multicastDeleteNamespace(ns, nsInfo)

go-controller/pkg/ovn/ovn.go

@@ -200,6 +200,14 @@ type Controller struct {
 	// channel to indicate we need to retry pods immediately
 	retryPodsChan chan struct{}
 
+	// Map of network policies that need to be retried, and the timestamp of when they last failed
+	// keyed by namespace/name
+	retryNetPolices map[string]*retryNetPolEntry
+	retryNetPolLock sync.Mutex
+
+	// channel to indicate we need to retry policy immediately
+	retryPolicyChan chan struct{}
+
 	metricsRecorder *metrics.ControlPlaneRecorder
 }
 
@@ -211,6 +219,16 @@ type retryEntry struct {
 	ignore bool
 }
 
+type retryNetPolEntry struct {
+	newPolicy  *kapisnetworking.NetworkPolicy
+	oldPolicy  *kapisnetworking.NetworkPolicy
+	np         *networkPolicy
+	timeStamp  time.Time
+	backoffSec time.Duration
+	// whether to include this NP in retry iterations
+	ignore bool
+}
+
 const (
 	// TCP is the constant string for the string "TCP"
 	TCP = "TCP"
@@ -283,6 +301,8 @@ func NewOvnController(ovnClient *util.OVNClientset, wf *factory.WatchFactory, st
 		joinSwIPManager:          nil,
 		retryPods:                make(map[types.UID]*retryEntry),
 		retryPodsChan:            make(chan struct{}, 1),
+		retryNetPolices:          make(map[string]*retryNetPolEntry),
+		retryPolicyChan:          make(chan struct{}, 1),
 		recorder:                 recorder,
 		nbClient:                 libovsdbOvnNBClient,
 		sbClient:                 libovsdbOvnSBClient,
@@ -741,23 +761,68 @@ func (oc *Controller) WatchPods() {
 // WatchNetworkPolicy starts the watching of network policy resource and calls
 // back the appropriate handler logic
 func (oc *Controller) WatchNetworkPolicy() {
+	go func() {
+		// track the retryNetworkPolicies map and every 30 seconds check if any pods need to be retried
+		for {
+			select {
+			case <-time.After(30 * time.Second):
+				oc.iterateRetryNetworkPolicies(false)
+			case <-oc.retryPolicyChan:
+				oc.iterateRetryNetworkPolicies(true)
+			case <-oc.stopChan:
+				return
+			}
+		}
+	}()
+
 	start := time.Now()
 	oc.watchFactory.AddPolicyHandler(cache.ResourceEventHandlerFuncs{
 		AddFunc: func(obj interface{}) {
 			policy := obj.(*kapisnetworking.NetworkPolicy)
-			oc.addNetworkPolicy(policy)
+			oc.initRetryPolicy(policy)
+			if err := oc.addNetworkPolicy(policy); err != nil {
+				klog.Errorf("Failed to create network policy %s, error: %v",
+					getPolicyNamespacedName(policy), err)
+				oc.unSkipRetryPolicy(policy)
+				return
+			}
+			oc.checkAndDeleteRetryPolicy(policy)
 		},
 		UpdateFunc: func(old, newer interface{}) {
 			oldPolicy := old.(*kapisnetworking.NetworkPolicy)
 			newPolicy := newer.(*kapisnetworking.NetworkPolicy)
 			if !reflect.DeepEqual(oldPolicy, newPolicy) {
-				oc.deleteNetworkPolicy(oldPolicy)
-				oc.addNetworkPolicy(newPolicy)
+				oc.checkAndSkipRetryPolicy(oldPolicy)
+				if err := oc.deleteNetworkPolicy(oldPolicy, nil); err != nil {
+					oc.initRetryPolicyWithDelete(oldPolicy, nil)
+					oc.initRetryPolicy(newPolicy)
+					oc.unSkipRetryPolicy(oldPolicy)
+					klog.Errorf("Failed to delete network policy %s, during update: %v",
+						getPolicyNamespacedName(oldPolicy), err)
+					return
+				}
+				// remove the old policy from retry entry since it was correctly deleted
+				oc.removeDeleteFromRetryPolicy(oldPolicy)
+				if err := oc.addNetworkPolicy(newPolicy); err != nil {
+					oc.initRetryPolicy(newPolicy)
+					oc.unSkipRetryPolicy(newPolicy)
+					klog.Errorf("Failed to create network policy %s, during update: %v",
+						getPolicyNamespacedName(newPolicy), err)
+					return
+				}
+				oc.checkAndDeleteRetryPolicy(newPolicy)
 			}
 		},
 		DeleteFunc: func(obj interface{}) {
 			policy := obj.(*kapisnetworking.NetworkPolicy)
-			oc.deleteNetworkPolicy(policy)
+			oc.checkAndSkipRetryPolicy(policy)
+			oc.initRetryPolicyWithDelete(policy, nil)
+			if err := oc.deleteNetworkPolicy(policy, nil); err != nil {
+				oc.unSkipRetryPolicy(policy)
+				klog.Errorf("Failed to delete network policy %s, error: %v", getPolicyNamespacedName(policy), err)
+				return
+			}
+			oc.checkAndDeleteRetryPolicy(policy)
 		},
 	}, oc.syncNetworkPolicies)
 	klog.Infof("Bootstrapping existing policies and cleaning stale policies took %v", time.Since(start))

go-controller/pkg/ovn/ovn_test.go

@@ -1,6 +1,7 @@
 package ovn
 
 import (
+	"context"
 	"github.com/onsi/gomega"
 	libovsdbclient "github.com/ovn-org/libovsdb/client"
 	"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/factory"
@@ -9,7 +10,6 @@ import (
 	libovsdbtest "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/testing/libovsdb"
 	"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/types"
 	util "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/util"
-
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/kubernetes/fake"
 	"k8s.io/client-go/tools/record"
@@ -106,3 +106,15 @@ func (o *FakeOVN) init() {
 	o.controller.multicastSupport = true
 	o.controller.loadBalancerGroupUUID = types.ClusterLBGroupName + "-UUID"
 }
+
+func (o *FakeOVN) resetNBClient(ctx context.Context) {
+	if o.controller.nbClient.Connected() {
+		o.controller.nbClient.Close()
+	}
+	gomega.Eventually(o.controller.nbClient.Connected()).Should(gomega.BeFalse())
+	err := o.controller.nbClient.Connect(ctx)
+	gomega.Expect(err).NotTo(gomega.HaveOccurred())
+	gomega.Eventually(o.controller.nbClient.Connected()).Should(gomega.BeTrue())
+	_, err = o.controller.nbClient.MonitorAll(ctx)
+	gomega.Expect(err).NotTo(gomega.HaveOccurred())
+}