[Windows] CNI Server installs OpenFlow entries after PortStatus messa…

…ge is received

Signed-off-by: Wenying Dong <[email protected]>

cmd/antrea-agent/agent.go

@@ -600,6 +600,7 @@ func run(o *Options) error {
 			o.config.CNISocket,
 			o.config.HostProcPathPrefix,
 			nodeConfig,
+			localPodInformer.Get(),
 			k8sClient,
 			routeClient,
 			isChaining,

go.mod

@@ -3,8 +3,8 @@ module antrea.io/antrea
 go 1.23.0
 
 require (
-	antrea.io/libOpenflow v0.14.0
-	antrea.io/ofnet v0.12.0
+	antrea.io/libOpenflow v0.15.0
+	antrea.io/ofnet v0.14.0
 	github.com/ClickHouse/clickhouse-go/v2 v2.6.1
 	github.com/DATA-DOG/go-sqlmock v1.5.2
 	github.com/Mellanox/sriovnet v1.1.0
@@ -113,10 +113,9 @@ require (
 	github.com/aws/smithy-go v1.12.1 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/blang/semver/v4 v4.0.0 // indirect
-	github.com/cenk/hub v1.0.1 // indirect
 	github.com/cenkalti/backoff/v4 v4.3.0 // indirect
-	github.com/cenkalti/hub v1.0.1 // indirect
-	github.com/cenkalti/rpc2 v0.0.0-20180727162946-9642ea02d0aa // indirect
+	github.com/cenkalti/hub v1.0.2 // indirect
+	github.com/cenkalti/rpc2 v1.0.3 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/chai2010/gettext-go v1.0.2 // indirect
 	github.com/containerd/cgroups v1.1.0 // indirect

go.sum

@@ -1,7 +1,7 @@
-antrea.io/libOpenflow v0.14.0 h1:6MS1E52nGQyz/AJ8j3OrotgAc/1ubef5vc5i8ytIxFE=
-antrea.io/libOpenflow v0.14.0/go.mod h1:U8YR0ithWrjwLdUUhyeCsYnlKg/mEFjG5CbPNt1V+j0=
-antrea.io/ofnet v0.12.0 h1:pgygAsEZJUPK/kGmeuIesDh5hoGLpYeavSLdG/Dp8Ao=
-antrea.io/ofnet v0.12.0/go.mod h1:MB3qaInEimj+T8qtpBcIQK+EqeNiY1S/WbUdGk+TzNg=
+antrea.io/libOpenflow v0.15.0 h1:wGk+IVCf8piGZgC4+lbf4qfGrJG5ikzfq5Y1T5LzqmI=
+antrea.io/libOpenflow v0.15.0/go.mod h1:Mq1JEjYrb6eTVA7qjZRj9plVTKcsLM8wnQ87sLLYuiQ=
+antrea.io/ofnet v0.14.0 h1:BGOqg5DdRkvxpBqyoEgWmvGd4EvpacdU/Py1s6qOvSc=
+antrea.io/ofnet v0.14.0/go.mod h1:W5JPYFFcRM7tLwsItgmsKqIhtW/QofyIeNsUIecFaBo=
 cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
 cloud.google.com/go v0.34.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
 cloud.google.com/go v0.38.0/go.mod h1:990N+gfupTy94rShfmMCWGDn0LpTmnzTp2qbd1dvSRU=
@@ -114,14 +114,12 @@ github.com/blang/semver v3.5.1+incompatible h1:cQNTCjp13qL8KC3Nbxr/y2Bqb63oX6wdn
 github.com/blang/semver v3.5.1+incompatible/go.mod h1:kRBLl5iJ+tD4TcOOxsy/0fnwebNt5EWlYSAyrTnjyyk=
 github.com/blang/semver/v4 v4.0.0 h1:1PFHFE6yCCTv8C1TeyNNarDzntLi7wMI5i/pzqYIsAM=
 github.com/blang/semver/v4 v4.0.0/go.mod h1:IbckMUScFkM3pff0VJDNKRiT6TG/YpiHIM2yvyW5YoQ=
-github.com/cenk/hub v1.0.1 h1:RBwXNOF4a8KjD8BJ08XqN8KbrqaGiQLDrgvUGJSHuPA=
-github.com/cenk/hub v1.0.1/go.mod h1:rJM1LNAW0ppT8FMMuPK6c2NP/R2nH/UthtuRySSaf6Y=
 github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8=
 github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
-github.com/cenkalti/hub v1.0.1 h1:UMtjc6dHSaOQTO15SVA50MBIR9zQwvsukQupDrkIRtg=
-github.com/cenkalti/hub v1.0.1/go.mod h1:tcYwtS3a2d9NO/0xDXVJWx3IedurUjYCqFCmpi0lpHs=
-github.com/cenkalti/rpc2 v0.0.0-20180727162946-9642ea02d0aa h1:t+iWhuJE2aropY4uxKMVbyP+IJ29o422f7YAd73aTjg=
-github.com/cenkalti/rpc2 v0.0.0-20180727162946-9642ea02d0aa/go.mod h1:v2npkhrXyk5BCnkNIiPdRI23Uq6uWPUQGL2hnRcRr/M=
+github.com/cenkalti/hub v1.0.2 h1:Nqv9TNaA9boeO2wQFW8o87BY3zKthtnzXmWGmJqhAV8=
+github.com/cenkalti/hub v1.0.2/go.mod h1:8LAFAZcCasb83vfxatMUnZHRoQcffho2ELpHb+kaTJU=
+github.com/cenkalti/rpc2 v1.0.3 h1:OkMsNP/sP9seN1VRCLqhX1xkVGHPoLwWS6fZR14Ji/k=
+github.com/cenkalti/rpc2 v1.0.3/go.mod h1:2yfU5b86vOr16+iY1jN3MvT6Kxc9Nf8j5iZWwUf7iaw=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
 github.com/cespare/xxhash v1.1.0/go.mod h1:XrSqR1VqqWfGrhpAt58auRo0WTKS1nRRg3ghfAqPWnc=
 github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=

pkg/agent/cniserver/pod_configuration.go

@@ -20,11 +20,19 @@ import (
 	"net"
 	"strings"
 	"sync"
+	"time"
 
 	current "github.com/containernetworking/cni/pkg/types/100"
 	"github.com/containernetworking/cni/pkg/version"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
+	k8swait "k8s.io/apimachinery/pkg/util/wait"
+	clientset "k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes/scheme"
+	typedv1 "k8s.io/client-go/kubernetes/typed/core/v1"
+	v1 "k8s.io/client-go/listers/core/v1"
+	"k8s.io/client-go/tools/record"
+	"k8s.io/client-go/util/workqueue"
 	"k8s.io/klog/v2"
 
 	"antrea.io/antrea/pkg/agent/cniserver/ipam"
@@ -61,6 +69,10 @@ const (
 
 var (
 	getNSPath = util.GetNSPath
+	// retryInterval is the interval to re-install Pod OpenFlow entries if any error happened.
+	// Note, using a variable rather than constant for retryInterval because we may use a shorter time in the
+	// test code.
+	retryInterval = 5 * time.Second
 )
 
 type podConfigurator struct {
@@ -76,6 +88,13 @@ type podConfigurator struct {
 	// isSecondaryNetwork is true if this instance of podConfigurator is used to configure
 	// Pod secondary network interfaces.
 	isSecondaryNetwork bool
+	podIfMonitor       *podIfaceMonitor
+
+	eventBroadcaster record.EventBroadcaster
+	record           record.EventRecorder
+	podLister        v1.PodLister
+	kubeClient       clientset.Interface
+	queue            workqueue.TypedDelayingInterface[string]
 }
 
 func newPodConfigurator(
@@ -88,11 +107,23 @@ func newPodConfigurator(
 	isOvsHardwareOffloadEnabled bool,
 	disableTXChecksumOffload bool,
 	podUpdateNotifier channel.Notifier,
+	podLister v1.PodLister,
 ) (*podConfigurator, error) {
 	ifConfigurator, err := newInterfaceConfigurator(ovsDatapathType, isOvsHardwareOffloadEnabled, disableTXChecksumOffload)
 	if err != nil {
 		return nil, err
 	}
+	queue := workqueue.NewTypedDelayingQueueWithConfig[string](
+		workqueue.TypedDelayingQueueConfig[string]{
+			Name: "podMonitor",
+		},
+	)
+	eventBroadcaster := record.NewBroadcaster()
+	recorder := eventBroadcaster.NewRecorder(
+		scheme.Scheme,
+		corev1.EventSource{Component: "AntreaAgentPodConfigurator"},
+	)
+	ifMonitor := newPodInterfaceMonitor(ofClient, ifaceStore, queue)
 	return &podConfigurator{
 		ovsBridgeClient:   ovsBridgeClient,
 		ofClient:          ofClient,
@@ -101,6 +132,10 @@ func newPodConfigurator(
 		gatewayMAC:        gatewayMAC,
 		ifConfigurator:    ifConfigurator,
 		podUpdateNotifier: podUpdateNotifier,
+		podIfMonitor:      ifMonitor,
+		record:            recorder,
+		podLister:         podLister,
+		queue:             queue,
 	}, nil
 }
 
@@ -166,13 +201,13 @@ func getContainerIPsString(ips []net.IP) string {
 // not created for a Pod interface.
 func ParseOVSPortInterfaceConfig(portData *ovsconfig.OVSPortData, portConfig *interfacestore.OVSPortConfig) *interfacestore.InterfaceConfig {
 	if portData.ExternalIDs == nil {
-		klog.V(2).Infof("OVS port %s has no external_ids", portData.Name)
+		klog.V(2).InfoS("OVS port has no external_ids", "port", portData.Name)
 		return nil
 	}
 
 	containerID, found := portData.ExternalIDs[ovsExternalIDContainerID]
 	if !found {
-		klog.V(2).Infof("OVS port %s has no %s in external_ids", portData.Name, ovsExternalIDContainerID)
+		klog.V(2).InfoS("OVS port has no containerID in external_ids", "port", portData.Name)
 		return nil
 	}
 
@@ -187,8 +222,7 @@ func ParseOVSPortInterfaceConfig(portData *ovsconfig.OVSPortData, portConfig *in
 
 	containerMAC, err := net.ParseMAC(portData.ExternalIDs[ovsExternalIDMAC])
 	if err != nil {
-		klog.Errorf("Failed to parse MAC address from OVS external config %s: %v",
-			portData.ExternalIDs[ovsExternalIDMAC], err)
+		klog.ErrorS(err, "Failed to parse MAC address from OVS external config")
 	}
 	podName, _ := portData.ExternalIDs[ovsExternalIDPodName]
 	podNamespace, _ := portData.ExternalIDs[ovsExternalIDPodNamespace]
@@ -279,7 +313,7 @@ func (pc *podConfigurator) createOVSPort(ovsPortName string, ovsAttachInfo map[s
 func (pc *podConfigurator) removeInterfaces(containerID string) error {
 	containerConfig, found := pc.ifaceStore.GetContainerInterface(containerID)
 	if !found {
-		klog.V(2).Infof("Did not find the port for container %s in local cache", containerID)
+		klog.V(2).InfoS("Did not find the port for container in local cache", "container", containerID)
 		return nil
 	}
 
@@ -498,7 +532,7 @@ func (pc *podConfigurator) reconcile(pods []corev1.Pod, containerAccess *contain
 // disconnectInterfaceFromOVS disconnects an existing interface from ovs br-int.
 func (pc *podConfigurator) disconnectInterfaceFromOVS(containerConfig *interfacestore.InterfaceConfig) error {
 	containerID := containerConfig.ContainerID
-	klog.V(2).Infof("Deleting Openflow entries for container %s", containerID)
+	klog.V(2).InfoS("Deleting Openflow entries for container", "container", containerID)
 	if !pc.isSecondaryNetwork {
 		if err := pc.ofClient.UninstallPodFlows(containerConfig.InterfaceName); err != nil {
 			return fmt.Errorf("failed to delete Openflow entries for container %s: %v", containerID, err)
@@ -513,6 +547,12 @@ func (pc *podConfigurator) disconnectInterfaceFromOVS(containerConfig *interface
 	if err := pc.ovsBridgeClient.DeletePort(containerConfig.PortUUID); err != nil {
 		return fmt.Errorf("failed to delete OVS port for container %s interface %s: %v", containerID, containerConfig.InterfaceName, err)
 	}
+
+	// Remove unready Pod info from local cache when Pod is deleted. This is called only on Windows.
+	if pc.podIfMonitor != nil {
+		pc.podIfMonitor.deleteUnreadyPod(containerConfig.InterfaceName)
+	}
+
 	// Remove container configuration from cache.
 	pc.ifaceStore.DeleteInterface(containerConfig)
 	if !pc.isSecondaryNetwork {
@@ -558,7 +598,7 @@ func (pc *podConfigurator) connectInterceptedInterface(
 func (pc *podConfigurator) disconnectInterceptedInterface(podName, podNamespace, containerID string) error {
 	containerConfig, found := pc.ifaceStore.GetContainerInterface(containerID)
 	if !found {
-		klog.V(2).Infof("Did not find the port for container %s in local cache", containerID)
+		klog.V(2).InfoS("Did not find the port for container in local cache", "container", containerID)
 		return nil
 	}
 	for _, ip := range containerConfig.IPs {
@@ -570,3 +610,98 @@ func (pc *podConfigurator) disconnectInterceptedInterface(podName, podNamespace,
 	return pc.disconnectInterfaceFromOVS(containerConfig)
 	// TODO recover pre-connect state? repatch vethpair to original bridge etc ?? to make first CNI happy??
 }
+
+func (pc *podConfigurator) Run(stopCh <-chan struct{}) {
+	pc.eventBroadcaster.StartStructuredLogging(0)
+	pc.eventBroadcaster.StartRecordingToSink(&typedv1.EventSinkImpl{
+		Interface: pc.kubeClient.CoreV1().Events(""),
+	})
+	defer pc.eventBroadcaster.Shutdown()
+	go k8swait.Until(pc.worker, time.Second, stopCh)
+	pc.podIfMonitor.Run(stopCh)
+}
+
+func (pc *podConfigurator) processNextWorkItem() bool {
+	key, quit := pc.queue.Get()
+	if quit {
+		return false
+	}
+	defer pc.queue.Done(key)
+
+	if err := pc.updateUnreadyPod(key); err != nil {
+		// Put the item back on the workqueue to handle any transient errors.
+		pc.queue.AddAfter(key, retryInterval)
+	}
+	return true
+}
+
+// worker is a long-running function that will continually call the processNextWorkItem function in
+// order to read and process a message on the workqueue.
+func (pc *podConfigurator) worker() {
+	for pc.processNextWorkItem() {
+	}
+}
+
+func (pc *podConfigurator) updateUnreadyPod(ovsPort string) error {
+	if !pc.podIfMonitor.unreadyInterfaceExists(ovsPort) {
+		klog.InfoS("Interface does not exist in un-ready state", "name", ovsPort)
+		return nil
+	}
+
+	ifConfig, found := pc.ifaceStore.GetInterfaceByName(ovsPort)
+	if !found {
+		klog.InfoS("Interface config is not found in local cache, remove from unready cache", "name", ovsPort)
+		pc.podIfMonitor.deleteUnreadyPod(ovsPort)
+		return nil
+	}
+
+	if ifConfig.OFPort == 0 {
+		// Add Pod not-ready event if the pod flows are not successfully installed, and the OpenFlow port is not allocated.
+		// Returns error so that we can have a retry after 5s.
+		_ = pc.processPodEvents(ifConfig, false)
+		return fmt.Errorf("pod's OpenFlow port is not ready yet")
+	}
+
+	if !pc.podIfMonitor.interfaceFlowsInstalled(ovsPort) {
+		// Install OpenFlow entries for the Pod.
+		klog.V(2).InfoS("Setting up Openflow entries for OVS port", "port", ovsPort)
+		if err := pc.ofClient.InstallPodFlows(ovsPort, ifConfig.IPs, ifConfig.MAC, uint32(ifConfig.OFPort), ifConfig.VLANID, nil); err != nil {
+			// Add Pod not-ready event if the pod flows installation are failed.
+			// Returns error so that we can have a retry after 5s.
+			_ = pc.processPodEvents(ifConfig, false)
+			return fmt.Errorf("failed to add Openflow entries for OVS port %s: %v", ovsPort, err)
+		}
+
+		// Notify the Pod update event to required components.
+		event := agenttypes.PodUpdate{
+			PodName:      ifConfig.PodName,
+			PodNamespace: ifConfig.PodNamespace,
+			IsAdd:        true,
+			ContainerID:  ifConfig.ContainerID,
+		}
+		pc.podUpdateNotifier.Notify(event)
+	}
+
+	if err := pc.processPodEvents(ifConfig, true); err != nil {
+		return err
+	}
+	pc.podIfMonitor.deleteUnreadyPod(ovsPort)
+	return nil
+}
+
+func (pc *podConfigurator) processPodEvents(ifConfig *interfacestore.InterfaceConfig, installed bool) error {
+	pod, err := pc.podLister.Pods(ifConfig.PodNamespace).Get(ifConfig.PodName)
+	if err != nil {
+		klog.ErrorS(err, "Failed to get Pod, retrying", "Pod", klog.KRef(ifConfig.PodNamespace, ifConfig.PodName))
+		return err
+	}
+
+	if installed {
+		// Add normal event to record Pod network is ready.
+		pc.record.Eventf(pod, corev1.EventTypeNormal, "NetworkIsReady", "Installed Pod '%s/%s' network forwarding rules", ifConfig.PodNamespace, ifConfig.PodName)
+		return nil
+	}
+
+	pc.record.Eventf(pod, corev1.EventTypeWarning, "NetworkNotReady", "Pod '%s/%s' network forwarding rules not installed", ifConfig.PodNamespace, ifConfig.PodName)
+	return nil
+}

pkg/agent/cniserver/pod_configuration_linux_test.go

@@ -687,7 +687,7 @@ func createPodConfigurator(controller *gomock.Controller, testIfaceConfigurator
 	mockOFClient = openflowtest.NewMockClient(controller)
 	ifaceStore = interfacestore.NewInterfaceStore()
 	mockRoute = routetest.NewMockInterface(controller)
-	configurator, _ := newPodConfigurator(mockOVSBridgeClient, mockOFClient, mockRoute, ifaceStore, gwMAC, "system", false, false, channel.NewSubscribableChannel("PodUpdate", 100))
+	configurator, _ := newPodConfigurator(mockOVSBridgeClient, mockOFClient, mockRoute, ifaceStore, gwMAC, "system", false, false, channel.NewSubscribableChannel("PodUpdate", 100), nil)
 	configurator.ifConfigurator = testIfaceConfigurator
 	return configurator
 }