Enable inactivity check on ovndb connection

Signed-off-by: fanriming <fanriming@chinatelecom.cn>
kubeovn · May 31, 2024 · dbd656c · dbd656c
1 parent 3b73df6
commit dbd656c
Show file tree

Hide file tree

Showing 6 changed files with 95 additions and 37 deletions.
diff --git a/pkg/controller/config.go b/pkg/controller/config.go
@@ -20,14 +20,16 @@ import (
 
 // Configuration is the controller conf
 type Configuration struct {
-	BindAddress          string
-	OvnNbAddr            string
-	OvnSbAddr            string
-	OvnTimeout           int
-	CustCrdRetryMaxDelay int
-	CustCrdRetryMinDelay int
-	KubeConfigFile       string
-	KubeRestConfig       *rest.Config
+	BindAddress            string
+	OvnNbAddr              string
+	OvnSbAddr              string
+	OvnTimeout             int
+	OvsDbConnectTimeout    int
+	OvsDbInactivityTimeout int
+	CustCrdRetryMaxDelay   int
+	CustCrdRetryMinDelay   int
+	KubeConfigFile         string
+	KubeRestConfig         *rest.Config
 
 	KubeClient      kubernetes.Interface
 	KubeOvnClient   clientset.Interface
@@ -106,12 +108,14 @@ type Configuration struct {
 // TODO: validate configuration
 func ParseFlags() (*Configuration, error) {
 	var (
-		argOvnNbAddr            = pflag.String("ovn-nb-addr", "", "ovn-nb address")
-		argOvnSbAddr            = pflag.String("ovn-sb-addr", "", "ovn-sb address")
-		argOvnTimeout           = pflag.Int("ovn-timeout", 60, "")
-		argCustCrdRetryMinDelay = pflag.Int("cust-crd-retry-min-delay", 1, "The min delay seconds between custom crd two retries")
-		argCustCrdRetryMaxDelay = pflag.Int("cust-crd-retry-max-delay", 20, "The max delay seconds between custom crd two retries")
-		argKubeConfigFile       = pflag.String("kubeconfig", "", "Path to kubeconfig file with authorization and master location information. If not set use the inCluster token.")
+		argOvnNbAddr              = pflag.String("ovn-nb-addr", "", "ovn-nb address")
+		argOvnSbAddr              = pflag.String("ovn-sb-addr", "", "ovn-sb address")
+		argOvnTimeout             = pflag.Int("ovn-timeout", 60, "")
+		argOvsDbConTimeout        = pflag.Int("ovsdb-con-timeout", 3, "")
+		argOvsDbInactivityTimeout = pflag.Int("ovsdb-inactivity-timeout", 10, "")
+		argCustCrdRetryMinDelay   = pflag.Int("cust-crd-retry-min-delay", 1, "The min delay seconds between custom crd two retries")
+		argCustCrdRetryMaxDelay   = pflag.Int("cust-crd-retry-max-delay", 20, "The max delay seconds between custom crd two retries")
+		argKubeConfigFile         = pflag.String("kubeconfig", "", "Path to kubeconfig file with authorization and master location information. If not set use the inCluster token.")
 
 		argDefaultLogicalSwitch  = pflag.String("default-ls", util.DefaultSubnet, "The default logical switch name")
 		argDefaultCIDR           = pflag.String("default-cidr", "10.16.0.0/16", "Default CIDR for namespace with no logical switch annotation")
@@ -195,6 +199,8 @@ func ParseFlags() (*Configuration, error) {
 		OvnNbAddr:                      *argOvnNbAddr,
 		OvnSbAddr:                      *argOvnSbAddr,
 		OvnTimeout:                     *argOvnTimeout,
+		OvsDbConnectTimeout:            *argOvsDbConTimeout,
+		OvsDbInactivityTimeout:         *argOvsDbInactivityTimeout,
 		CustCrdRetryMinDelay:           *argCustCrdRetryMinDelay,
 		CustCrdRetryMaxDelay:           *argCustCrdRetryMaxDelay,
 		KubeConfigFile:                 *argKubeConfigFile,

diff --git a/pkg/controller/controller.go b/pkg/controller/controller.go
@@ -472,10 +472,19 @@ func Run(ctx context.Context, config *Configuration) {
 	}
 
 	var err error
-	if controller.OVNNbClient, err = ovs.NewOvnNbClient(config.OvnNbAddr, config.OvnTimeout); err != nil {
+	if controller.OVNNbClient, err = ovs.NewOvnNbClient(
+		config.OvnNbAddr,
+		config.OvnTimeout,
+		config.OvsDbConnectTimeout,
+		config.OvsDbInactivityTimeout); err != nil {
 		util.LogFatalAndExit(err, "failed to create ovn nb client")
 	}
-	if controller.OVNSbClient, err = ovs.NewOvnSbClient(config.OvnSbAddr, config.OvnTimeout); err != nil {
+	if controller.OVNSbClient, err = ovs.NewOvnSbClient(
+		config.OvnSbAddr,
+		config.OvnTimeout,
+		config.OvsDbConnectTimeout,
+		config.OvsDbInactivityTimeout,
+	); err != nil {
 		util.LogFatalAndExit(err, "failed to create ovn sb client")
 	}
 	if config.EnableLb {

diff --git a/pkg/ovn_ic_controller/config.go b/pkg/ovn_ic_controller/config.go
@@ -21,10 +21,12 @@ type Configuration struct {
 	KubeClient     kubernetes.Interface
 	KubeOvnClient  clientset.Interface
 
-	PodNamespace string
-	OvnNbAddr    string
-	OvnSbAddr    string
-	OvnTimeout   int
+	PodNamespace           string
+	OvnNbAddr              string
+	OvnSbAddr              string
+	OvnTimeout             int
+	OvsDbConnectTimeout    int
+	OvsDbInactivityTimeout int
 
 	NodeSwitch     string
 	ClusterRouter  string
@@ -35,9 +37,11 @@ func ParseFlags() (*Configuration, error) {
 	var (
 		argKubeConfigFile = pflag.String("kubeconfig", "", "Path to kubeconfig file with authorization and master location information. If not set use the inCluster token.")
 
-		argOvnNbAddr  = pflag.String("ovn-nb-addr", "", "ovn-nb address")
-		argOvnSbAddr  = pflag.String("ovn-sb-addr", "", "ovn-sb address")
-		argOvnTimeout = pflag.Int("ovn-timeout", 60, "")
+		argOvnNbAddr              = pflag.String("ovn-nb-addr", "", "ovn-nb address")
+		argOvnSbAddr              = pflag.String("ovn-sb-addr", "", "ovn-sb address")
+		argOvnTimeout             = pflag.Int("ovn-timeout", 60, "")
+		argOvsDbConTimeout        = pflag.Int("ovsdb-con-timeout", 3, "")
+		argOvsDbInactivityTimeout = pflag.Int("ovsdb-inactivity-timeout", 10, "")
 
 		argClusterRouter  = pflag.String("cluster-router", util.DefaultVpc, "The router name for cluster router")
 		argNodeSwitch     = pflag.String("node-switch", "join", "The name of node gateway switch which help node to access pod network")
@@ -71,10 +75,12 @@ func ParseFlags() (*Configuration, error) {
 	config := &Configuration{
 		KubeConfigFile: *argKubeConfigFile,
 
-		PodNamespace: os.Getenv("POD_NAMESPACE"),
-		OvnNbAddr:    *argOvnNbAddr,
-		OvnSbAddr:    *argOvnSbAddr,
-		OvnTimeout:   *argOvnTimeout,
+		PodNamespace:           os.Getenv("POD_NAMESPACE"),
+		OvnNbAddr:              *argOvnNbAddr,
+		OvnSbAddr:              *argOvnSbAddr,
+		OvnTimeout:             *argOvnTimeout,
+		OvsDbConnectTimeout:    *argOvsDbConTimeout,
+		OvsDbInactivityTimeout: *argOvsDbInactivityTimeout,
 
 		ClusterRouter:  *argClusterRouter,
 		NodeSwitch:     *argNodeSwitch,

diff --git a/pkg/ovn_ic_controller/controller.go b/pkg/ovn_ic_controller/controller.go
@@ -87,10 +87,20 @@ func NewController(config *Configuration) *Controller {
 	}
 
 	var err error
-	if controller.OVNNbClient, err = ovs.NewOvnNbClient(config.OvnNbAddr, config.OvnTimeout); err != nil {
+	if controller.OVNNbClient, err = ovs.NewOvnNbClient(
+		config.OvnNbAddr,
+		config.OvnTimeout,
+		config.OvsDbConnectTimeout,
+		config.OvsDbInactivityTimeout,
+	); err != nil {
 		util.LogFatalAndExit(err, "failed to create ovn nb client")
 	}
-	if controller.OVNSbClient, err = ovs.NewOvnSbClient(config.OvnSbAddr, config.OvnTimeout); err != nil {
+	if controller.OVNSbClient, err = ovs.NewOvnSbClient(
+		config.OvnSbAddr,
+		config.OvnTimeout,
+		config.OvsDbConnectTimeout,
+		config.OvsDbInactivityTimeout,
+	); err != nil {
 		util.LogFatalAndExit(err, "failed to create ovn sb client")
 	}
 

diff --git a/pkg/ovs/ovn.go b/pkg/ovs/ovn.go
@@ -53,7 +53,7 @@ func NewLegacyClient(timeout int) *LegacyClient {
 	}
 }
 
-func NewOvnNbClient(ovnNbAddr string, ovnNbTimeout int) (*OVNNbClient, error) {
+func NewOvnNbClient(ovnNbAddr string, ovnNbTimeout int, ovsDbConTimeout int, ovsDbInactivityTimeout int) (*OVNNbClient, error) {
 	dbModel, err := ovnnb.FullDatabaseModel()
 	if err != nil {
 		klog.Error(err)
@@ -83,7 +83,14 @@ func NewOvnNbClient(ovnNbAddr string, ovnNbTimeout int) (*OVNNbClient, error) {
 	maxRetry := 60
 	var nbClient client.Client
 	for {
-		nbClient, err = ovsclient.NewOvsDbClient(ovsclient.NBDB, ovnNbAddr, dbModel, monitors)
+		nbClient, err = ovsclient.NewOvsDbClient(
+			ovsclient.NBDB,
+			ovnNbAddr,
+			dbModel,
+			monitors,
+			ovsDbConTimeout,
+			ovsDbInactivityTimeout,
+		)
 		if err != nil {
 			klog.Errorf("failed to create OVN NB client: %v", err)
 		} else {
@@ -105,7 +112,7 @@ func NewOvnNbClient(ovnNbAddr string, ovnNbTimeout int) (*OVNNbClient, error) {
 	return c, nil
 }
 
-func NewOvnSbClient(ovnSbAddr string, ovnSbTimeout int) (*OVNSbClient, error) {
+func NewOvnSbClient(ovnSbAddr string, ovnSbTimeout int, ovsDbConTimeout int, ovsDbInactivityTimeout int) (*OVNSbClient, error) {
 	dbModel, err := ovnsb.FullDatabaseModel()
 	if err != nil {
 		klog.Error(err)
@@ -120,7 +127,14 @@ func NewOvnSbClient(ovnSbAddr string, ovnSbTimeout int) (*OVNSbClient, error) {
 	try := 0
 	var sbClient client.Client
 	for {
-		sbClient, err = ovsclient.NewOvsDbClient(ovsclient.SBDB, ovnSbAddr, dbModel, monitors)
+		sbClient, err = ovsclient.NewOvsDbClient(
+			ovsclient.SBDB,
+			ovnSbAddr,
+			dbModel,
+			monitors,
+			ovsDbConTimeout,
+			ovsDbInactivityTimeout,
+		)
 		if err != nil {
 			klog.Errorf("failed to create OVN SB client: %v", err)
 		} else {

diff --git a/pkg/ovsdb/client/client.go b/pkg/ovsdb/client/client.go
@@ -25,7 +25,6 @@ const (
 	ICNBDB = "icnbdb"
 	ICSBDB = "icsbdb"
 )
-const timeout = 3 * time.Second
 
 var namedUUIDCounter uint32
 
@@ -42,10 +41,24 @@ func NamedUUID() string {
 }
 
 // NewOvsDbClient creates a new ovsdb client
-func NewOvsDbClient(db, addr string, dbModel model.ClientDBModel, monitors []client.MonitorOption) (client.Client, error) {
+func NewOvsDbClient(
+	db string,
+	addr string,
+	dbModel model.ClientDBModel,
+	monitors []client.MonitorOption,
+	ovsDbConTimeout int,
+	ovsDbInactivityTimeout int,
+) (client.Client, error) {
 	logger := klog.NewKlogr().WithName("libovsdb").WithValues("db", db)
+	connectTimeout := time.Duration(ovsDbConTimeout) * time.Second
+	inactivityTimeout := time.Duration(ovsDbInactivityTimeout) * time.Second
 	options := []client.Option{
-		client.WithReconnect(timeout, &backoff.ConstantBackOff{Interval: time.Second}),
+		// Reading and parsing the DB after reconnect at scale can (unsurprisingly)
+		// take longer than a normal ovsdb operation. Give it a bit more time so
+		// we don't time out and enter a reconnect loop. In addition it also enables
+		// inactivity check on the ovsdb connection.
+		client.WithInactivityCheck(inactivityTimeout, connectTimeout, &backoff.ZeroBackOff{}),
+
 		client.WithLeaderOnly(true),
 		client.WithLogger(&logger),
 	}
@@ -84,7 +97,7 @@ func NewOvsDbClient(db, addr string, dbModel model.ClientDBModel, monitors []cli
 		klog.Error(err)
 		return nil, err
 	}
-	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(len(endpoints)+1)*timeout)
+	ctx, cancel := context.WithTimeout(context.Background(), connectTimeout)
 	defer cancel()
 	if err = c.Connect(ctx); err != nil {
 		klog.Errorf("failed to connect to OVN NB server %s: %v", addr, err)