Merge pull request #1619 from luthermonson/refactor-win-calico

Refactored how we start calico with kubelet context to cancel

pkg/pebinaryexecutor/pebinary.go

@@ -145,15 +145,24 @@ func (p *PEBinaryConfig) Kubelet(args []string) error {
 	logrus.Infof("Running RKE2 kubelet %v", cleanArgs)
 	go func() {
 		for {
+			ctx, cancel := context.WithCancel(context.Background())
+			go func() {
+				if err := p.cni.Start(ctx, p.cniConig); err != nil {
+					logrus.Errorf("error in cni start: %s", err)
+				}
+			}()
+
 			cmd := exec.Command(p.KubeletPath, cleanArgs...)
 			cmd.Stdout = os.Stdout
 			cmd.Stderr = os.Stderr
-			err := cmd.Run()
-			logrus.Errorf("Kubelet exited: %v", err)
+			if err := cmd.Run(); err != nil {
+				logrus.Errorf("Kubelet exited: %v", err)
+			}
+			cancel()
 			time.Sleep(5 * time.Second)
 		}
 	}()
-	return p.cni.Start(p.cniConig)
+	return nil
 }
 
 // KubeProxy starts the kubeproxy in a subprocess with watching goroutine.
@@ -173,13 +182,14 @@ func (p *PEBinaryConfig) KubeProxy(args []string) error {
 		extraArgs["enable-dsr"] = "true"
 	}
 
+	var vip string
 	for range time.Tick(time.Second * 5) {
 		endpoint, err := hcsshim.GetHNSEndpointByName("Calico_ep")
 		if err != nil {
 			logrus.WithError(err).Warningf("can't find %s, retrying", "Calico_ep")
 			continue
 		}
-		extraArgs["source-vip"] = endpoint.IPAddress.String()
+		vip = endpoint.IPAddress.String()
 		break
 	}
 
@@ -190,6 +200,13 @@ func (p *PEBinaryConfig) KubeProxy(args []string) error {
 	}
 
 	args = append(getArgs(extraArgs), args...)
+
+	for i, arg := range args {
+		if strings.Contains(arg, "source-vip") {
+			args[i] = "--source-vip=" + vip
+		}
+	}
+
 	logrus.Infof("Running RKE2 kube-proxy %s", args)
 	go func() {
 		for {

pkg/windows/calico.go

@@ -106,24 +106,25 @@ func (c *Calico) Setup(ctx context.Context, dataDir string, nodeConfig *daemonco
 	if err := createCNIConfig(cfg); err != nil {
 		return nil, err
 	}
+
+	logrus.Info("Generating HNS networks, please wait")
+	if err := generateCalicoNetworks(cfg.CalicoConfig.NetworkingBackend); err != nil {
+		return nil, err
+	}
+
 	return cfg, nil
 }
 
 // Start starts the CNI services on the Windows node.
-func (c *Calico) Start(config *CNIConfig) error {
-	if err := generateCalicoNetworks(config.CalicoConfig.NetworkingBackend); err != nil {
-		return err
-	}
-
-	if err := startCalico(config.CalicoConfig); err != nil {
-		return err
+func (c *Calico) Start(ctx context.Context, config *CNIConfig) error {
+	for {
+		if err := startCalico(ctx, config.CalicoConfig); err != nil {
+			continue
+		}
+		break
 	}
+	go startFelix(ctx, config.CalicoConfig)
 
-	time.Sleep(5 * time.Second)
-
-	if err := startFelix(config.CalicoConfig); err != nil {
-		return err
-	}
 	return nil
 }
 
@@ -190,7 +191,7 @@ func getDefaultConfig(config *CNIConfig, dataDir string, nodeConfig *config.Node
 	return nil
 }
 
-func startFelix(config *CalicoConfig) error {
+func startFelix(ctx context.Context, config *CalicoConfig) {
 	specificEnvs := []string{
 		fmt.Sprintf("FELIX_FELIXHOSTNAME=%s", config.Hostname),
 		fmt.Sprintf("FELIX_VXLANVNI=%s", config.Felix.Vxlanvni),
@@ -202,21 +203,16 @@ func startFelix(config *CalicoConfig) error {
 	}
 
 	logrus.Infof("Felix Envs: ", append(generateGeneralCalicoEnvs(config), specificEnvs...))
-	go func() {
-		for {
-			cmd := exec.Command("calico-node.exe", args...)
-			cmd.Env = append(generateGeneralCalicoEnvs(config), specificEnvs...)
-			cmd.Stdout = os.Stdout
-			cmd.Stderr = os.Stderr
-			err := cmd.Run()
-			logrus.Errorf("Felix exited: %v", err)
-			time.Sleep(5 * time.Second)
-		}
-	}()
-	return nil
+	cmd := exec.CommandContext(ctx, "calico-node.exe", args...)
+	cmd.Env = append(generateGeneralCalicoEnvs(config), specificEnvs...)
+	cmd.Stdout = os.Stdout
+	cmd.Stderr = os.Stderr
+	if err := cmd.Run(); err != nil {
+		logrus.Errorf("Felix exited: %v", err)
+	}
 }
 
-func startCalico(config *CalicoConfig) error {
+func startCalico(ctx context.Context, config *CalicoConfig) error {
 	specificEnvs := []string{
 		fmt.Sprintf("CALICO_NODENAME_FILE=%s", config.NodeNameFile),
 	}
@@ -225,17 +221,14 @@ func startCalico(config *CalicoConfig) error {
 		"-startup",
 	}
 	logrus.Infof("Calico Envs: ", append(generateGeneralCalicoEnvs(config), specificEnvs...))
-	go func() {
-		for {
-			cmd := exec.Command("calico-node.exe", args...)
-			cmd.Env = append(generateGeneralCalicoEnvs(config), specificEnvs...)
-			cmd.Stdout = os.Stdout
-			cmd.Stderr = os.Stderr
-			err := cmd.Run()
-			logrus.Errorf("Calico exited: %v", err)
-			time.Sleep(5 * time.Second)
-		}
-	}()
+	cmd := exec.CommandContext(ctx, "calico-node.exe", args...)
+	cmd.Env = append(generateGeneralCalicoEnvs(config), specificEnvs...)
+	cmd.Stdout = os.Stdout
+	cmd.Stderr = os.Stderr
+	if err := cmd.Run(); err != nil {
+		logrus.Errorf("Calico exited: %v", err)
+		return err
+	}
 	return nil
 }
 
@@ -309,7 +302,6 @@ func generateCalicoNetworks(backend string) error {
 
 	logrus.Debug("Waiting for management ip..")
 	mgmt := waitForManagementIP(CalicoHnsNetworkName)
-	time.Sleep(90 * time.Second)
 	platform, err := getPlatformType()
 	if err != nil {
 		return err

pkg/windows/types.go

@@ -11,8 +11,8 @@ import (
 )
 
 type CNI interface {
-	Setup(ctx context.Context, dataDir string, nodeConfig *daemonconfig.Node, restConfig *rest.Config) (*CNIConfig, error)
-	Start(config *CNIConfig) error
+	Setup(context.Context, string, *daemonconfig.Node, *rest.Config) (*CNIConfig, error)
+	Start(context.Context, *CNIConfig) error
 }
 
 type CNIConfig struct {