Support OF port state "down" when installing Pod forwarding rules (#6889

)

On Windows, OVS has an issue which doesn't correctly update the port status
after an OpenFlow port is successfully installed. So OVS may send out PortStatus
message with Port state as LIND_DOWN, this issue doesn't impact on the datapath
forwarding.

This change is a workaround to ensure Pod's OpenFlow entries are installed as
long as the OpenFlow port is allocated.

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

go.mod

@@ -58,7 +58,7 @@ require (
 	golang.org/x/mod v0.19.0
 	golang.org/x/net v0.27.0
 	golang.org/x/sync v0.7.0
-	golang.org/x/sys v0.22.0
+	golang.org/x/sys v0.25.0
 	golang.org/x/time v0.5.0
 	golang.org/x/tools v0.23.0
 	golang.zx2c4.com/wireguard/wgctrl v0.0.0-20210506160403-92e472f520a5
@@ -73,7 +73,7 @@ require (
 	k8s.io/apiserver v0.29.2
 	k8s.io/client-go v0.29.2
 	k8s.io/component-base v0.29.2
-	k8s.io/klog/v2 v2.110.1
+	k8s.io/klog/v2 v2.130.1
 	k8s.io/kube-aggregator v0.29.2
 	k8s.io/kube-openapi v0.0.0-20231010175941-2dd684a91f00
 	k8s.io/kubectl v0.29.2

go.sum

@@ -250,7 +250,6 @@ github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9
 github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V4qmtdjCk=
 github.com/go-logr/logr v0.1.0/go.mod h1:ixOQHD9gLJUVQQ2ZOR7zLEifBX6tGkNJF4QyIY7sIas=
 github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
-github.com/go-logr/logr v1.3.0/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
 github.com/go-logr/logr v1.4.2 h1:6pFjapn8bFcIbiKo3XT4j/BhANplGihG6tvd+8rYgrY=
 github.com/go-logr/logr v1.4.2/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
 github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
@@ -976,8 +975,8 @@ golang.org/x/sys v0.2.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.4.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.10.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.22.0 h1:RI27ohtqKCnwULzJLqkv897zojh5/DwS/ENaMzUOaWI=
-golang.org/x/sys v0.22.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.25.0 h1:r+8e+loiHxRqhXVl6ML1nO3l1+oFoWbnlu2Ehimmi34=
+golang.org/x/sys v0.25.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
@@ -1141,8 +1140,8 @@ k8s.io/klog v0.0.0-20181102134211-b9b56d5dfc92/go.mod h1:Gq+BEi5rUBO/HRz0bTSXDUc
 k8s.io/klog v0.3.0/go.mod h1:Gq+BEi5rUBO/HRz0bTSXDUcqjScdoY3a9IHpCEIOOfk=
 k8s.io/klog v1.0.0/go.mod h1:4Bi6QPql/J/LkTDqv7R/cd3hPo4k2DG6Ptcz060Ez5I=
 k8s.io/klog/v2 v2.0.0/go.mod h1:PBfzABfn139FHAV07az/IF9Wp1bkk3vpT2XSJ76fSDE=
-k8s.io/klog/v2 v2.110.1 h1:U/Af64HJf7FcwMcXyKm2RPM22WZzyR7OSpYj5tg3cL0=
-k8s.io/klog/v2 v2.110.1/go.mod h1:YGtd1984u+GgbuZ7e08/yBuAfKLSO0+uR1Fhi6ExXjo=
+k8s.io/klog/v2 v2.130.1 h1:n9Xl7H1Xvksem4KFG4PYbdQCQxqc/tTUyrgXaOhHSzk=
+k8s.io/klog/v2 v2.130.1/go.mod h1:3Jpz1GvMt720eyJH1ckRHK1EDfpxISzJ7I9OYgaDtPE=
 k8s.io/kms v0.29.2 h1:MDsbp98gSlEQs7K7dqLKNNTwKFQRYYvO4UOlBOjNy6Y=
 k8s.io/kms v0.29.2/go.mod h1:s/9RC4sYRZ/6Tn6yhNjbfJuZdb8LzlXhdlBnKizeFDo=
 k8s.io/kube-aggregator v0.29.2 h1:z9qJn5wlGmGaX6EfM7OEhr6fq6SBjDKR6tPRZ/qgxeY=

hack/update-codegen-dockerized.sh

@@ -140,17 +140,17 @@ function generate_antrea_client_code {
     --go-header-file hack/boilerplate/license_header.go.txt
 
   $GOPATH/bin/deepcopy-gen \
-    --output-file zz_generated.deepcopy.go \
-    --go-header-file hack/boilerplate/license_header.go.txt \
-     "${ANTREA_PKG}/pkg/apis/controlplane" \
-     "${ANTREA_PKG}/pkg/apis/controlplane/v1beta2" \
-     "${ANTREA_PKG}/pkg/apis/system/v1beta1" \
-     "${ANTREA_PKG}/pkg/apis/crd/v1alpha1" \
-     "${ANTREA_PKG}/pkg/apis/crd/v1alpha2" \
-     "${ANTREA_PKG}/pkg/apis/crd/v1beta1" \
-     "${ANTREA_PKG}/pkg/apis/stats" \
-     "${ANTREA_PKG}/pkg/apis/stats/v1alpha1" \
-     "${ANTREA_PKG}/pkg/agent/interfacestore"
+    --input-dirs "${ANTREA_PKG}/pkg/apis/controlplane" \
+    --input-dirs "${ANTREA_PKG}/pkg/apis/controlplane/v1beta2" \
+    --input-dirs "${ANTREA_PKG}/pkg/apis/system/v1beta1" \
+    --input-dirs "${ANTREA_PKG}/pkg/apis/crd/v1alpha1" \
+    --input-dirs "${ANTREA_PKG}/pkg/apis/crd/v1alpha2" \
+    --input-dirs "${ANTREA_PKG}/pkg/apis/crd/v1beta1" \
+    --input-dirs "${ANTREA_PKG}/pkg/apis/stats" \
+    --input-dirs "${ANTREA_PKG}/pkg/apis/stats/v1alpha1" \
+    --input-dirs "${ANTREA_PKG}/pkg/agent/interfacestore" \
+    -O zz_generated.deepcopy \
+    --go-header-file hack/boilerplate/license_header.go.txt
 
   $GOPATH/bin/conversion-gen  \
     --input-dirs "${ANTREA_PKG}/pkg/apis/controlplane/v1beta2,${ANTREA_PKG}/pkg/apis/controlplane/" \

pkg/agent/cniserver/pod_configuration.go

@@ -65,6 +65,8 @@ const (
 
 const (
 	defaultIFDevName = "eth0"
+	minRetryDelay    = 5 * time.Second
+	maxRetryDelay    = 300 * time.Second
 )
 
 var (
@@ -95,7 +97,7 @@ type podConfigurator struct {
 	podListerSynced  cache.InformerSynced
 	podLister        v1.PodLister
 	kubeClient       clientset.Interface
-	unreadyPortQueue workqueue.TypedDelayingInterface[string]
+	unreadyPortQueue workqueue.RateLimitingInterface
 	statusCh         chan *openflow15.PortStatus
 }
 
@@ -607,7 +609,7 @@ func (pc *podConfigurator) processNextWorkItem() bool {
 	}
 	defer pc.unreadyPortQueue.Done(key)
 
-	if err := pc.updateUnreadyPod(key); err != nil {
+	if err := pc.updateUnreadyPod(key.(string)); err != nil {
 		klog.ErrorS(err, "Failed install OpenFlow entries for OVS port interface", "name", key)
 		// Put the item back on the workqueue to handle any transient errors.
 		pc.unreadyPortQueue.AddAfter(key, retryInterval)
@@ -677,12 +679,26 @@ func (pc *podConfigurator) recordPodEvent(ifConfig *interfacestore.InterfaceConf
 }
 
 func (pc *podConfigurator) processPortStatusMessage(status *openflow15.PortStatus) {
-	// Update Pod OpenFlow entries only after the OpenFlow port state is live.
-	if status.Desc.State != openflow15.PS_LIVE {
+	ofPort := status.Desc.PortNo
+	state := status.Desc.State
+	// Update Pod OpenFlow entries only after the OpenFlow port state is live or down.
+	// Accepting Port state "openflow15.PS_LINK_DOWN" is a workaround for Windows OVS issue https://github.com/openvswitch/ovs-issues/issues/351.
+	// In which OVS does not correctly implement function netdev_windows_update_flags, so OVS doesn't update ifp_flags
+	// after a new OpenFlow port is successfully installed. Since this OVS issue doesn't have side impact on datapath
+	// packets forwarding, antrea-agent will ignore the bad state to ensure the Pod's OpenFlow entries are installed as
+	// long as the port number is allocated.
+	if state != openflow15.PS_LIVE && state != openflow15.PS_LINK_DOWN {
+		klog.InfoS("Ignoring the OVS port status message with undesired state", "ofPort", ofPort, "state", state)
+		return
+	}
+
+	if ofPort == 0 {
+		klog.InfoS("Ignoring the OVS port status message with undesired port number", "ofPort", ofPort, "state", state)
 		return
 	}
+
 	ovsPort := string(bytes.Trim(status.Desc.Name, "\x00"))
-	ofPort := status.Desc.PortNo
+	klog.InfoS("Processing OVS port status message", "ovsPort", ovsPort, "ofPort", ofPort, "state", state)
 
 	ifConfig, found := pc.ifaceStore.GetInterfaceByName(ovsPort)
 	if !found {

pkg/agent/cniserver/pod_configuration_test.go

@@ -78,7 +78,7 @@ type mockClients struct {
 }
 
 func newMockClients(ctrl *gomock.Controller, nodeName string, objects ...runtime.Object) *mockClients {
-	kubeClient := fakeclientset.NewClientset(objects...)
+	kubeClient := fakeclientset.NewSimpleClientset(objects...)
 
 	localPodInformer := coreinformers.NewFilteredPodInformer(
 		kubeClient,
@@ -155,11 +155,8 @@ func mockRetryInterval(t *testing.T) {
 func newTestPodConfigurator(testClients *mockClients, waiter *asyncWaiter) *podConfigurator {
 	interfaceStore := interfacestore.NewInterfaceStore()
 	eventBroadcaster := record.NewBroadcaster()
-	queue := workqueue.NewTypedDelayingQueueWithConfig[string](
-		workqueue.TypedDelayingQueueConfig[string]{
-			Name: "podConfigurator",
-		},
-	)
+	queue := workqueue.NewNamedRateLimitingQueue(
+		workqueue.NewItemExponentialFailureRateLimiter(minRetryDelay, maxRetryDelay), "podConfigurator")
 	podCfg := &podConfigurator{
 		kubeClient:       testClients.kubeClient,
 		ofClient:         testClients.ofClient,
@@ -393,7 +390,7 @@ func TestProcessPortStatusMessage(t *testing.T) {
 					PortNo: 1,
 					Length: 72,
 					Name:   []byte(fmt.Sprintf("%s\x00", podIfName)),
-					State:  openflow15.PS_LINK_DOWN,
+					State:  openflow15.PS_BLOCKED,
 				},
 			},
 			ovsPortName:     podIfName,
@@ -417,10 +414,8 @@ func TestProcessPortStatusMessage(t *testing.T) {
 		},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
-			queue := workqueue.NewTypedDelayingQueueWithConfig[string](
-				workqueue.TypedDelayingQueueConfig[string]{
-					Name: "podMonitor",
-				})
+			queue := workqueue.NewNamedRateLimitingQueue(
+				workqueue.NewItemExponentialFailureRateLimiter(minRetryDelay, maxRetryDelay), "podMonitor")
 			podCfg := &podConfigurator{
 				ifaceStore:       interfacestore.NewInterfaceStore(),
 				statusCh:         make(chan *openflow15.PortStatus),

pkg/agent/cniserver/pod_configuration_windows.go

@@ -141,11 +141,8 @@ func (pc *podConfigurator) reconcileMissingPods(ifConfigs []*interfacestore.Inte
 func (pc *podConfigurator) initPortStatusMonitor(podInformer cache.SharedIndexInformer) {
 	pc.podLister = v1.NewPodLister(podInformer.GetIndexer())
 	pc.podListerSynced = podInformer.HasSynced
-	pc.unreadyPortQueue = workqueue.NewTypedDelayingQueueWithConfig[string](
-		workqueue.TypedDelayingQueueConfig[string]{
-			Name: workerName,
-		},
-	)
+	pc.unreadyPortQueue = workqueue.NewNamedRateLimitingQueue(
+		workqueue.NewItemExponentialFailureRateLimiter(minRetryDelay, maxRetryDelay), workerName)
 	eventBroadcaster := record.NewBroadcaster()
 	pc.eventBroadcaster = eventBroadcaster
 	pc.recorder = eventBroadcaster.NewRecorder(

pkg/agent/cniserver/server_linux_test.go

@@ -612,7 +612,7 @@ func TestCmdCheck(t *testing.T) {
 
 func TestReconcile(t *testing.T) {
 	controller := gomock.NewController(t)
-	kubeClient := fakeclientset.NewClientset(pod1, pod2, pod3)
+	kubeClient := fakeclientset.NewSimpleClientset(pod1, pod2, pod3)
 	mockOVSBridgeClient = ovsconfigtest.NewMockOVSBridgeClient(controller)
 	mockOFClient = openflowtest.NewMockClient(controller)
 	ifaceStore = interfacestore.NewInterfaceStore()