optimize provider network (#2099)

dist/images/install.sh

@@ -720,6 +720,10 @@ spec:
                   type: array
                   items:
                     type: string
+                notReadyNodes:
+                  type: array
+                  items:
+                    type: string
                 vlans:
                   type: array
                   items:

go.mod

@@ -12,6 +12,7 @@ require (
 	github.com/containernetworking/plugins v0.9.1
 	github.com/coreos/go-iptables v0.6.0
 	github.com/emicklei/go-restful/v3 v3.9.0
+	github.com/evanphx/json-patch v5.6.0+incompatible
 	github.com/golang/protobuf v1.5.2
 	github.com/greenpau/ovsdb v0.0.0-20181114004433-3582b85e8968
 	github.com/k8snetworkplumbingwg/network-attachment-definition-client v1.1.1-0.20210510153419-66a699ae3b05
@@ -54,7 +55,6 @@ require (
 	github.com/eapache/channels v1.1.0 // indirect
 	github.com/eapache/queue v1.1.0 // indirect
 	github.com/elazarl/goproxy v0.0.0-20190630181448-f1e96bc0f4c5 // indirect
-	github.com/evanphx/json-patch v5.6.0+incompatible // indirect
 	github.com/fsnotify/fsnotify v1.5.1 // indirect
 	github.com/go-ini/ini v1.42.0 // indirect
 	github.com/go-kit/kit v0.9.0 // indirect

pkg/apis/kubeovn/v1/types.go

@@ -270,6 +270,9 @@ type ProviderNetworkStatus struct {
 	// +optional
 	ReadyNodes []string `json:"readyNodes,omitempty"`
 
+	// +optional
+	NotReadyNodes []string `json:"notReadyNodes,omitempty"`
+
 	// +optional
 	Vlans []string `json:"vlans,omitempty"`
 

pkg/controller/controller.go

@@ -83,9 +83,8 @@ type Controller struct {
 	vlansLister kubeovnlister.VlanLister
 	vlanSynced  cache.InformerSynced
 
-	providerNetworksLister     kubeovnlister.ProviderNetworkLister
-	providerNetworkSynced      cache.InformerSynced
-	updateProviderNetworkQueue workqueue.RateLimitingInterface
+	providerNetworksLister kubeovnlister.ProviderNetworkLister
+	providerNetworkSynced  cache.InformerSynced
 
 	addVlanQueue    workqueue.RateLimitingInterface
 	delVlanQueue    workqueue.RateLimitingInterface
@@ -211,9 +210,8 @@ func NewController(config *Configuration) *Controller {
 		delVlanQueue:    workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "DelVlan"),
 		updateVlanQueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "UpdateVlan"),
 
-		providerNetworksLister:     providerNetworkInformer.Lister(),
-		providerNetworkSynced:      providerNetworkInformer.Informer().HasSynced,
-		updateProviderNetworkQueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "UpdateProviderNetwork"),
+		providerNetworksLister: providerNetworkInformer.Lister(),
+		providerNetworkSynced:  providerNetworkInformer.Informer().HasSynced,
 
 		podsLister:             podInformer.Lister(),
 		podsSynced:             podInformer.Informer().HasSynced,
@@ -323,10 +321,6 @@ func NewController(config *Configuration) *Controller {
 		UpdateFunc: controller.enqueueUpdateVlan,
 	})
 
-	providerNetworkInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
-		UpdateFunc: controller.enqueueUpdateProviderNetwork,
-	})
-
 	if config.EnableNP {
 		npInformer := informerFactory.Networking().V1().NetworkPolicies()
 		controller.npsLister = npInformer.Lister()
@@ -457,8 +451,6 @@ func (c *Controller) shutdown() {
 	c.delVlanQueue.ShutDown()
 	c.updateVlanQueue.ShutDown()
 
-	c.updateProviderNetworkQueue.ShutDown()
-
 	c.addOrUpdateVpcQueue.ShutDown()
 	c.updateVpcStatusQueue.ShutDown()
 	c.delVpcQueue.ShutDown()
@@ -543,7 +535,6 @@ func (c *Controller) startWorkers(stopCh <-chan struct{}) {
 
 	go wait.Until(c.runDelVpcWorker, time.Second, stopCh)
 	go wait.Until(c.runUpdateVpcStatusWorker, time.Second, stopCh)
-	go wait.Until(c.runUpdateProviderNetworkWorker, time.Second, stopCh)
 
 	if c.config.EnableLb {
 		// run in a single worker to avoid delete the last vip, which will lead ovn to delete the loadbalancer

pkg/controller/provider-network.go

@@ -4,126 +4,73 @@ import (
 	"context"
 	"fmt"
 
-	corev1 "k8s.io/api/core/v1"
-	k8serrors "k8s.io/apimachinery/pkg/api/errors"
+	"github.com/kubeovn/kube-ovn/pkg/util"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
-	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
-	"k8s.io/client-go/tools/cache"
 	"k8s.io/klog/v2"
-
-	kubeovnv1 "github.com/kubeovn/kube-ovn/pkg/apis/kubeovn/v1"
-	"github.com/kubeovn/kube-ovn/pkg/util"
 )
 
-func (c *Controller) enqueueUpdateProviderNetwork(_, obj interface{}) {
-	if !c.isLeader() {
-		return
-	}
-
-	key, err := cache.MetaNamespaceKeyFunc(obj)
+func (c *Controller) resyncProviderNetworkStatus() {
+	klog.Infof("start to sync ProviderNetwork status")
+	pns, err := c.providerNetworksLister.List(labels.Everything())
 	if err != nil {
-		utilruntime.HandleError(err)
+		klog.Errorf("failed to list provider network: %v", err)
 		return
 	}
 
-	klog.V(3).Infof("enqueue update provider network %s", key)
-	c.updateProviderNetworkQueue.Add(key)
-}
-
-func (c *Controller) runUpdateProviderNetworkWorker() {
-	for c.processNextUpdateProviderNetworkWorkItem() {
-	}
-}
-
-func (c *Controller) processNextUpdateProviderNetworkWorkItem() bool {
-	obj, shutdown := c.updateProviderNetworkQueue.Get()
-	if shutdown {
-		return false
-	}
-
-	err := func(obj interface{}) error {
-		defer c.updateProviderNetworkQueue.Done(obj)
-		var key string
-		var ok bool
-		if key, ok = obj.(string); !ok {
-			c.updateProviderNetworkQueue.Forget(obj)
-			utilruntime.HandleError(fmt.Errorf("expected string in workqueue but got %#v", obj))
-			return nil
-		}
-		if err := c.handleUpdateProviderNetwork(key); err != nil {
-			c.updateProviderNetworkQueue.AddRateLimited(key)
-			return fmt.Errorf("error syncing '%s': %s, requeuing", key, err.Error())
-		}
-		c.updateProviderNetworkQueue.Forget(obj)
-		return nil
-	}(obj)
-
-	if err != nil {
-		utilruntime.HandleError(err)
-	}
-
-	return true
-}
-
-func (c *Controller) handleUpdateProviderNetwork(key string) error {
-	pn, err := c.providerNetworksLister.Get(key)
-	if err != nil {
-		if k8serrors.IsNotFound(err) {
-			return nil
-		}
-		return err
-	}
-
 	nodes, err := c.nodesLister.List(labels.Everything())
 	if err != nil {
 		klog.Errorf("failed to list nodes: %v", err)
-		return err
+		return
 	}
 
-	if providerNetworkIsReady(pn, nodes) != pn.Status.Ready {
-		newPn := pn.DeepCopy()
-		newPn.Status.Ready = !pn.Status.Ready
-		_, err = c.config.KubeOvnClient.KubeovnV1().ProviderNetworks().UpdateStatus(context.Background(), newPn, metav1.UpdateOptions{})
-		if err != nil {
-			klog.Errorf("failed to update status of provider network %s: %v", pn.Name, err)
-			return err
+	for _, cachedPn := range pns {
+		pn := cachedPn.DeepCopy()
+		var readyNodes, notReadyNodes, expectNodes []string
+		for _, node := range nodes {
+			if util.ContainsString(pn.Spec.ExcludeNodes, node.Name) {
+				pn.Status.RemoveNodeConditions(node.Name)
+				continue
+			}
+			if node.Labels[fmt.Sprintf(util.ProviderNetworkReadyTemplate, pn.Name)] == "true" {
+				pn.Status.SetNodeReady(node.Name, "InitOVSBridgeSucceeded", "")
+				readyNodes = append(readyNodes, node.Name)
+			} else {
+				pods, err := c.config.KubeClient.CoreV1().Pods("").List(context.Background(), metav1.ListOptions{
+					LabelSelector: "app=kube-ovn-cni",
+					FieldSelector: fmt.Sprintf("spec.nodeName=%s", node.Name),
+				})
+				if err != nil {
+					klog.Errorf("failed to list pod: %v", err)
+					continue
+				}
+
+				var errMsg string
+				if len(pods.Items) == 1 && pods.Items[0].Annotations != nil {
+					errMsg = pods.Items[0].Annotations[fmt.Sprintf(util.ProviderNetworkErrMessageTemplate, pn.Name)]
+				}
+				pn.Status.SetNodeNotReady(node.Name, "InitOVSBridgeFailed", errMsg)
+				notReadyNodes = append(notReadyNodes, node.Name)
+			}
 		}
-	}
 
-	return nil
-}
-
-func (c *Controller) resyncProviderNetworkStatus() {
-	nodeList, err := c.nodesLister.List(labels.Everything())
-	if err != nil {
-		klog.Errorf("failed to get nodes %v", err)
-		return
-	}
-	pnList, err := c.providerNetworksLister.List(labels.Everything())
-	if err != nil {
-		klog.Errorf("failed to get provider networks %v", err)
-		return
-	}
-
-	for _, pn := range pnList {
-		if providerNetworkIsReady(pn, nodeList) != pn.Status.Ready {
-			newPn := pn.DeepCopy()
-			newPn.Status.Ready = !newPn.Status.Ready
-			_, err = c.config.KubeOvnClient.KubeovnV1().ProviderNetworks().UpdateStatus(context.Background(), newPn, metav1.UpdateOptions{})
-			if err != nil {
-				klog.Errorf("failed to update status of provider network %s: %v", pn.Name, err)
+		expectNodes = append(readyNodes, notReadyNodes...)
+		conditionsChange := false
+		for _, c := range pn.Status.Conditions {
+			if !util.ContainsString(expectNodes, c.Node) {
+				pn.Status.RemoveNodeConditions(c.Node)
+				conditionsChange = true
 			}
 		}
-	}
-}
 
-func providerNetworkIsReady(pn *kubeovnv1.ProviderNetwork, nodes []*corev1.Node) bool {
-	for _, node := range nodes {
-		if !util.ContainsString(pn.Spec.ExcludeNodes, node.Name) &&
-			!util.ContainsString(pn.Status.ReadyNodes, node.Name) {
-			return false
+		if conditionsChange || len(util.DiffStringSlice(pn.Status.ReadyNodes, readyNodes)) != 0 ||
+			len(util.DiffStringSlice(pn.Status.NotReadyNodes, notReadyNodes)) != 0 {
+			pn.Status.ReadyNodes = readyNodes
+			pn.Status.NotReadyNodes = notReadyNodes
+			pn.Status.Ready = (len(notReadyNodes) == 0)
+			if _, err = c.config.KubeOvnClient.KubeovnV1().ProviderNetworks().UpdateStatus(context.Background(), pn, metav1.UpdateOptions{}); err != nil {
+				klog.Errorf("failed to update provider network %s: %v", pn.Name, err)
+			}
 		}
 	}
-	return true
 }