Record an event when Egress IP assignment changes (antrea-io#5765)

Record an event when EgressIP is assigned to the Node interface or when
it is unassigned from the Node interface.

Signed-off-by: Pulkit Jain <[email protected]>

build/charts/antrea/templates/agent/clusterrole.yaml

@@ -219,3 +219,11 @@ rules:
     - get
     - list
     - watch
+  - apiGroups:
+      - ""
+    resources:
+      - events
+    verbs:
+      - create
+      - patch
+      - update

build/yamls/antrea-aks.yml

@@ -6283,6 +6283,14 @@ rules:
     - get
     - list
     - watch
+  - apiGroups:
+      - ""
+    resources:
+      - events
+    verbs:
+      - create
+      - patch
+      - update
 ---
 # Source: antrea/templates/antctl/clusterrole.yaml
 kind: ClusterRole

build/yamls/antrea-eks.yml

@@ -6283,6 +6283,14 @@ rules:
     - get
     - list
     - watch
+  - apiGroups:
+      - ""
+    resources:
+      - events
+    verbs:
+      - create
+      - patch
+      - update
 ---
 # Source: antrea/templates/antctl/clusterrole.yaml
 kind: ClusterRole

build/yamls/antrea-gke.yml

@@ -6283,6 +6283,14 @@ rules:
     - get
     - list
     - watch
+  - apiGroups:
+      - ""
+    resources:
+      - events
+    verbs:
+      - create
+      - patch
+      - update
 ---
 # Source: antrea/templates/antctl/clusterrole.yaml
 kind: ClusterRole

build/yamls/antrea-ipsec.yml

@@ -6296,6 +6296,14 @@ rules:
     - get
     - list
     - watch
+  - apiGroups:
+      - ""
+    resources:
+      - events
+    verbs:
+      - create
+      - patch
+      - update
 ---
 # Source: antrea/templates/antctl/clusterrole.yaml
 kind: ClusterRole

build/yamls/antrea.yml

@@ -6283,6 +6283,14 @@ rules:
     - get
     - list
     - watch
+  - apiGroups:
+      - ""
+    resources:
+      - events
+    verbs:
+      - create
+      - patch
+      - update
 ---
 # Source: antrea/templates/antctl/clusterrole.yaml
 kind: ClusterRole

cmd/antrea-agent/agent.go

@@ -516,7 +516,7 @@ func run(o *Options) error {
 	}
 	if o.enableEgress {
 		egressController, err = egress.NewEgressController(
-			ofClient, antreaClientProvider, crdClient, ifaceStore, routeClient, nodeConfig.Name, nodeConfig.NodeTransportInterfaceName,
+			ofClient, k8sClient, antreaClientProvider, crdClient, ifaceStore, routeClient, nodeConfig.Name, nodeConfig.NodeTransportInterfaceName,
 			memberlistCluster, egressInformer, nodeInformer, podUpdateChannel, serviceCIDRProvider, o.config.Egress.MaxEgressIPsPerNode,
 			features.DefaultFeatureGate.Enabled(features.EgressTrafficShaping),
 		)

pkg/agent/controller/egress/egress_controller.go

@@ -33,7 +33,10 @@ import (
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/apimachinery/pkg/watch"
 	coreinformers "k8s.io/client-go/informers/core/v1"
+	"k8s.io/client-go/kubernetes"
+	v1 "k8s.io/client-go/kubernetes/typed/core/v1"
 	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/tools/record"
 	"k8s.io/client-go/util/retry"
 	"k8s.io/client-go/util/workqueue"
 	"k8s.io/klog/v2"
@@ -49,6 +52,7 @@ import (
 	cpv1b2 "antrea.io/antrea/pkg/apis/controlplane/v1beta2"
 	crdv1b1 "antrea.io/antrea/pkg/apis/crd/v1beta1"
 	clientsetversioned "antrea.io/antrea/pkg/client/clientset/versioned"
+	"antrea.io/antrea/pkg/client/clientset/versioned/scheme"
 	crdinformers "antrea.io/antrea/pkg/client/informers/externalversions/crd/v1beta1"
 	crdlisters "antrea.io/antrea/pkg/client/listers/crd/v1beta1"
 	"antrea.io/antrea/pkg/controller/metrics"
@@ -136,6 +140,7 @@ type egressBinding struct {
 type EgressController struct {
 	ofClient             openflow.Client
 	routeClient          route.Interface
+	k8sClient            kubernetes.Interface
 	crdClient            clientsetversioned.Interface
 	antreaClientProvider agent.AntreaClientProvider
 
@@ -175,10 +180,14 @@ type EgressController struct {
 	serviceCIDRUpdateRetryDelay time.Duration
 
 	trafficShapingEnabled bool
+
+	eventBroadcaster record.EventBroadcaster
+	record           record.EventRecorder
 }
 
 func NewEgressController(
 	ofClient openflow.Client,
+	k8sClient kubernetes.Interface,
 	antreaClientGetter agent.AntreaClientProvider,
 	crdClient clientsetversioned.Interface,
 	ifaceStore interfacestore.InterfaceStore,
@@ -196,9 +205,17 @@ func NewEgressController(
 	if trafficShapingEnabled && !openflow.OVSMetersAreSupported() {
 		klog.Info("EgressTrafficShaping feature gate is enabled, but it is ignored because OVS meters are not supported.")
 	}
+
+	eventBroadcaster := record.NewBroadcaster()
+	recorder := eventBroadcaster.NewRecorder(
+		scheme.Scheme,
+		corev1.EventSource{Component: controllerName},
+	)
+
 	c := &EgressController{
 		ofClient:             ofClient,
 		routeClient:          routeClient,
+		k8sClient:            k8sClient,
 		antreaClientProvider: antreaClientGetter,
 		crdClient:            crdClient,
 		queue:                workqueue.NewNamedRateLimitingQueue(workqueue.NewItemExponentialFailureRateLimiter(minRetryDelay, maxRetryDelay), "egressgroup"),
@@ -220,6 +237,9 @@ func NewEgressController(
 		serviceCIDRUpdateRetryDelay: 10 * time.Second,
 
 		trafficShapingEnabled: openflow.OVSMetersAreSupported() && trafficShapingEnabled,
+
+		eventBroadcaster: eventBroadcaster,
+		record:           recorder,
 	}
 	ipAssigner, err := newIPAssigner(nodeTransportInterface, egressDummyDevice)
 	if err != nil {
@@ -388,6 +408,12 @@ func (c *EgressController) Run(stopCh <-chan struct{}) {
 	klog.Infof("Starting %s", controllerName)
 	defer klog.Infof("Shutting down %s", controllerName)
 
+	c.eventBroadcaster.StartStructuredLogging(0)
+	c.eventBroadcaster.StartRecordingToSink(&v1.EventSinkImpl{
+		Interface: c.k8sClient.CoreV1().Events(""),
+	})
+	defer c.eventBroadcaster.Shutdown()
+
 	go c.localIPDetector.Run(stopCh)
 	go c.egressIPScheduler.Run(stopCh)
 	go c.ipAssigner.Run(stopCh)
@@ -848,14 +874,22 @@ func (c *EgressController) syncEgress(egressName string) error {
 		// Ensure the Egress IP is assigned to the system. Force advertising the IP if it was previously assigned to
 		// another Node in the Egress API. This could force refreshing other peers' neighbor cache when the Egress IP is
 		// obtained by this Node and another Node at the same time in some situations, e.g. split brain.
-		if err := c.ipAssigner.AssignIP(desiredEgressIP, egress.Status.EgressNode != c.nodeName); err != nil {
+		assigned, err := c.ipAssigner.AssignIP(desiredEgressIP, egress.Status.EgressNode != c.nodeName)
+		if err != nil {
 			return err
 		}
+		if assigned {
+			c.record.Eventf(egress, corev1.EventTypeNormal, "IPAssigned", "Assigned Egress %s with IP %s on Node %s", egress.Name, desiredEgressIP, desiredNode)
+		}
 	} else {
 		// Unassign the Egress IP from the local Node if it was assigned by the agent.
-		if err := c.ipAssigner.UnassignIP(desiredEgressIP); err != nil {
+		unassigned, err := c.ipAssigner.UnassignIP(desiredEgressIP)
+		if err != nil {
 			return err
 		}
+		if unassigned {
+			c.record.Eventf(egress, corev1.EventTypeNormal, "IPUnassigned", "Unassigned Egress %s with IP %s from Node %s", egress.Name, desiredEgressIP, c.nodeName)
+		}
 	}
 
 	// Realize the latest EgressIP and get the desired mark.
@@ -951,7 +985,7 @@ func (c *EgressController) uninstallEgress(egressName string, eState *egressStat
 		}
 	}
 	// Unassign the Egress IP from the local Node if it was assigned by the agent.
-	if err := c.ipAssigner.UnassignIP(eState.egressIP); err != nil {
+	if _, err := c.ipAssigner.UnassignIP(eState.egressIP); err != nil {
 		return err
 	}
 	// Remove the Egress's state.

pkg/agent/controller/egress/egress_controller_test.go

@@ -182,6 +182,7 @@ func newFakeController(t *testing.T, initObjects []runtime.Object) *fakeControll
 	mockServiceCIDRProvider := servicecidrtest.NewMockInterface(controller)
 	mockServiceCIDRProvider.EXPECT().AddEventHandler(gomock.Any())
 	egressController, _ := NewEgressController(mockOFClient,
+		k8sClient,
 		&antreaClientGetter{clientset},
 		crdClient,
 		ifaceStore,

pkg/agent/controller/serviceexternalip/controller.go

@@ -393,7 +393,7 @@ func (c *ServiceExternalIPController) assignIP(ip string, service apimachineryty
 	c.assignedIPsMutex.Lock()
 	defer c.assignedIPsMutex.Unlock()
 	if _, ok := c.assignedIPs[ip]; !ok {
-		if err := c.ipAssigner.AssignIP(ip, true); err != nil {
+		if _, err := c.ipAssigner.AssignIP(ip, true); err != nil {
 			return err
 		}
 		c.assignedIPs[ip] = sets.New[string](service.String())
@@ -411,7 +411,7 @@ func (c *ServiceExternalIPController) unassignIP(ip string, service apimachinery
 		return nil
 	}
 	if assigned.Len() == 1 && assigned.Has(service.String()) {
-		if err := c.ipAssigner.UnassignIP(ip); err != nil {
+		if _, err := c.ipAssigner.UnassignIP(ip); err != nil {
 			return err
 		}
 		delete(c.assignedIPs, ip)

pkg/agent/ipassigner/ip_assigner.go

@@ -19,9 +19,14 @@ import "k8s.io/apimachinery/pkg/util/sets"
 // IPAssigner provides methods to assign or unassign IP.
 type IPAssigner interface {
 	// AssignIP ensures the provided IP is assigned to the system.
-	AssignIP(ip string, forceAdvertise bool) error
+	// It returns true only in the case when there is no error and the IP provided
+	// was not assigned to the interface before the operation, in all other cases it
+	// returns false.
+	AssignIP(ip string, forceAdvertise bool) (bool, error)
 	// UnassignIP ensures the provided IP is not assigned to the system.
-	UnassignIP(ip string) error
+	// It returns true only in the case when there is no error and the IP provided
+	// was assigned to the interface before the operation.
+	UnassignIP(ip string) (bool, error)
 	// AssignedIPs return the IPs that are assigned to the system by this IPAssigner.
 	AssignedIPs() sets.Set[string]
 	// InitIPs ensures the IPs that are assigned to the system match the given IPs.

pkg/agent/ipassigner/ip_assigner_linux.go

@@ -143,10 +143,10 @@ func (a *ipAssigner) loadIPAddresses() (sets.Set[string], error) {
 }
 
 // AssignIP ensures the provided IP is assigned to the dummy device and the ARP/NDP responders.
-func (a *ipAssigner) AssignIP(ip string, forceAdvertise bool) error {
+func (a *ipAssigner) AssignIP(ip string, forceAdvertise bool) (bool, error) {
 	parsedIP := net.ParseIP(ip)
 	if parsedIP == nil {
-		return fmt.Errorf("invalid IP %s", ip)
+		return false, fmt.Errorf("invalid IP %s", ip)
 	}
 	a.mutex.Lock()
 	defer a.mutex.Unlock()
@@ -156,14 +156,14 @@ func (a *ipAssigner) AssignIP(ip string, forceAdvertise bool) error {
 		if forceAdvertise {
 			a.advertise(parsedIP)
 		}
-		return nil
+		return false, nil
 	}
 
 	if a.dummyDevice != nil {
 		addr := util.NewIPNet(parsedIP)
 		if err := netlink.AddrAdd(a.dummyDevice, &netlink.Addr{IPNet: addr}); err != nil {
 			if !errors.Is(err, unix.EEXIST) {
-				return fmt.Errorf("failed to add IP %v to interface %s: %v", ip, a.dummyDevice.Attrs().Name, err)
+				return false, fmt.Errorf("failed to add IP %v to interface %s: %v", ip, a.dummyDevice.Attrs().Name, err)
 			} else {
 				klog.InfoS("IP was already assigned to interface", "ip", parsedIP, "interface", a.dummyDevice.Attrs().Name)
 			}
@@ -174,18 +174,18 @@ func (a *ipAssigner) AssignIP(ip string, forceAdvertise bool) error {
 
 	if utilnet.IsIPv4(parsedIP) && a.arpResponder != nil {
 		if err := a.arpResponder.AddIP(parsedIP); err != nil {
-			return fmt.Errorf("failed to assign IP %v to ARP responder: %v", ip, err)
+			return false, fmt.Errorf("failed to assign IP %v to ARP responder: %v", ip, err)
 		}
 	}
 	if utilnet.IsIPv6(parsedIP) && a.ndpResponder != nil {
 		if err := a.ndpResponder.AddIP(parsedIP); err != nil {
-			return fmt.Errorf("failed to assign IP %v to NDP responder: %v", ip, err)
+			return false, fmt.Errorf("failed to assign IP %v to NDP responder: %v", ip, err)
 		}
 	}
 	// Always advertise the IP when the IP is newly assigned to this Node.
 	a.advertise(parsedIP)
 	a.assignedIPs.Insert(ip)
-	return nil
+	return true, nil
 }
 
 func (a *ipAssigner) advertise(ip net.IP) {
@@ -203,24 +203,24 @@ func (a *ipAssigner) advertise(ip net.IP) {
 }
 
 // UnassignIP ensures the provided IP is not assigned to the dummy device.
-func (a *ipAssigner) UnassignIP(ip string) error {
+func (a *ipAssigner) UnassignIP(ip string) (bool, error) {
 	parsedIP := net.ParseIP(ip)
 	if parsedIP == nil {
-		return fmt.Errorf("invalid IP %s", ip)
+		return false, fmt.Errorf("invalid IP %s", ip)
 	}
 	a.mutex.Lock()
 	defer a.mutex.Unlock()
 
 	if !a.assignedIPs.Has(ip) {
 		klog.V(2).InfoS("The IP is not assigned", "ip", ip)
-		return nil
+		return false, nil
 	}
 
 	if a.dummyDevice != nil {
 		addr := util.NewIPNet(parsedIP)
 		if err := netlink.AddrDel(a.dummyDevice, &netlink.Addr{IPNet: addr}); err != nil {
 			if !errors.Is(err, unix.EADDRNOTAVAIL) {
-				return fmt.Errorf("failed to delete IP %v from interface %s: %v", ip, a.dummyDevice.Attrs().Name, err)
+				return false, fmt.Errorf("failed to delete IP %v from interface %s: %v", ip, a.dummyDevice.Attrs().Name, err)
 			} else {
 				klog.InfoS("IP does not exist on interface", "ip", parsedIP, "interface", a.dummyDevice.Attrs().Name)
 			}
@@ -230,17 +230,17 @@ func (a *ipAssigner) UnassignIP(ip string) error {
 
 	if utilnet.IsIPv4(parsedIP) && a.arpResponder != nil {
 		if err := a.arpResponder.RemoveIP(parsedIP); err != nil {
-			return fmt.Errorf("failed to remove IP %v from ARP responder: %v", ip, err)
+			return false, fmt.Errorf("failed to remove IP %v from ARP responder: %v", ip, err)
 		}
 	}
 	if utilnet.IsIPv6(parsedIP) && a.ndpResponder != nil {
 		if err := a.ndpResponder.RemoveIP(parsedIP); err != nil {
-			return fmt.Errorf("failed to remove IP %v from NDP responder: %v", ip, err)
+			return false, fmt.Errorf("failed to remove IP %v from NDP responder: %v", ip, err)
 		}
 	}
 
 	a.assignedIPs.Delete(ip)
-	return nil
+	return true, nil
 }
 
 // AssignedIPs return the IPs that are assigned to the dummy device.