Merge pull request #939 from s1061123/fix/server

Fix multus-daemon in case of user given config

cmd/multus-daemon/main.go

@@ -18,6 +18,7 @@
 package main
 
 import (
+	"context"
 	"flag"
 	"fmt"
 	"io"
@@ -58,43 +59,25 @@ const (
 	defaultMultusReadinessIndicatorFile = ""
 )
 
-const (
-	cniConfigDirVarName          = "cni-config-dir"
-	multusAutoconfigDirVarName   = "multus-autoconfig-dir"
-	multusCNIVersion             = "cni-version"
-	multusConfigFileVarName      = "multus-conf-file"
-	multusGlobalNamespaces       = "global-namespaces"
-	multusLogFile                = "multus-log-file"
-	multusLogMaxSize             = "multus-log-max-size"
-	multusLogMaxAge              = "multus-log-max-age"
-	multusLogMaxBackups          = "multus-log-max-backups"
-	multusLogCompress            = "multus-log-compress"
-	multusLogLevel               = "multus-log-level"
-	multusLogToStdErr            = "multus-log-to-stderr"
-	multusMasterCNIFileVarName   = "multus-master-cni-file"
-	multusNamespaceIsolation     = "namespace-isolation"
-	multusReadinessIndicatorFile = "readiness-indicator-file"
-)
-
 func main() {
 	flag.CommandLine = flag.NewFlagSet(os.Args[0], flag.ExitOnError)
 
-	cniConfigDir := flag.String(cniConfigDirVarName, defaultCniConfigDir, "CNI config dir")
-	multusConfigFile := flag.String(multusConfigFileVarName, "auto", "The multus configuration file to use. By default, a new configuration is generated.")
-	multusMasterCni := flag.String(multusMasterCNIFileVarName, "", "The relative name of the configuration file of the cluster primary CNI.")
-	multusAutoconfigDir := flag.String(multusAutoconfigDirVarName, *cniConfigDir, "The directory path for the generated multus configuration.")
-	namespaceIsolation := flag.Bool(multusNamespaceIsolation, false, "If the network resources are only available within their defined namespaces.")
-	globalNamespaces := flag.String(multusGlobalNamespaces, "", "Comma-separated list of namespaces which can be referred to globally when namespace isolation is enabled.")
-	logToStdErr := flag.Bool(multusLogToStdErr, false, "If the multus logs are also to be echoed to stderr.")
-	logLevel := flag.String(multusLogLevel, "", "One of: debug/verbose/error/panic. Used only with --multus-conf-file=auto.")
-	logFile := flag.String(multusLogFile, "", "Path where to multus will log. Used only with --multus-conf-file=auto.")
-	logMaxSize := flag.Int(multusLogMaxSize, defaultMultusLogMaxSize, "the maximum size in megabytes of the log file before it gets rotated")
-	logMaxAge := flag.Int(multusLogMaxAge, defaultMultusLogMaxAge, "the maximum number of days to retain old log files in their filename")
-	logMaxBackups := flag.Int(multusLogMaxBackups, defaultMultusLogMaxBackups, "the maximum number of old log files to retain")
-	logCompress := flag.Bool(multusLogCompress, defaultMultusLogCompress, "compress determines if the rotated log files should be compressed using gzip")
-	cniVersion := flag.String(multusCNIVersion, "", "Allows you to specify CNI spec version. Used only with --multus-conf-file=auto.")
+	cniConfigDir := flag.String("cni-config-dir", defaultCniConfigDir, "CNI config dir")
+	multusConfigFile := flag.String("multus-conf-file", "auto", "The multus configuration file to use. By default, a new configuration is generated.")
+	multusMasterCni := flag.String("multus-master-cni-file", "", "The relative name of the configuration file of the cluster primary CNI.")
+	multusAutoconfigDir := flag.String("multus-autoconfig-dir", *cniConfigDir, "The directory path for the generated multus configuration.")
+	namespaceIsolation := flag.Bool("namespace-isolation", false, "If the network resources are only available within their defined namespaces.")
+	globalNamespaces := flag.String("global-namespaces", "", "Comma-separated list of namespaces which can be referred to globally when namespace isolation is enabled.")
+	logToStdErr := flag.Bool("multus-log-to-stderr", false, "If the multus logs are also to be echoed to stderr.")
+	logLevel := flag.String("multus-log-level", "", "One of: debug/verbose/error/panic. Used only with --multus-conf-file=auto.")
+	logFile := flag.String("multus-log-file", "", "Path where to multus will log. Used only with --multus-conf-file=auto.")
+	logMaxSize := flag.Int("multus-log-max-size", defaultMultusLogMaxSize, "the maximum size in megabytes of the log file before it gets rotated")
+	logMaxAge := flag.Int("multus-log-max-age", defaultMultusLogMaxAge, "the maximum number of days to retain old log files in their filename")
+	logMaxBackups := flag.Int("multus-log-max-backups", defaultMultusLogMaxBackups, "the maximum number of old log files to retain")
+	logCompress := flag.Bool("multus-log-compress", defaultMultusLogCompress, "compress determines if the rotated log files should be compressed using gzip")
+	cniVersion := flag.String("cni-version", "", "Allows you to specify CNI spec version. Used only with --multus-conf-file=auto.")
 	forceCNIVersion := flag.Bool("force-cni-version", false, "force to use given CNI version. only for kind-e2e testing") // this is only for kind-e2e
-	readinessIndicator := flag.String(multusReadinessIndicatorFile, "", "Which file should be used as the readiness indicator. Used only with --multus-conf-file=auto.")
+	readinessIndicator := flag.String("readiness-indicator-file", "", "Which file should be used as the readiness indicator. Used only with --multus-conf-file=auto.")
 	overrideNetworkName := flag.Bool("override-network-name", false, "Used when we need overrides the name of the multus configuration with the name of the delegated primary CNI")
 	version := flag.Bool("version", false, "Show version")
 
@@ -107,7 +90,11 @@ func main() {
 		os.Exit(4)
 	}
 
-	if err := startMultusDaemon(*configFilePath); err != nil {
+	configWatcherStopChannel := make(chan struct{})
+	configWatcherDoneChannel := make(chan struct{})
+	serverStopChannel := make(chan struct{})
+	serverDoneChannel := make(chan struct{})
+	if err := startMultusDaemon(*configFilePath, serverStopChannel, serverDoneChannel); err != nil {
 		logging.Panicf("failed start the multus thick-plugin listener: %v", err)
 		os.Exit(3)
 	}
@@ -206,25 +193,39 @@ func main() {
 			_ = logging.Errorf("failed to persist the multus configuration: %v", err)
 		}
 
-		configWatcherDoneChannel := make(chan struct{})
-		go func(stopChannel chan struct{}, doneChannel chan struct{}) {
-			defer func() {
-				stopChannel <- struct{}{}
-			}()
-			if err := configManager.MonitorPluginConfiguration(configWatcherDoneChannel, stopChannel); err != nil {
+		go func(stopChannel chan<- struct{}, doneChannel chan<- struct{}) {
+			if err := configManager.MonitorPluginConfiguration(configWatcherStopChannel, doneChannel); err != nil {
 				_ = logging.Errorf("error watching file: %v", err)
 			}
-		}(make(chan struct{}), configWatcherDoneChannel)
+		}(configWatcherStopChannel, configWatcherDoneChannel)
 
 		<-configWatcherDoneChannel
 	} else {
 		if err := copyUserProvidedConfig(*multusConfigFile, *cniConfigDir); err != nil {
 			logging.Errorf("failed to copy the user provided configuration %s: %v", *multusConfigFile, err)
 		}
 	}
+
+	serverDone := false
+	configWatcherDone := false
+	for {
+		select {
+		case <-configWatcherDoneChannel:
+			logging.Verbosef("ConfigWatcher done")
+			configWatcherDone = true
+		case <-serverDoneChannel:
+			logging.Verbosef("multus-server done.")
+			serverDone = true
+		}
+
+		if serverDone && configWatcherDone {
+			return
+		}
+	}
+	// never reached
 }
 
-func startMultusDaemon(configFilePath string) error {
+func startMultusDaemon(configFilePath string, stopCh chan struct{}, done chan struct{}) error {
 	daemonConfig, config, err := types.LoadDaemonNetConf(configFilePath)
 	if err != nil {
 		logging.Panicf("failed to load the multus-daemon configuration: %v", err)
@@ -245,11 +246,11 @@ func startMultusDaemon(configFilePath string) error {
 	}
 
 	if daemonConfig.MetricsPort != nil {
-		go utilwait.Forever(func() {
+		go utilwait.Until(func() {
 			http.Handle("/metrics", promhttp.Handler())
 			logging.Debugf("metrics port: %d", *daemonConfig.MetricsPort)
 			logging.Debugf("metrics: %s", http.ListenAndServe(fmt.Sprintf(":%d", *daemonConfig.MetricsPort), nil))
-		}, 0)
+		}, 0, stopCh)
 	}
 
 	l, err := srv.GetListener(api.SocketPath(daemonConfig.MultusSocketDir))
@@ -258,12 +259,16 @@ func startMultusDaemon(configFilePath string) error {
 	}
 
 	server.SetKeepAlivesEnabled(false)
-	go utilwait.Forever(func() {
-		logging.Debugf("open for business")
-		if err := server.Serve(l); err != nil {
-			utilruntime.HandleError(fmt.Errorf("CNI server Serve() failed: %v", err))
-		}
-	}, 0)
+	go func() {
+		utilwait.Until(func() {
+			logging.Debugf("open for business")
+			if err := server.Serve(l); err != nil {
+				utilruntime.HandleError(fmt.Errorf("CNI server Serve() failed: %v", err))
+			}
+		}, 0, stopCh)
+		server.Shutdown(context.TODO())
+		close(done)
+	}()
 
 	return nil
 }
@@ -279,7 +284,7 @@ func copyUserProvidedConfig(multusConfigPath string, cniConfigDir string) error
 	if err != nil {
 		return fmt.Errorf("creating copying file %s: %w", dstFileName, err)
 	}
-	nBytes, err := io.Copy(srcFile, dstFile)
+	nBytes, err := io.Copy(dstFile, srcFile)
 	if err != nil {
 		return fmt.Errorf("error copying file: %w", err)
 	}

go.sum

@@ -229,7 +229,6 @@ github.com/containerd/zfs v1.0.0/go.mod h1:m+m51S1DvAP6r3FcmYCp54bQ34pyOwTieQDNR
 github.com/containernetworking/cni v0.7.1/go.mod h1:LGwApLUm2FpoOfxTDEeq8T9ipbpZ61X79hmU3w8FmsY=
 github.com/containernetworking/cni v0.8.0/go.mod h1:LGwApLUm2FpoOfxTDEeq8T9ipbpZ61X79hmU3w8FmsY=
 github.com/containernetworking/cni v0.8.1/go.mod h1:LGwApLUm2FpoOfxTDEeq8T9ipbpZ61X79hmU3w8FmsY=
-github.com/containernetworking/cni v1.0.1 h1:9OIL/sZmMYDBe+G8svzILAlulUpaDTUjeAbtH/JNLBo=
 github.com/containernetworking/cni v1.0.1/go.mod h1:AKuhXbN5EzmD4yTNtfSsX3tPcmtrBI6QcRV0NiNt15Y=
 github.com/containernetworking/cni v1.1.2 h1:wtRGZVv7olUHMOqouPpn3cXJWpJgM6+EUl31EQbXALQ=
 github.com/containernetworking/cni v1.1.2/go.mod h1:sDpYKmGVENF3s6uvMvGgldDWeG8dMxakj/u+i9ht9vw=
@@ -599,14 +598,14 @@ github.com/onsi/ginkgo v1.14.0/go.mod h1:iSB4RoI2tjJc9BBv4NKIKWKya62Rps+oPG/Lv9k
 github.com/onsi/ginkgo v1.16.4/go.mod h1:dX+/inL/fNMqNlz0e9LfyB9TswhZpCVdJM/Z6Vvnwo0=
 github.com/onsi/ginkgo v1.16.5 h1:8xi0RTUf59SOSfEtZMvwTvXYMzG4gV23XVHOZiXNtnE=
 github.com/onsi/ginkgo v1.16.5/go.mod h1:+E8gABHa3K6zRBolWtd+ROzc/U5bkGt0FwiG042wbpU=
+github.com/onsi/ginkgo/v2 v2.1.3 h1:e/3Cwtogj0HA+25nMP1jCMDIf8RtRYbGwGGuBIFztkc=
 github.com/onsi/ginkgo/v2 v2.1.3/go.mod h1:vw5CSIxN1JObi/U8gcbwft7ZxR2dgaR70JSE3/PpL4c=
 github.com/onsi/gomega v0.0.0-20151007035656-2152b45fa28a/go.mod h1:C1qb7wdrVGGVU+Z6iS04AVkA3Q65CEZX59MT0QO5uiA=
 github.com/onsi/gomega v0.0.0-20170829124025-dcabb60a477c/go.mod h1:C1qb7wdrVGGVU+Z6iS04AVkA3Q65CEZX59MT0QO5uiA=
 github.com/onsi/gomega v1.7.0/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
 github.com/onsi/gomega v1.7.1/go.mod h1:XdKZgCCFLUoM/7CFJVPcG8C1xQ1AJ0vpAezJrB7JYyY=
 github.com/onsi/gomega v1.10.1/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo=
 github.com/onsi/gomega v1.10.3/go.mod h1:V9xEwhxec5O8UDM77eCW8vLymOMltsqPVYWrpDsH8xc=
-github.com/onsi/gomega v1.15.0 h1:WjP/FQ/sk43MRmnEcT+MlDw2TFvkrXlprrPST/IudjU=
 github.com/onsi/gomega v1.15.0/go.mod h1:cIuvLEne0aoVhAgh/O6ac0Op8WWw9H6eYCriF+tEHG0=
 github.com/onsi/gomega v1.17.0 h1:9Luw4uT5HTjHTN8+aNcSThgH1vdXnmdJ8xIfZ4wyTRE=
 github.com/onsi/gomega v1.17.0/go.mod h1:HnhC7FXeEQY45zxNK3PPoIUhzk/80Xly9PcubAlGdZY=

pkg/server/config/manager.go

@@ -166,7 +166,7 @@ func (m Manager) GenerateConfig() (string, error) {
 // MonitorPluginConfiguration monitors the configuration file pointed
 // to by the primaryCNIPluginName attribute, and re-generates the multus
 // configuration whenever the primary CNI config is updated.
-func (m Manager) MonitorPluginConfiguration(shutDown chan struct{}, done chan struct{}) error {
+func (m Manager) MonitorPluginConfiguration(shutDown <-chan struct{}, done chan<- struct{}) error {
 	logging.Verbosef("started to watch file %s", m.primaryCNIConfigPath)
 
 	for {
@@ -206,7 +206,7 @@ func (m Manager) MonitorPluginConfiguration(shutDown chan struct{}, done chan st
 		case <-shutDown:
 			logging.Verbosef("Stopped monitoring, closing channel ...")
 			_ = m.configWatcher.Close()
-			done <- struct{}{}
+			close(done)
 			return nil
 		}
 	}

pkg/server/config/manager_test.go

@@ -98,9 +98,6 @@ var _ = Describe("Configuration Manager", func() {
 
 		configWatcherDoneChannel := make(chan struct{})
 		go func(stopChannel chan struct{}, doneChannel chan struct{}) {
-			defer func() {
-				stopChannel <- struct{}{}
-			}()
 			err := configManager.MonitorPluginConfiguration(configWatcherDoneChannel, stopChannel)
 			Expect(err).NotTo(HaveOccurred())
 		}(make(chan struct{}), configWatcherDoneChannel)