Merge pull request #3296 from sivchari/fix-goroutine-leak

fix goroutine leak

Author: k8s-ci-robot

kinder/pkg/cluster/manager/actions/kubeadm-init.go

@@ -18,6 +18,7 @@ package actions
 
 import (
 	"bytes"
+	"context"
 	"fmt"
 	"log"
 	"net"
@@ -226,7 +227,7 @@ func postInit(c *status.Cluster, wait time.Duration) error {
 	//	return errors.Wrap(err, "failed to add default storage class")
 	//}
 
-	if err := waitNewControlPlaneNodeReady(c, cp1, wait); err != nil {
+	if err := waitNewControlPlaneNodeReady(context.Background(), c, cp1, wait); err != nil {
 		return err
 	}
 

kinder/pkg/cluster/manager/actions/kubeadm-join.go

@@ -17,6 +17,7 @@ limitations under the License.
 package actions
 
 import (
+	"context"
 	"fmt"
 	"time"
 
@@ -83,7 +84,7 @@ func joinControlPlanes(c *status.Cluster, usePhases bool, copyCertsMode CopyCert
 			return err
 		}
 
-		if err := waitNewControlPlaneNodeReady(c, cp2, wait); err != nil {
+		if err := waitNewControlPlaneNodeReady(context.Background(), c, cp2, wait); err != nil {
 			return err
 		}
 	}
@@ -189,7 +190,7 @@ func joinWorkers(c *status.Cluster, usePhases bool, discoveryMode DiscoveryMode,
 			return err
 		}
 
-		if err := waitNewWorkerNodeReady(c, w, wait); err != nil {
+		if err := waitNewWorkerNodeReady(context.Background(), c, w, wait); err != nil {
 			return err
 		}
 	}

kinder/pkg/cluster/manager/actions/kubeadm-upgrade.go

@@ -17,6 +17,7 @@ limitations under the License.
 package actions
 
 import (
+	"context"
 	"fmt"
 	"path/filepath"
 	"time"
@@ -220,7 +221,7 @@ func kubeadmUpgradeApply(c *status.Cluster, cp1 *status.Node, configVersion stri
 		return err
 	}
 
-	if err := waitControlPlaneUpgraded(c, cp1, upgradeVersion, wait); err != nil {
+	if err := waitControlPlaneUpgraded(context.Background(), c, cp1, upgradeVersion, wait); err != nil {
 		return err
 	}
 
@@ -231,7 +232,7 @@ func kubeadmUpgradeNode(c *status.Cluster, n *status.Node, configVersion string,
 	// waitKubeletHasRBAC waits for the kubelet to have access to the expected config map
 	// please note that this is a temporary workaround for a problem we are observing on upgrades while
 	// executing node upgrades immediately after control-plane upgrade.
-	if err := waitKubeletHasRBAC(c, n, upgradeVersion, wait); err != nil {
+	if err := waitKubeletHasRBAC(context.Background(), c, n, upgradeVersion, wait); err != nil {
 		return err
 	}
 
@@ -255,7 +256,7 @@ func kubeadmUpgradeNode(c *status.Cluster, n *status.Node, configVersion string,
 	}
 
 	if n.IsControlPlane() {
-		if err := waitControlPlaneUpgraded(c, n, upgradeVersion, wait); err != nil {
+		if err := waitControlPlaneUpgraded(context.Background(), c, n, upgradeVersion, wait); err != nil {
 			return err
 		}
 	}
@@ -302,7 +303,7 @@ func upgradeKubeletKubectl(c *status.Cluster, n *status.Node, upgradeVersion *ve
 		return err
 	}
 
-	if err := waitKubeletUpgraded(c, n, upgradeVersion, wait); err != nil {
+	if err := waitKubeletUpgraded(context.Background(), c, n, upgradeVersion, wait); err != nil {
 		return err
 	}
 

kinder/pkg/cluster/manager/actions/smoke-test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package actions
 
 import (
+	"context"
 	"fmt"
 	"strings"
 	"time"
@@ -46,7 +47,7 @@ func SmokeTest(c *status.Cluster, wait time.Duration) error {
 		return err
 	}
 
-	if err := waitForPodsRunning(c, cp1, wait, "nginx", 1); err != nil {
+	if err := waitForPodsRunning(context.Background(), c, cp1, wait, "nginx", 1); err != nil {
 		return err
 	}
 
@@ -67,7 +68,7 @@ func SmokeTest(c *status.Cluster, wait time.Duration) error {
 	}
 
 	for _, n := range c.K8sNodes() {
-		err = waitForNodePort(c, n, 30*time.Second, nodePort)
+		err = waitForNodePort(context.Background(), c, n, 30*time.Second, nodePort)
 		if err != nil {
 			return err
 		}

kinder/pkg/cluster/manager/actions/waiter.go

@@ -17,6 +17,7 @@ limitations under the License.
 package actions
 
 import (
+	"context"
 	"fmt"
 	"math/rand"
 	"regexp"
@@ -30,9 +31,9 @@ import (
 )
 
 // waitNewControlPlaneNodeReady waits for a new control plane node reaching the target state after init/join
-func waitNewControlPlaneNodeReady(c *status.Cluster, n *status.Node, wait time.Duration) error {
+func waitNewControlPlaneNodeReady(ctx context.Context, c *status.Cluster, n *status.Node, wait time.Duration) error {
 	n.Infof("waiting for Node and control-plane Pods to become Ready (timeout %s)", wait)
-	if pass := waitFor(c, n, wait,
+	if pass := waitFor(ctx, c, n, wait,
 		nodeIsReady,
 		staticPodIsReady("kube-apiserver"),
 		staticPodIsReady("kube-controller-manager"),
@@ -44,8 +45,8 @@ func waitNewControlPlaneNodeReady(c *status.Cluster, n *status.Node, wait time.D
 	return nil
 }
 
-func waitForPodsRunning(c *status.Cluster, n *status.Node, wait time.Duration, label string, replicas int) error {
-	if pass := waitFor(c, n, wait,
+func waitForPodsRunning(ctx context.Context, c *status.Cluster, n *status.Node, wait time.Duration, label string, replicas int) error {
+	if pass := waitFor(ctx, c, n, wait,
 		podsAreRunning(n, label, replicas),
 	); !pass {
 		return errors.New("timeout: Node and control-plane did not reach target state")
@@ -55,9 +56,9 @@ func waitForPodsRunning(c *status.Cluster, n *status.Node, wait time.Duration, l
 }
 
 // waitForNodePort waits for a nodePort to become ready
-func waitForNodePort(c *status.Cluster, n *status.Node, wait time.Duration, nodePort string) error {
+func waitForNodePort(ctx context.Context, c *status.Cluster, n *status.Node, wait time.Duration, nodePort string) error {
 	n.Infof("waiting for NodePort %q to become ready (timeout %s)", nodePort, wait)
-	if pass := waitFor(c, n, wait,
+	if pass := waitFor(ctx, c, n, wait,
 		nodePortIsReady(n, nodePort),
 	); !pass {
 		return errors.New("timeout: NodePort not ready")
@@ -67,9 +68,9 @@ func waitForNodePort(c *status.Cluster, n *status.Node, wait time.Duration, node
 }
 
 // waitNewWorkerNodeReady waits for a new control plane node reaching the target state after join
-func waitNewWorkerNodeReady(c *status.Cluster, n *status.Node, wait time.Duration) error {
+func waitNewWorkerNodeReady(ctx context.Context, c *status.Cluster, n *status.Node, wait time.Duration) error {
 	n.Infof("waiting for Node to become Ready (timeout %s)", wait)
-	if pass := waitFor(c, n, wait,
+	if pass := waitFor(ctx, c, n, wait,
 		nodeIsReady,
 	); !pass {
 		return errors.New("timeout: Node did not reach target state")
@@ -79,11 +80,11 @@ func waitNewWorkerNodeReady(c *status.Cluster, n *status.Node, wait time.Duratio
 }
 
 // waitControlPlaneUpgraded waits for a control plane node reaching the target state after upgrade
-func waitControlPlaneUpgraded(c *status.Cluster, n *status.Node, upgradeVersion *K8sVersion.Version, wait time.Duration) error {
+func waitControlPlaneUpgraded(ctx context.Context, c *status.Cluster, n *status.Node, upgradeVersion *K8sVersion.Version, wait time.Duration) error {
 	version := kubernetesVersionToImageTag(upgradeVersion.String())
 
 	n.Infof("waiting for control-plane Pods to restart with the new version (timeout %s)", wait)
-	if pass := waitFor(c, n, wait,
+	if pass := waitFor(ctx, c, n, wait,
 		staticPodHasVersion("kube-apiserver", version),
 		staticPodHasVersion("kube-controller-manager", version),
 		staticPodHasVersion("kube-scheduler", version),
@@ -95,11 +96,11 @@ func waitControlPlaneUpgraded(c *status.Cluster, n *status.Node, upgradeVersion
 }
 
 // waitKubeletUpgraded waits for a node reaching the target state after upgrade
-func waitKubeletUpgraded(c *status.Cluster, n *status.Node, upgradeVersion *K8sVersion.Version, wait time.Duration) error {
+func waitKubeletUpgraded(ctx context.Context, c *status.Cluster, n *status.Node, upgradeVersion *K8sVersion.Version, wait time.Duration) error {
 	version := upgradeVersion.String()
 
 	n.Infof("waiting for node to restart with the new version (timeout %s)", wait)
-	if pass := waitFor(c, n, wait,
+	if pass := waitFor(ctx, c, n, wait,
 		nodeHasKubernetesVersion(version),
 	); !pass {
 		return errors.New("timeout: node did not reach target state")
@@ -111,9 +112,9 @@ func waitKubeletUpgraded(c *status.Cluster, n *status.Node, upgradeVersion *K8sV
 // waitKubeletHasRBAC waits for the kubelet to have access to the expected config map
 // please note that this is a temporary workaround for a problem we are observing on upgrades while
 // executing node upgrades immediately after control-plane upgrade.
-func waitKubeletHasRBAC(c *status.Cluster, n *status.Node, upgradeVersion *K8sVersion.Version, wait time.Duration) error {
+func waitKubeletHasRBAC(ctx context.Context, c *status.Cluster, n *status.Node, upgradeVersion *K8sVersion.Version, wait time.Duration) error {
 	n.Infof("waiting for kubelet RBAC validation - workaround (timeout %s)", wait)
-	if pass := waitFor(c, n, wait,
+	if pass := waitFor(ctx, c, n, wait,
 		kubeletHasRBAC(upgradeVersion.Major(), upgradeVersion.Minor()),
 	); !pass {
 		return errors.New("timeout: Node did not reach target state")
@@ -127,18 +128,18 @@ type try func(*status.Cluster, *status.Node) bool
 
 // waitFor implements the waiter core logic that is responsible for testing all the given contitions
 // until are satisfied or a timeout are reached
-func waitFor(c *status.Cluster, n *status.Node, timeout time.Duration, conditions ...try) bool {
+func waitFor(ctx context.Context, c *status.Cluster, n *status.Node, timeout time.Duration, conditions ...try) bool {
 	// if timeout is 0 or no conditions are defined, exit fast
 	if timeout == time.Duration(0) {
 		fmt.Println("Timeout set 0, skipping wait")
 		return true
 	}
 
-	// sets the timeout timer
-	timer := time.NewTimer(timeout)
+	ctx, cancel := context.WithTimeout(ctx, timeout)
+	defer cancel()
 
 	// runs all the conditions in parallel
-	pass := make(chan bool)
+	pass := make(chan struct{}, len(conditions))
 	for _, wc := range conditions {
 		// clone the condition func to make the closure point to right value
 		// even after the for loop moves to the next condition
@@ -147,15 +148,33 @@ func waitFor(c *status.Cluster, n *status.Node, timeout time.Duration, condition
 		// run the condition in a go routine until it pass
 		go func() {
 			// creates an arbitrary skew before starting a wait loop
-			time.Sleep(time.Duration(rand.Intn(500)) * time.Millisecond)
+			timer := time.NewTimer(time.Duration(rand.Intn(500)) * time.Millisecond)
+			defer timer.Stop()
+
+			select {
+			case <-ctx.Done():
+				return
+			case <-timer.C:
+			}
 
 			for {
+				select {
+				case <-ctx.Done():
+					return
+				default:
+				}
+
 				if x(c, n) {
-					<-pass
+					pass <- struct{}{}
 					break
 				}
 				// add a little delay + jitter before retry
-				time.Sleep(1*time.Second + time.Duration(rand.Intn(500))*time.Millisecond)
+				timer.Reset(1*time.Second + time.Duration(rand.Intn(500))*time.Millisecond)
+				select {
+				case <-ctx.Done():
+					return
+				case <-timer.C:
+				}
 			}
 		}()
 	}
@@ -164,14 +183,12 @@ func waitFor(c *status.Cluster, n *status.Node, timeout time.Duration, condition
 	passed := 0
 	for {
 		select {
-		case pass <- true:
+		case <-pass:
 			passed++
 			if passed == len(conditions) {
 				return true
 			}
-		case <-timer.C:
-			// close the channel if timeout occurs, this will release all the blocked receives
-			close(pass)
+		case <-ctx.Done():
 			return false
 		}
 	}