Add distributed status management for EgressFirewall.

Add Status.Messages field to record statuses from zones,
make egressfirewall status a subresource.

Signed-off-by: Nadia Pinaeva <[email protected]>

dist/images/ovnkube.sh

@@ -1952,6 +1952,12 @@ ovn-cluster-manager() {
   fi
   echo "egressservice_enabled_flag=${egressservice_enabled_flag}"
 
+  egressfirewall_enabled_flag=
+  if [[ ${ovn_egressfirewall_enable} == "true" ]]; then
+	  egressfirewall_enabled_flag="--enable-egress-firewall"
+  fi
+  echo "egressfirewall_enabled_flag=${egressfirewall_enabled_flag}"
+
   hybrid_overlay_flags=
   if [[ ${ovn_hybrid_overlay_enable} == "true" ]]; then
     hybrid_overlay_flags="--enable-hybrid-overlay"
@@ -2029,6 +2035,7 @@ ovn-cluster-manager() {
 
   echo "=============== ovn-cluster-manager ========== MASTER ONLY"
   /usr/bin/ovnkube --init-cluster-manager ${K8S_NODE} \
+    ${egressfirewall_enabled_flag} \
     ${egressip_enabled_flag} \
     ${egressip_healthcheck_port_flag} \
     ${egressservice_enabled_flag} \

dist/templates/k8s.ovn.org_egressfirewalls.yaml.j2

@@ -3,8 +3,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   annotations:
-    controller-gen.kubebuilder.io/version: v0.11.3
-  creationTimestamp: null
+    controller-gen.kubebuilder.io/version: v0.13.0
   name: egressfirewalls.k8s.ovn.org
 spec:
   group: k8s.ovn.org
@@ -157,12 +156,20 @@ spec:
           status:
             description: Observed status of EgressFirewall
             properties:
+              messages:
+                items:
+                  type: string
+                type: array
+                x-kubernetes-list-type: set
               status:
                 type: string
+            required:
+            - messages
             type: object
         required:
         - spec
         type: object
     served: true
     storage: true
-    subresources: {}
+    subresources:
+      status: {}

dist/templates/rbac-ovnkube-cluster-manager.yaml.j2

@@ -61,6 +61,7 @@ rules:
           - egressips
           - egressservices
           - adminpolicybasedexternalroutes
+          - egressfirewalls
       verbs: [ "get", "list", "watch" ]
     - apiGroups: ["k8s.ovn.org"]
       resources:
@@ -80,4 +81,5 @@ rules:
     - apiGroups: ["k8s.ovn.org"]
       resources:
         - adminpolicybasedexternalroutes/status
+        - egressfirewalls/status
       verbs: [ "patch", "update" ]

dist/templates/rbac-ovnkube-master.yaml.j2

@@ -96,7 +96,7 @@ rules:
       verbs: [ "patch", "update" ]
     - apiGroups: ["k8s.ovn.org"]
       resources:
-          - egressfirewalls
+          - egressfirewalls/status
           - egressips
           - egressqoses
           - egressservices/status

dist/templates/rbac-ovnkube-node.yaml.j2

@@ -151,7 +151,7 @@ rules:
       verbs: ["list", "get", "watch"]
     - apiGroups: ["k8s.ovn.org"]
       resources:
-          - egressfirewalls
+          - egressfirewalls/status
           - adminpolicybasedexternalroutes/status
       verbs: [ "patch", "update" ]
     - apiGroups: ["policy.networking.k8s.io"]

go-controller/pkg/clustermanager/status_manager/resource_manager.go

@@ -9,6 +9,9 @@ import (
 	adminpolicybasedrouteapi "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/crd/adminpolicybasedroute/v1"
 	adminpolicybasedrouteclientset "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/crd/adminpolicybasedroute/v1/apis/clientset/versioned"
 	adminpolicybasedroutelisters "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/crd/adminpolicybasedroute/v1/apis/listers/adminpolicybasedroute/v1"
+	egressfirewallapi "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/crd/egressfirewall/v1"
+	egressfirewallclientset "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/crd/egressfirewall/v1/apis/clientset/versioned"
+	egressfirewalllisters "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/crd/egressfirewall/v1/apis/listers/egressfirewall/v1"
 	"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/types"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -66,3 +69,43 @@ func (m *apbRouteManager) updateStatus(route *adminpolicybasedrouteapi.AdminPoli
 	return []byte(fmt.Sprintf(`{"apiVersion":"k8s.ovn.org/adminpolicybasedroutelisters","kind":"AdminPolicyBasedExternalRoute","status":{"status":"%s"}}`, newStatus)),
 		route.Status.Status != newStatus
 }
+
+type egressFirewallManager struct {
+	lister egressfirewalllisters.EgressFirewallLister
+	client egressfirewallclientset.Interface
+}
+
+func newEgressFirewallManager(lister egressfirewalllisters.EgressFirewallLister, client egressfirewallclientset.Interface) *egressFirewallManager {
+	return &egressFirewallManager{
+		lister: lister,
+		client: client,
+	}
+}
+
+func (m *egressFirewallManager) get(namespace, name string) (*egressfirewallapi.EgressFirewall, error) {
+	return m.lister.EgressFirewalls(namespace).Get(name)
+}
+
+func (m *egressFirewallManager) patch(ctx context.Context, namespace, name string, pt ktypes.PatchType, data []byte, opts metav1.PatchOptions,
+	subresources ...string) (*egressfirewallapi.EgressFirewall, error) {
+	return m.client.K8sV1().EgressFirewalls(namespace).Patch(ctx, name, pt, data, opts, subresources...)
+}
+
+func (m *egressFirewallManager) statusChanged(oldObj, newObj *egressfirewallapi.EgressFirewall) bool {
+	return !reflect.DeepEqual(oldObj.Status.Messages, newObj.Status.Messages)
+}
+
+func (m *egressFirewallManager) updateStatus(egressFirewall *egressfirewallapi.EgressFirewall) ([]byte, bool) {
+	if egressFirewall == nil || len(egressFirewall.Status.Messages) == 0 {
+		return nil, false
+	}
+	newStatus := "EgressFirewall Rules applied"
+	for _, message := range egressFirewall.Status.Messages {
+		if strings.Contains(message, types.EgressFirewallApplyError) {
+			newStatus = types.EgressFirewallApplyError
+			break
+		}
+	}
+	return []byte(fmt.Sprintf(`{"apiVersion":"k8s.ovn.org/v1","kind":"EgressFirewall","status":{"status":"%s"}}`, newStatus)),
+		egressFirewall.Status.Status != newStatus
+}

go-controller/pkg/clustermanager/status_manager/status_manager.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/config"
 	adminpolicybasedrouteapi "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/crd/adminpolicybasedroute/v1"
+	egressfirewallapi "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/crd/egressfirewall/v1"
 	"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/factory"
 	"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/util"
 
@@ -170,6 +171,14 @@ func NewStatusManager(wf *factory.WatchFactory, ovnClient *util.OVNClusterManage
 		)
 		sm.typedManagers["adminpolicybasedexternalroutes"] = apbRouteManager
 	}
+	if config.OVNKubernetesFeature.EnableEgressFirewall {
+		egressFirewallManager := newStatusManager[*egressfirewallapi.EgressFirewall](
+			"egressfirewalls",
+			wf.EgressFirewallInformer().Informer(),
+			newEgressFirewallManager(wf.EgressFirewallInformer().Lister(), ovnClient.EgressFirewallClient),
+		)
+		sm.typedManagers["egressfirewalls"] = egressFirewallManager
+	}
 	return sm
 }
 

go-controller/pkg/crd/egressfirewall/v1/types.go

@@ -17,6 +17,7 @@ const (
 // +resource:path=egressfirewall
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
 // +kubebuilder:printcolumn:name="EgressFirewall Status",type=string,JSONPath=".status.status"
+// +kubebuilder:subresource:status
 // EgressFirewall describes the current egress firewall for a Namespace.
 // Traffic from a pod to an IP address outside the cluster will be checked against
 // each EgressFirewallRule in the pod's namespace's EgressFirewall, in
@@ -35,6 +36,9 @@ type EgressFirewall struct {
 
 type EgressFirewallStatus struct {
 	Status string `json:"status,omitempty"`
+	// +patchStrategy=merge
+	// +listType=set
+	Messages []string `json:"messages"`
 }
 
 // EgressFirewallSpec is a desired state description of EgressFirewall.

go-controller/pkg/factory/factory.go

@@ -16,6 +16,7 @@ import (
 	egressfirewallapi "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/crd/egressfirewall/v1"
 	egressfirewallscheme "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/crd/egressfirewall/v1/apis/clientset/versioned/scheme"
 	egressfirewallinformerfactory "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/crd/egressfirewall/v1/apis/informers/externalversions"
+	egressfirewallinformer "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/crd/egressfirewall/v1/apis/informers/externalversions/egressfirewall/v1"
 	egressfirewalllister "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/crd/egressfirewall/v1/apis/listers/egressfirewall/v1"
 	"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/types"
 	"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/util"
@@ -546,6 +547,7 @@ func NewNodeWatchFactory(ovnClientset *util.OVNNodeClientset, nodeName string) (
 func NewClusterManagerWatchFactory(ovnClientset *util.OVNClusterManagerClientset) (*WatchFactory, error) {
 	wf := &WatchFactory{
 		iFactory:             informerfactory.NewSharedInformerFactory(ovnClientset.KubeClient, resyncInterval),
+		efFactory:            egressfirewallinformerfactory.NewSharedInformerFactory(ovnClientset.EgressFirewallClient, resyncInterval),
 		eipFactory:           egressipinformerfactory.NewSharedInformerFactory(ovnClientset.EgressIPClient, resyncInterval),
 		cpipcFactory:         ocpcloudnetworkinformerfactory.NewSharedInformerFactory(ovnClientset.CloudNetworkClient, resyncInterval),
 		egressServiceFactory: egressserviceinformerfactory.NewSharedInformerFactoryWithOptions(ovnClientset.EgressServiceClient, resyncInterval),
@@ -631,6 +633,11 @@ func NewClusterManagerWatchFactory(ovnClientset *util.OVNClusterManagerClientset
 		wf.apbRouteFactory.K8s().V1().AdminPolicyBasedExternalRoutes().Informer()
 	}
 
+	if config.OVNKubernetesFeature.EnableEgressFirewall {
+		// make sure shared informer is created for a factory, so on wf.efFactory.Start() it is initialized and caches are synced.
+		wf.efFactory.K8s().V1().EgressFirewalls().Informer()
+	}
+
 	return wf, nil
 }
 
@@ -1243,6 +1250,10 @@ func (wf *WatchFactory) EgressIPInformer() egressipinformer.EgressIPInformer {
 	return wf.eipFactory.K8s().V1().EgressIPs()
 }
 
+func (wf *WatchFactory) EgressFirewallInformer() egressfirewallinformer.EgressFirewallInformer {
+	return wf.efFactory.K8s().V1().EgressFirewalls()
+}
+
 // withServiceNameAndNoHeadlessServiceSelector returns a LabelSelector (added to the
 // watcher for EndpointSlices) that will only choose EndpointSlices with a non-empty
 // "kubernetes.io/service-name" label and without "service.kubernetes.io/headless"

go-controller/pkg/factory/factory_test.go

@@ -355,10 +355,11 @@ var _ = Describe("Watch Factory Operations", func() {
 			EgressServiceClient:  egressServiceFakeClient,
 		}
 		ovnCMClientset = &util.OVNClusterManagerClientset{
-			KubeClient:          fakeClient,
-			EgressIPClient:      egressIPFakeClient,
-			CloudNetworkClient:  cloudNetworkFakeClient,
-			EgressServiceClient: egressServiceFakeClient,
+			KubeClient:           fakeClient,
+			EgressIPClient:       egressIPFakeClient,
+			CloudNetworkClient:   cloudNetworkFakeClient,
+			EgressServiceClient:  egressServiceFakeClient,
+			EgressFirewallClient: egressFirewallFakeClient,
 		}
 
 		pods = make([]*v1.Pod, 0)

go-controller/pkg/ovn/default_network_controller.go

@@ -768,16 +768,9 @@ func (h *defaultNetworkControllerEventHandler) AddResource(obj interface{}, from
 		}
 
 	case factory.EgressFirewallType:
-		var err error
 		egressFirewall := obj.(*egressfirewall.EgressFirewall).DeepCopy()
-		if err = h.oc.addEgressFirewall(egressFirewall); err != nil {
-			egressFirewall.Status.Status = egressFirewallAddError
-		} else {
-			egressFirewall.Status.Status = egressFirewallAppliedCorrectly
-			metrics.UpdateEgressFirewallRuleCount(float64(len(egressFirewall.Spec.Egress)))
-			metrics.IncrementEgressFirewallCount()
-		}
-		if statusErr := h.oc.updateEgressFirewallStatusWithRetry(egressFirewall); statusErr != nil {
+		err := h.oc.addEgressFirewall(egressFirewall)
+		if statusErr := h.oc.setEgressFirewallStatus(egressFirewall, err); statusErr != nil {
 			klog.Errorf("Failed to update egress firewall status %s, error: %v",
 				getEgressFirewallNamespacedName(egressFirewall), statusErr)
 		}

go-controller/pkg/ovn/egressfirewall.go

@@ -1,6 +1,7 @@
 package ovn
 
 import (
+	"context"
 	"fmt"
 	"net"
 	"strconv"
@@ -12,6 +13,7 @@ import (
 	egressfirewallapi "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/crd/egressfirewall/v1"
 	libovsdbops "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/libovsdb/ops"
 	libovsdbutil "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/libovsdb/util"
+	"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/metrics"
 	"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/nbdb"
 	"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/types"
 	"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/util"
@@ -20,16 +22,16 @@ import (
 	kapi "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
+	ktypes "k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/apimachinery/pkg/util/sets"
-	"k8s.io/client-go/util/retry"
 	"k8s.io/klog/v2"
 	utilnet "k8s.io/utils/net"
+	utilpointer "k8s.io/utils/pointer"
 )
 
 const (
 	egressFirewallAppliedCorrectly = "EgressFirewall Rules applied"
-	egressFirewallAddError         = "EgressFirewall Rules not correctly added"
 	aclDeleteBatchSize             = 1000
 )
 
@@ -205,6 +207,7 @@ func (oc *DefaultNetworkController) addEgressFirewall(egressFirewall *egressfire
 	defer ef.Unlock()
 	// egressFirewall may already exist, if previous add failed, cleanup
 	if _, loaded := oc.egressFirewalls.Load(egressFirewall.Namespace); loaded {
+		klog.Infof("Egress firewall in namespace 5s already exists, cleanup", egressFirewall.Namespace)
 		err := oc.deleteEgressFirewall(egressFirewall)
 		if err != nil {
 			return fmt.Errorf("failed to cleanup existing egress firewall %s on add: %v", egressFirewall.Namespace, err)
@@ -215,8 +218,8 @@ func (oc *DefaultNetworkController) addEgressFirewall(egressFirewall *egressfire
 	for i, egressFirewallRule := range egressFirewall.Spec.Egress {
 		// process Rules into egressFirewallRules for egressFirewall struct
 		if i > types.EgressFirewallStartPriority-types.MinimumReservedEgressFirewallPriority {
-			klog.Warningf("egressFirewall for namespace %s has too many rules, the rest will be ignored",
-				egressFirewall.Namespace)
+			errorList = append(errorList, fmt.Errorf("egressFirewall for namespace %s has too many rules, max allowed number is %v",
+				egressFirewall.Namespace, types.EgressFirewallStartPriority-types.MinimumReservedEgressFirewallPriority))
 			break
 		}
 		efr, err := oc.newEgressFirewallRule(egressFirewallRule, i)
@@ -287,23 +290,6 @@ func (oc *DefaultNetworkController) deleteEgressFirewall(egressFirewallObj *egre
 	return nil
 }
 
-func (oc *DefaultNetworkController) updateEgressFirewallStatusWithRetry(egressFirewall *egressfirewallapi.EgressFirewall) error {
-	retryErr := retry.RetryOnConflict(retry.DefaultRetry, func() error {
-		ef, err := oc.watchFactory.GetEgressFirewall(egressFirewall.Namespace, egressFirewall.Name)
-		if err != nil {
-			return err
-		}
-		c := ef.DeepCopy()
-		c.Status.Status = egressFirewall.Status.Status
-		return oc.kube.UpdateEgressFirewall(c)
-	})
-	if retryErr != nil {
-		return fmt.Errorf("error in updating status on EgressFirewall %s/%s: %v",
-			egressFirewall.Namespace, egressFirewall.Name, retryErr)
-	}
-	return nil
-}
-
 func (oc *DefaultNetworkController) addEgressFirewallRules(ef *egressFirewall, hashedAddressSetNameIPv4,
 	hashedAddressSetNameIPv6 string, aclLogging *libovsdbutil.ACLLoggingLevels, ruleIDs ...int) error {
 	var ops []libovsdb.Operation
@@ -766,3 +752,36 @@ func (oc *DefaultNetworkController) updateEgressFirewallForNode(oldNode, newNode
 
 	return efErr
 }
+
+func (oc *DefaultNetworkController) setEgressFirewallStatus(egressFirewall *egressfirewallapi.EgressFirewall, handlerErr error) error {
+	newMsg := oc.zone + ": "
+	if handlerErr != nil {
+		newMsg += types.EgressFirewallApplyError + ": " + handlerErr.Error()
+	} else {
+		newMsg += egressFirewallAppliedCorrectly
+		metrics.UpdateEgressFirewallRuleCount(float64(len(egressFirewall.Spec.Egress)))
+		metrics.IncrementEgressFirewallCount()
+	}
+	needsUpdate := true
+	for _, message := range egressFirewall.Status.Messages {
+		if message == newMsg {
+			// found previous status
+			needsUpdate = false
+			break
+		}
+	}
+	if !needsUpdate {
+		return nil
+	}
+
+	patchOptions := metav1.PatchOptions{
+		Force:        utilpointer.Bool(true),
+		FieldManager: oc.zone,
+	}
+
+	patch := []byte(fmt.Sprintf(`{"apiVersion":"k8s.ovn.org/v1","kind":"EgressFirewall","status":{"messages":["%s"]}}`, newMsg))
+	_, err := oc.kube.EgressFirewallClient.K8sV1().EgressFirewalls(egressFirewall.Namespace).Patch(context.TODO(), egressFirewall.Name,
+		ktypes.ApplyPatchType, patch, patchOptions, "status")
+
+	return err
+}

go-controller/pkg/types/resource_status.go

@@ -2,5 +2,6 @@ package types
 
 // this file defines error messages that are used to figure out if a resource reconciliation failed
 const (
-	APBRouteErrorMsg = "failed to apply policy"
+	APBRouteErrorMsg         = "failed to apply policy"
+	EgressFirewallApplyError = "EgressFirewall Rules not correctly applied"
 )

go-controller/pkg/util/kube.go

@@ -103,6 +103,7 @@ type OVNClusterManagerClientset struct {
 	NetworkAttchDefClient  networkattchmentdefclientset.Interface
 	EgressServiceClient    egressserviceclientset.Interface
 	AdminPolicyRouteClient adminpolicybasedrouteclientset.Interface
+	EgressFirewallClient   egressfirewallclientset.Interface
 }
 
 const (
@@ -163,6 +164,7 @@ func (cs *OVNClientset) GetClusterManagerClientset() *OVNClusterManagerClientset
 		NetworkAttchDefClient:  cs.NetworkAttchDefClient,
 		EgressServiceClient:    cs.EgressServiceClient,
 		AdminPolicyRouteClient: cs.AdminPolicyRouteClient,
+		EgressFirewallClient:   cs.EgressFirewallClient,
 	}
 }