Delete secondaryNetwork OVS ports correctly after an Agent restart

cmd/antrea-agent/agent.go

@@ -646,6 +646,19 @@ func run(o *Options) error {
 		}
 	}
 
+	// Secondary network Controller should be created before CNIServer.Run() to make sure no Pod CNI updates will be missed.
+	var secondaryNetworkController *secondarynetwork.Controller
+	if features.DefaultFeatureGate.Enabled(features.SecondaryNetwork) {
+		secondaryNetworkController, err = secondarynetwork.NewController(
+			o.config.ClientConnection, o.config.KubeAPIServerOverride,
+			k8sClient, localPodInformer.Get(),
+			podUpdateChannel, ifaceStore,
+			&o.config.SecondaryNetwork, ovsdbConnection)
+		if err != nil {
+			return fmt.Errorf("failed to create secondary network controller: %w", err)
+		}
+	}
+
 	var traceflowController *traceflow.Controller
 	if features.DefaultFeatureGate.Enabled(features.Traceflow) {
 		traceflowController = traceflow.NewTraceflowController(
@@ -760,18 +773,6 @@ func run(o *Options) error {
 		go ipamController.Run(stopCh)
 	}
 
-	var secondaryNetworkController *secondarynetwork.Controller
-	if features.DefaultFeatureGate.Enabled(features.SecondaryNetwork) {
-		secondaryNetworkController, err = secondarynetwork.NewController(
-			o.config.ClientConnection, o.config.KubeAPIServerOverride,
-			k8sClient, localPodInformer.Get(),
-			podUpdateChannel,
-			&o.config.SecondaryNetwork, ovsdbConnection)
-		if err != nil {
-			return fmt.Errorf("failed to create secondary network controller: %w", err)
-		}
-	}
-
 	var bgpController *bgp.Controller
 	if features.DefaultFeatureGate.Enabled(features.BGPPolicy) {
 		bgpPolicyInformer := crdInformerFactory.Crd().V1alpha1().BGPPolicies()

pkg/agent/secondarynetwork/init.go

@@ -24,6 +24,7 @@ import (
 	componentbaseconfig "k8s.io/component-base/config"
 	"k8s.io/klog/v2"
 
+	"antrea.io/antrea/pkg/agent/interfacestore"
 	"antrea.io/antrea/pkg/agent/secondarynetwork/podwatch"
 	agentconfig "antrea.io/antrea/pkg/config/agent"
 	"antrea.io/antrea/pkg/ovs/ovsconfig"
@@ -47,6 +48,7 @@ func NewController(
 	k8sClient clientset.Interface,
 	podInformer cache.SharedIndexInformer,
 	podUpdateSubscriber channel.Subscriber,
+	pIfaceStore interfacestore.InterfaceStore,
 	secNetConfig *agentconfig.SecondaryNetworkConfig, ovsdb *ovsdb.OVSDB,
 ) (*Controller, error) {
 	ovsBridgeClient, err := createOVSBridge(secNetConfig.OVSBridges, ovsdb)
@@ -65,7 +67,7 @@ func NewController(
 	// k8s.v1.cni.cncf.io/networks Annotation defined.
 	podWatchController, err := podwatch.NewPodController(
 		k8sClient, netAttachDefClient, podInformer,
-		podUpdateSubscriber, ovsBridgeClient)
+		podUpdateSubscriber, pIfaceStore, ovsBridgeClient)
 	if err != nil {
 		return nil, err
 	}

pkg/agent/secondarynetwork/podwatch/controller.go

@@ -102,6 +102,7 @@ func NewPodController(
 	netAttachDefClient netdefclient.K8sCniCncfIoV1Interface,
 	podInformer cache.SharedIndexInformer,
 	podUpdateSubscriber channel.Subscriber,
+	pIfaceStore interfacestore.InterfaceStore,
 	ovsBridgeClient ovsconfig.OVSBridgeClient,
 ) (*PodController, error) {
 	ifaceStore := interfacestore.NewInterfaceStore()
@@ -133,6 +134,18 @@ func NewPodController(
 		},
 		resyncPeriod,
 	)
+
+	err = pc.initializeSecondaryInterfaceStore()
+	if err != nil {
+		klog.ErrorS(err, "Failed to initialize the secondary interface store")
+		return &pc, err
+	}
+
+	if err := pc.reconcileSecondaryInterfaces(pIfaceStore); err != nil {
+		klog.ErrorS(err, "Failed to restore CNI cache and reconcile secondary interfaces")
+		return &pc, err
+	}
+
 	// podUpdateSubscriber can be nil with test code.
 	if podUpdateSubscriber != nil {
 		// Subscribe Pod CNI add/del events.
@@ -502,3 +515,82 @@ func checkForPodSecondaryNetworkAttachement(pod *corev1.Pod) (string, bool) {
 		return netObj, false
 	}
 }
+
+// initializeSecondaryInterfaceStore restore secondary interfacestore when agent restart.
+func (pc *PodController) initializeSecondaryInterfaceStore() error {
+	if pc.ovsBridgeClient == nil {
+		return nil
+	}
+
+	ovsPorts, err := pc.ovsBridgeClient.GetPortList()
+	if err != nil {
+		klog.ErrorS(err, "Failed to list OVS ports for the secondary bridge", "bridgeName", pc.ovsBridgeClient.GetBridgeName())
+		return err
+	}
+
+	ifaceList := make([]*interfacestore.InterfaceConfig, 0, len(ovsPorts))
+	for index := range ovsPorts {
+		port := &ovsPorts[index]
+		ovsPort := &interfacestore.OVSPortConfig{
+			PortUUID: port.UUID,
+			OFPort:   port.OFPort,
+		}
+
+		interfaceType, ok := port.ExternalIDs[interfacestore.AntreaInterfaceTypeKey]
+		if !ok {
+			klog.InfoS("Interface type is not set for the secondary bridge", "interfaceName", port.Name)
+			continue
+		}
+
+		var intf *interfacestore.InterfaceConfig
+		switch interfaceType {
+		case interfacestore.AntreaContainer:
+			intf = cniserver.ParseOVSPortInterfaceConfig(port, ovsPort)
+		default:
+			klog.InfoS("Unknown Antrea interface type for the secondary bridge", "type", interfaceType)
+		}
+
+		if intf != nil {
+			ifaceList = append(ifaceList, intf)
+		}
+
+	}
+
+	pc.interfaceStore.Initialize(ifaceList)
+	klog.InfoS("Successfully initialized the secondary bridge interface store")
+
+	return nil
+}
+
+// reconcileSecondaryInterfaces restore cniCache when agent restart using primary interfacestore.
+func (pc *PodController) reconcileSecondaryInterfaces(pIfaceStore interfacestore.InterfaceStore) error {
+	knownInterfaces := pIfaceStore.GetInterfacesByType(interfacestore.ContainerInterface)
+	for _, containerConfig := range knownInterfaces {
+		config := containerConfig.ContainerInterfaceConfig
+		podKey := podKeyGet(config.PodName, config.PodNamespace)
+
+		pc.cniCache.Store(podKey, &podCNIInfo{
+			containerID: config.ContainerID,
+		})
+		pc.queue.Add(podKey)
+	}
+
+	var staleInterfaces []*interfacestore.InterfaceConfig
+	// secondaryInterfaces is the list of interfaces currently in the secondary local cache and delete ports not in the CNI cache.
+	secondaryInterfaces := pc.interfaceStore.GetInterfacesByType(interfacestore.ContainerInterface)
+	for _, containerConfig := range secondaryInterfaces {
+		_, exists := pIfaceStore.GetContainerInterface(containerConfig.ContainerID)
+		if !exists || containerConfig.OFPort == -1 {
+			staleInterfaces = append(staleInterfaces, containerConfig)
+		}
+	}
+
+	// If there are any stale interfaces, pass them to removeInterfaces()
+	if len(staleInterfaces) > 0 {
+		if err := pc.removeInterfaces(staleInterfaces); err != nil {
+			return fmt.Errorf("failed to remove stale interfaces: %w", err)
+		}
+	}
+
+	return nil
+}

pkg/agent/secondarynetwork/podwatch/controller_test.go

@@ -205,7 +205,7 @@ func TestPodControllerRun(t *testing.T) {
 		client,
 		netdefclient,
 		informerFactory.Core().V1().Pods().Informer(),
-		nil, nil)
+		nil, nil, nil)
 	podController.interfaceConfigurator = interfaceConfigurator
 	podController.ipamAllocator = mockIPAM
 	cniCache := &podController.cniCache