Merge pull request #1154 from dcbw/shared-informer

Performance and efficiency improvements in daemon/server mode

cmd/multus-daemon/main.go

@@ -30,7 +30,6 @@ import (
 	"syscall"
 	"time"
 
-	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	utilwait "k8s.io/apimachinery/pkg/util/wait"
 
 	"gopkg.in/k8snetworkplumbingwg/multus-cni.v4/pkg/logging"
@@ -58,9 +57,6 @@ func main() {
 		os.Exit(4)
 	}
 
-	configWatcherDoneChannel := make(chan struct{})
-	serverDoneChannel := make(chan struct{})
-	multusConfigFile := ""
 	ctx := context.Background()
 	ctx, cancel := context.WithCancel(ctx)
 
@@ -87,65 +83,42 @@ func main() {
 		logging.Verbosef("Readiness Indicator file check done!")
 	}
 
-	if err := startMultusDaemon(ctx, daemonConf, serverDoneChannel); err != nil {
-		logging.Panicf("failed start the multus thick-plugin listener: %v", err)
-		os.Exit(3)
-	}
-
-	// Wait until daemon ready
-	logging.Verbosef("API readiness check")
-	if waitUntilAPIReady(daemonConf.SocketDir) != nil {
-		logging.Panicf("failed to ready multus-daemon socket: %v", err)
-		os.Exit(1)
-	}
-	logging.Verbosef("API readiness check done!")
-
-	// Generate multus CNI config from current CNI config
+	var configManager *config.Manager
+	var ignoreReadinessIndicator bool
 	if multusConf.MultusConfigFile == "auto" {
 		if multusConf.CNIVersion == "" {
 			_ = logging.Errorf("the CNI version is a mandatory parameter when the '-multus-config-file=auto' option is used")
 		}
 
-		var configManager *config.Manager
-		if multusConf.MultusMasterCni == "" {
-			configManager, err = config.NewManager(*multusConf, multusConf.MultusAutoconfigDir, multusConf.ForceCNIVersion)
-		} else {
-			configManager, err = config.NewManagerWithExplicitPrimaryCNIPlugin(
-				*multusConf, multusConf.MultusAutoconfigDir, multusConf.MultusMasterCni, multusConf.ForceCNIVersion)
-		}
+		// Generate multus CNI config from current CNI config
+		configManager, err = config.NewManager(*multusConf)
 		if err != nil {
 			_ = logging.Errorf("failed to create the configuration manager for the primary CNI plugin: %v", err)
 			os.Exit(2)
 		}
-
-		if multusConf.OverrideNetworkName {
-			if err := configManager.OverrideNetworkName(); err != nil {
-				_ = logging.Errorf("could not override the network name: %v", err)
-			}
-		}
-
-		generatedMultusConfig, err := configManager.GenerateConfig()
-		if err != nil {
-			_ = logging.Errorf("failed to generated the multus configuration: %v", err)
-		}
-		logging.Verbosef("Generated MultusCNI config: %s", generatedMultusConfig)
-
-		multusConfigFile, err = configManager.PersistMultusConfig(generatedMultusConfig)
-		if err != nil {
-			_ = logging.Errorf("failed to persist the multus configuration: %v", err)
-		}
-
-		go func(ctx context.Context, doneChannel chan<- struct{}) {
-			if err := configManager.MonitorPluginConfiguration(ctx, doneChannel); err != nil {
-				_ = logging.Errorf("error watching file: %v", err)
-			}
-		}(ctx, configWatcherDoneChannel)
+		// ConfigManager watches the readiness indicator file (if configured)
+		// and exits the daemon when that is removed. The CNIServer does
+		// not need to re-do that check every CNI operation
+		ignoreReadinessIndicator = true
 	} else {
 		if err := copyUserProvidedConfig(multusConf.MultusConfigFile, multusConf.CniConfigDir); err != nil {
 			logging.Errorf("failed to copy the user provided configuration %s: %v", multusConf.MultusConfigFile, err)
 		}
 	}
 
+	if err := startMultusDaemon(ctx, daemonConf, ignoreReadinessIndicator); err != nil {
+		logging.Panicf("failed start the multus thick-plugin listener: %v", err)
+		os.Exit(3)
+	}
+
+	// Wait until daemon ready
+	logging.Verbosef("API readiness check")
+	if waitUntilAPIReady(daemonConf.SocketDir) != nil {
+		logging.Panicf("failed to ready multus-daemon socket: %v", err)
+		os.Exit(1)
+	}
+	logging.Verbosef("API readiness check done!")
+
 	signalCh := make(chan os.Signal, 16)
 	signal.Notify(signalCh, syscall.SIGINT, syscall.SIGTERM)
 	go func() {
@@ -156,15 +129,11 @@ func main() {
 	}()
 
 	var wg sync.WaitGroup
-	if multusConf.MultusConfigFile == "auto" {
-		wg.Add(1)
-		go func() {
-			<-configWatcherDoneChannel
-			logging.Verbosef("ConfigWatcher done")
-			logging.Verbosef("Delete old config @ %v", multusConfigFile)
-			os.Remove(multusConfigFile)
-			wg.Done()
-		}()
+	if configManager != nil {
+		if err := configManager.Start(ctx, &wg); err != nil {
+			_ = logging.Errorf("failed to start config manager: %v", err)
+			os.Exit(3)
+		}
 	}
 
 	wg.Wait()
@@ -181,7 +150,7 @@ func waitUntilAPIReady(socketPath string) error {
 	})
 }
 
-func startMultusDaemon(ctx context.Context, daemonConfig *srv.ControllerNetConf, done chan struct{}) error {
+func startMultusDaemon(ctx context.Context, daemonConfig *srv.ControllerNetConf, ignoreReadinessIndicator bool) error {
 	if user, err := user.Current(); err != nil || user.Uid != "0" {
 		return fmt.Errorf("failed to run multus-daemon with root: %v, now running in uid: %s", err, user.Uid)
 	}
@@ -190,7 +159,7 @@ func startMultusDaemon(ctx context.Context, daemonConfig *srv.ControllerNetConf,
 		return fmt.Errorf("failed to prepare the cni-socket for communicating with the shim: %w", err)
 	}
 
-	server, err := srv.NewCNIServer(daemonConfig, daemonConfig.ConfigFileContents)
+	server, err := srv.NewCNIServer(daemonConfig, daemonConfig.ConfigFileContents, ignoreReadinessIndicator)
 	if err != nil {
 		return fmt.Errorf("failed to create the server: %v", err)
 	}
@@ -208,15 +177,8 @@ func startMultusDaemon(ctx context.Context, daemonConfig *srv.ControllerNetConf,
 		return fmt.Errorf("failed to start the CNI server using socket %s. Reason: %+v", api.SocketPath(daemonConfig.SocketDir), err)
 	}
 
-	server.SetKeepAlivesEnabled(false)
-	go func() {
-		utilwait.UntilWithContext(ctx, func(ctx context.Context) {
-			logging.Debugf("open for business")
-			if err := server.Serve(l); err != nil {
-				utilruntime.HandleError(fmt.Errorf("CNI server Serve() failed: %v", err))
-			}
-		}, 0)
-	}()
+	server.Start(ctx, l)
+
 	go func() {
 		<-ctx.Done()
 		server.Shutdown(context.Background())

cmd/multus/main.go

@@ -45,7 +45,7 @@ func main() {
 
 	skel.PluginMain(
 		func(args *skel.CmdArgs) error {
-			result, err := multus.CmdAdd(args, nil, nil)
+			result, err := multus.CmdAdd(args, nil, nil, nil)
 			if err != nil {
 				return err
 			}
@@ -54,6 +54,6 @@ func main() {
 		func(args *skel.CmdArgs) error {
 			return multus.CmdCheck(args, nil, nil)
 		},
-		func(args *skel.CmdArgs) error { return multus.CmdDel(args, nil, nil) },
+		func(args *skel.CmdArgs) error { return multus.CmdDel(args, nil, nil, nil) },
 		cniversion.All, "meta-plugin that delegates to other CNI plugins")
 }

deployments/multus-daemonset-thick.yml

@@ -69,6 +69,7 @@ rules:
       - pods/status
     verbs:
       - get
+      - list
       - update
   - apiGroups:
       - ""
@@ -183,6 +184,11 @@ spec:
             - name: hostroot
               mountPath: /hostroot
               mountPropagation: HostToContainer
+          env:
+            - name: MULTUS_NODE_NAME
+              valueFrom:
+                fieldRef:
+                  fieldPath: spec.nodeName
       initContainers:
         - name: install-multus-binary
           image: ghcr.io/k8snetworkplumbingwg/multus-cni:snapshot-thick

e2e/templates/multus-daemonset-thick.yml.j2

@@ -43,6 +43,7 @@ rules:
       - pods/status
     verbs:
       - get
+      - list
       - update
   - apiGroups:
       - ""
@@ -157,6 +158,11 @@ spec:
         - name: multus-daemon-config
           mountPath: /etc/cni/net.d/multus.d
           readOnly: true
+        env:
+        - name: MULTUS_NODE_NAME
+          valueFrom:
+            fieldRef:
+              fieldPath: spec.nodeName
       initContainers:
       - name: install-multus-shim
         image: localhost:5000/multus:e2e

go.mod

@@ -19,7 +19,7 @@ require (
 	gopkg.in/natefinch/lumberjack.v2 v2.0.0
 	k8s.io/api v0.22.8
 	k8s.io/apimachinery v0.22.8
-	k8s.io/client-go v0.22.8
+	k8s.io/client-go v1.5.2
 	k8s.io/klog v1.0.0
 	k8s.io/klog/v2 v2.60.1 // indirect
 	k8s.io/kube-openapi v0.0.0-20220413171646-5e7f5fdc6da6 // indirect

pkg/k8sclient/k8sclient.go

@@ -396,13 +396,14 @@ func TryLoadPodDelegates(pod *v1.Pod, conf *types.NetConf, clientInfo *ClientInf
 // InClusterK8sClient returns the `k8s.ClientInfo` struct to use to connect to
 // the k8s API.
 func InClusterK8sClient() (*ClientInfo, error) {
-	config, err := rest.InClusterConfig()
+	clientInfo, err := GetK8sClient("", nil)
 	if err != nil {
 		return nil, err
 	}
-
-	logging.Debugf("InClusterK8sClient: in cluster config: %+v", config)
-	return NewClientInfo(config)
+	if clientInfo == nil {
+		return nil, fmt.Errorf("failed to create in-cluster kube client")
+	}
+	return clientInfo, err
 }
 
 // GetK8sClient gets client info from kubeconfig
@@ -440,13 +441,17 @@ func GetK8sClient(kubeconfig string, kubeClient *ClientInfo) (*ClientInfo, error
 	config.ContentType = "application/vnd.kubernetes.protobuf"
 	// Set the config timeout to one minute.
 	config.Timeout = time.Minute
+	// Allow multus (especially in server mode) to make more concurrent requests
+	// to reduce client-side throttling
+	config.QPS = 50
+	config.Burst = 50
 
-	return NewClientInfo(config)
+	return newClientInfo(config)
 }
 
-// NewClientInfo returns a `ClientInfo` from a configuration created from an
+// newClientInfo returns a `ClientInfo` from a configuration created from an
 // existing kubeconfig file.
-func NewClientInfo(config *rest.Config) (*ClientInfo, error) {
+func newClientInfo(config *rest.Config) (*ClientInfo, error) {
 	client, err := kubernetes.NewForConfig(config)
 	if err != nil {
 		return nil, err