feat: don't use time.Sleep and allow context cancellation (#1269)

Author: pmalek

internal/utils/cluster_init.go

@@ -28,6 +28,9 @@ const (
 // regardless of which cluster type (e.g. kind, gke). This includes the creation
 // of some special administrative namespaces and service accounts.
 func ClusterInitHooks(ctx context.Context, cluster clusters.Cluster) error {
+	const (
+		serviceAccountWaitTime = time.Second
+	)
 	// create the admin namespace if it doesn't already exist
 	namespace, err := cluster.Client().CoreV1().Namespaces().Create(ctx, &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: AdminNamespace}}, metav1.CreateOptions{})
 	if err != nil {
@@ -39,6 +42,7 @@ func ClusterInitHooks(ctx context.Context, cluster clusters.Cluster) error {
 	// wait for the default service account to be available
 	var defaultSAFound bool
 	var defaultSA *corev1.ServiceAccount
+
 	for !defaultSAFound {
 		select {
 		case <-ctx.Done():
@@ -47,8 +51,12 @@ func ClusterInitHooks(ctx context.Context, cluster clusters.Cluster) error {
 			defaultSA, err = cluster.Client().CoreV1().ServiceAccounts(namespace.Name).Get(ctx, "default", metav1.GetOptions{})
 			if err != nil {
 				if errors.IsNotFound(err) {
-					time.Sleep(time.Second)
-					continue // try again if its not there yet
+					select {
+					case <-time.After(serviceAccountWaitTime):
+						continue // try again if its not there yet
+					case <-ctx.Done():
+						return fmt.Errorf("context completed before cluster init hooks could finish: %w", ctx.Err())
+					}
 				}
 				return err // don't tolerate any errors except 404
 			}

pkg/clusters/addons/istio/addon.go

@@ -74,6 +74,10 @@ func (a *Addon) Version() semver.Version {
 // EnableMeshForNamespace will add the "istio-injection=enabled" label to the provided namespace
 // by name which will indicate to Istio to inject sidecard pods to add it to the mesh network.
 func (a *Addon) EnableMeshForNamespace(ctx context.Context, cluster clusters.Cluster, name string) error {
+	const (
+		namspaceWaitTime = time.Second
+	)
+
 	for {
 		select {
 		case <-ctx.Done():
@@ -89,8 +93,12 @@ func (a *Addon) EnableMeshForNamespace(ctx context.Context, cluster clusters.Clu
 				if errors.IsConflict(err) {
 					// if there's a conflict then an update happened since we pulled the namespace,
 					// simply pull and try again.
-					time.Sleep(time.Second)
-					continue
+					select {
+					case <-time.After(namspaceWaitTime):
+						continue // try again if its not there yet
+					case <-ctx.Done():
+						continue // this will return an error in the next iteration
+					}
 				}
 				return fmt.Errorf("could not enable mesh for namespace %s: %w", name, err)
 			}
@@ -141,6 +149,8 @@ func (a *Addon) Deploy(ctx context.Context, cluster clusters.Cluster) error {
 		return err
 	}
 
+	const jobWaitTime = time.Second
+
 	// wait for the job to complete
 	var deployJobCompleted bool
 	for !deployJobCompleted {
@@ -155,7 +165,12 @@ func (a *Addon) Deploy(ctx context.Context, cluster clusters.Cluster) error {
 			if a.istioDeployJob.Status.Succeeded > 0 {
 				deployJobCompleted = true
 			} else {
-				time.Sleep(time.Second)
+				select {
+				case <-time.After(jobWaitTime):
+					continue
+				case <-ctx.Done():
+					continue // this will return an error in the next iteration
+				}
 			}
 		}
 	}
@@ -302,8 +317,8 @@ func (a *Addon) deployExtras(ctx context.Context, cluster clusters.Cluster) erro
 }
 
 const (
-	defaultRetries          = 3
-	defaultRetryWaitSeconds = 3
+	defaultRetries       = 3
+	defaultRetryWaitTime = 3 * time.Second
 )
 
 // retryKubectlApply retries a command multiple times with a limit, and is particularly
@@ -319,8 +334,12 @@ func retryKubectlApply(ctx context.Context, args ...string) (err error) {
 		if err = cmd.Run(); err == nil {
 			break
 		}
-		time.Sleep(time.Second * defaultRetryWaitSeconds)
-		count--
+		select {
+		case <-ctx.Done():
+			continue
+		case <-time.After(defaultRetryWaitTime):
+			count--
+		}
 	}
 
 	if err != nil {

pkg/clusters/addons/knative/knative.go

@@ -214,6 +214,8 @@ func deleteKnative(ctx context.Context, cluster clusters.Cluster, version string
 		}
 	}
 
+	const namespaceWaitTime = time.Second
+
 	// wait for the namespace to tear down
 	for {
 		select {
@@ -229,7 +231,12 @@ func deleteKnative(ctx context.Context, cluster clusters.Cluster, version string
 				}
 				return err
 			}
-			time.Sleep(time.Second)
+
+			select {
+			case <-ctx.Done():
+				continue // this will return an error in the next iteration
+			case <-time.After(namespaceWaitTime):
+			}
 		}
 	}
 }

pkg/clusters/addons/kuma/addon.go

@@ -79,6 +79,8 @@ func (a *Addon) Version() (v semver.Version, ok bool) {
 // EnableMeshForNamespace will add the "kuma.io/sidecar-injection: enabled" label to the provided namespace,
 // enabling sidecar injections fo all Pods in the namespace
 func EnableMeshForNamespace(ctx context.Context, cluster clusters.Cluster, name string) error {
+	const namespaceWaitTime = time.Second
+
 	for {
 		select {
 		case <-ctx.Done():
@@ -94,7 +96,11 @@ func EnableMeshForNamespace(ctx context.Context, cluster clusters.Cluster, name
 				if errors.IsConflict(err) {
 					// if there's a conflict then an update happened since we pulled the namespace,
 					// simply pull and try again.
-					time.Sleep(time.Second)
+					select {
+					case <-ctx.Done():
+						continue // this will return an error in the next iteration
+					case <-time.After(namespaceWaitTime):
+					}
 					continue
 				}
 				return fmt.Errorf("could not enable mesh for namespace %s: %w", name, err)

pkg/clusters/types/gke/builder.go

@@ -232,7 +232,12 @@ func (b *Builder) Build(ctx context.Context) (clusters.Cluster, error) {
 				clusterReady = true
 				break
 			}
-			time.Sleep(waitForClusterTick)
+
+			select {
+			case <-ctx.Done():
+				continue // this will return an error in the next iteration
+			case <-time.After(waitForClusterTick):
+			}
 		}
 	}
 

pkg/environments/implementation.go

@@ -123,7 +123,13 @@ func (env *environment) WaitForReady(ctx context.Context) chan error {
 					hung.Stop()
 					return
 				}
-				time.Sleep(objectWaitSleepTime)
+				// Wait before retry to prevent spamming env with readiness check.
+				select {
+				case <-time.After(objectWaitSleepTime):
+				case <-ctx.Done():
+					hung.Stop()
+					return
+				}
 			}
 		}
 	}()