Merge pull request #9437 from nojnhuh/clusterctl-ctx

🌱 propagate ctx to retryWithExponentialBackoff in clusterctl

cmd/clusterctl/client/cluster/cert_manager.go

@@ -187,7 +187,7 @@ func (cm *certManagerClient) install(ctx context.Context) error {
 		o := objs[i]
 		// Create the Kubernetes object.
 		// Nb. The operation is wrapped in a retry loop to make ensureCerts more resilient to unexpected conditions.
-		if err := retryWithExponentialBackoff(createCertManagerBackoff, func() error {
+		if err := retryWithExponentialBackoff(ctx, createCertManagerBackoff, func(ctx context.Context) error {
 			return cm.createObj(ctx, o)
 		}); err != nil {
 			return err
@@ -306,7 +306,7 @@ func (cm *certManagerClient) deleteObjs(ctx context.Context, objs []unstructured
 			continue
 		}
 
-		if err := retryWithExponentialBackoff(deleteCertManagerBackoff, func() error {
+		if err := retryWithExponentialBackoff(ctx, deleteCertManagerBackoff, func(ctx context.Context) error {
 			if err := cm.deleteObj(ctx, obj); err != nil {
 				// tolerate NotFound errors when deleting the test resources
 				if apierrors.IsNotFound(err) {
@@ -560,7 +560,7 @@ func (cm *certManagerClient) waitForAPIReady(ctx context.Context, retry bool) er
 	deleteCertManagerBackoff := newWriteBackoff()
 	for i := range testObjs {
 		obj := testObjs[i]
-		if err := retryWithExponentialBackoff(deleteCertManagerBackoff, func() error {
+		if err := retryWithExponentialBackoff(ctx, deleteCertManagerBackoff, func(ctx context.Context) error {
 			if err := cm.deleteObj(ctx, obj); err != nil {
 				// tolerate NotFound errors when deleting the test resources
 				if apierrors.IsNotFound(err) {

cmd/clusterctl/client/cluster/client.go

@@ -219,13 +219,13 @@ func newClusterClient(kubeconfig Kubeconfig, configClient config.Client, options
 }
 
 // retryWithExponentialBackoff repeats an operation until it passes or the exponential backoff times out.
-func retryWithExponentialBackoff(opts wait.Backoff, operation func() error) error {
+func retryWithExponentialBackoff(ctx context.Context, opts wait.Backoff, operation func(ctx context.Context) error) error {
 	log := logf.Log
 
 	i := 0
-	err := wait.ExponentialBackoff(opts, func() (bool, error) {
+	err := wait.ExponentialBackoffWithContext(ctx, opts, func(ctx context.Context) (bool, error) {
 		i++
-		if err := operation(); err != nil {
+		if err := operation(ctx); err != nil {
 			if i < opts.Steps {
 				log.V(5).Info("Retrying with backoff", "Cause", err.Error())
 				return false, nil

cmd/clusterctl/client/cluster/components.go

@@ -80,7 +80,7 @@ func (p *providerComponents) Create(ctx context.Context, objs []unstructured.Uns
 
 		// Create the Kubernetes object.
 		// Nb. The operation is wrapped in a retry loop to make Create more resilient to unexpected conditions.
-		if err := retryWithExponentialBackoff(createComponentObjectBackoff, func() error {
+		if err := retryWithExponentialBackoff(ctx, createComponentObjectBackoff, func(ctx context.Context) error {
 			return p.createObj(ctx, obj)
 		}); err != nil {
 			return err
@@ -217,7 +217,7 @@ func (p *providerComponents) Delete(ctx context.Context, options DeleteOptions)
 		// Otherwise delete the object
 		log.V(5).Info("Deleting", logf.UnstructuredToValues(obj)...)
 		deleteBackoff := newWriteBackoff()
-		if err := retryWithExponentialBackoff(deleteBackoff, func() error {
+		if err := retryWithExponentialBackoff(ctx, deleteBackoff, func(ctx context.Context) error {
 			if err := cs.Delete(ctx, &obj); err != nil {
 				if apierrors.IsNotFound(err) {
 					// Tolerate IsNotFound error that might happen because we are not enforcing a deletion order

cmd/clusterctl/client/cluster/crd_migration.go

@@ -88,7 +88,7 @@ func (m *crdMigrator) run(ctx context.Context, newCRD *apiextensionsv1.CustomRes
 
 	// Get the current CRD.
 	currentCRD := &apiextensionsv1.CustomResourceDefinition{}
-	if err := retryWithExponentialBackoff(newReadBackoff(), func() error {
+	if err := retryWithExponentialBackoff(ctx, newReadBackoff(), func(ctx context.Context) error {
 		return m.Client.Get(ctx, client.ObjectKeyFromObject(newCRD), currentCRD)
 	}); err != nil {
 		// Return if the CRD doesn't exist yet. We only have to migrate if the CRD exists already.
@@ -152,7 +152,7 @@ func (m *crdMigrator) migrateResourcesForCRD(ctx context.Context, crd *apiextens
 
 	var i int
 	for {
-		if err := retryWithExponentialBackoff(newReadBackoff(), func() error {
+		if err := retryWithExponentialBackoff(ctx, newReadBackoff(), func(ctx context.Context) error {
 			return m.Client.List(ctx, list, client.Continue(list.GetContinue()))
 		}); err != nil {
 			return errors.Wrapf(err, "failed to list %q", list.GetKind())
@@ -162,7 +162,7 @@ func (m *crdMigrator) migrateResourcesForCRD(ctx context.Context, crd *apiextens
 			obj := list.Items[i]
 
 			log.V(5).Info("Migrating", logf.UnstructuredToValues(obj)...)
-			if err := retryWithExponentialBackoff(newWriteBackoff(), func() error {
+			if err := retryWithExponentialBackoff(ctx, newWriteBackoff(), func(ctx context.Context) error {
 				return handleMigrateErr(m.Client.Update(ctx, &obj))
 			}); err != nil {
 				return errors.Wrapf(err, "failed to migrate %s/%s", obj.GetNamespace(), obj.GetName())
@@ -187,7 +187,7 @@ func (m *crdMigrator) migrateResourcesForCRD(ctx context.Context, crd *apiextens
 
 func (m *crdMigrator) patchCRDStoredVersions(ctx context.Context, crd *apiextensionsv1.CustomResourceDefinition, currentStorageVersion string) error {
 	crd.Status.StoredVersions = []string{currentStorageVersion}
-	if err := retryWithExponentialBackoff(newWriteBackoff(), func() error {
+	if err := retryWithExponentialBackoff(ctx, newWriteBackoff(), func(ctx context.Context) error {
 		return m.Client.Status().Update(ctx, crd)
 	}); err != nil {
 		return errors.Wrapf(err, "failed to update status.storedVersions for CRD %q", crd.Name)

cmd/clusterctl/client/cluster/inventory.go

@@ -162,7 +162,7 @@ func (p *inventoryClient) EnsureCustomResourceDefinitions(ctx context.Context) e
 	// Nb. The operation is wrapped in a retry loop to make EnsureCustomResourceDefinitions more resilient to unexpected conditions.
 	var crdIsIstalled bool
 	listInventoryBackoff := newReadBackoff()
-	if err := retryWithExponentialBackoff(listInventoryBackoff, func() error {
+	if err := retryWithExponentialBackoff(ctx, listInventoryBackoff, func(ctx context.Context) error {
 		var err error
 		crdIsIstalled, err = checkInventoryCRDs(ctx, p.proxy)
 		return err
@@ -189,7 +189,7 @@ func (p *inventoryClient) EnsureCustomResourceDefinitions(ctx context.Context) e
 
 		// Create the Kubernetes object.
 		// Nb. The operation is wrapped in a retry loop to make EnsureCustomResourceDefinitions more resilient to unexpected conditions.
-		if err := retryWithExponentialBackoff(createInventoryObjectBackoff, func() error {
+		if err := retryWithExponentialBackoff(ctx, createInventoryObjectBackoff, func(ctx context.Context) error {
 			return p.createObj(ctx, o)
 		}); err != nil {
 			return err
@@ -272,7 +272,7 @@ func (p *inventoryClient) createObj(ctx context.Context, o unstructured.Unstruct
 func (p *inventoryClient) Create(ctx context.Context, m clusterctlv1.Provider) error {
 	// Create the Kubernetes object.
 	createInventoryObjectBackoff := newWriteBackoff()
-	return retryWithExponentialBackoff(createInventoryObjectBackoff, func() error {
+	return retryWithExponentialBackoff(ctx, createInventoryObjectBackoff, func(ctx context.Context) error {
 		cl, err := p.proxy.NewClient()
 		if err != nil {
 			return err
@@ -310,7 +310,7 @@ func (p *inventoryClient) List(ctx context.Context) (*clusterctlv1.ProviderList,
 	providerList := &clusterctlv1.ProviderList{}
 
 	listProvidersBackoff := newReadBackoff()
-	if err := retryWithExponentialBackoff(listProvidersBackoff, func() error {
+	if err := retryWithExponentialBackoff(ctx, listProvidersBackoff, func(ctx context.Context) error {
 		return listProviders(ctx, p.proxy, providerList)
 	}); err != nil {
 		return nil, err

cmd/clusterctl/client/cluster/mover.go

@@ -234,7 +234,7 @@ func (o *objectMover) checkProvisioningCompleted(ctx context.Context, graph *obj
 	for i := range clusters {
 		cluster := clusters[i]
 		clusterObj := &clusterv1.Cluster{}
-		if err := retryWithExponentialBackoff(readClusterBackoff, func() error {
+		if err := retryWithExponentialBackoff(ctx, readClusterBackoff, func(ctx context.Context) error {
 			return getClusterObj(ctx, o.fromProxy, cluster, clusterObj)
 		}); err != nil {
 			return err
@@ -264,7 +264,7 @@ func (o *objectMover) checkProvisioningCompleted(ctx context.Context, graph *obj
 	for i := range machines {
 		machine := machines[i]
 		machineObj := &clusterv1.Machine{}
-		if err := retryWithExponentialBackoff(readMachinesBackoff, func() error {
+		if err := retryWithExponentialBackoff(ctx, readMachinesBackoff, func(ctx context.Context) error {
 			return getMachineObj(ctx, o.fromProxy, machine, machineObj)
 		}); err != nil {
 			return err
@@ -574,7 +574,7 @@ func setClusterPause(ctx context.Context, proxy Proxy, clusters []*node, value b
 		log.V(5).Info("Set Cluster.Spec.Paused", "paused", value, "Cluster", klog.KRef(cluster.identity.Namespace, cluster.identity.Name))
 
 		// Nb. The operation is wrapped in a retry loop to make setClusterPause more resilient to unexpected conditions.
-		if err := retryWithExponentialBackoff(setClusterPauseBackoff, func() error {
+		if err := retryWithExponentialBackoff(ctx, setClusterPauseBackoff, func(ctx context.Context) error {
 			return patchCluster(ctx, proxy, cluster, patch, mutators...)
 		}); err != nil {
 			return errors.Wrapf(err, "error setting Cluster.Spec.Paused=%t", value)
@@ -601,7 +601,7 @@ func setClusterClassPause(ctx context.Context, proxy Proxy, clusterclasses []*no
 		}
 
 		// Nb. The operation is wrapped in a retry loop to make setClusterClassPause more resilient to unexpected conditions.
-		if err := retryWithExponentialBackoff(setClusterClassPauseBackoff, func() error {
+		if err := retryWithExponentialBackoff(ctx, setClusterClassPauseBackoff, func(ctx context.Context) error {
 			return pauseClusterClass(ctx, proxy, clusterclass, pause, mutators...)
 		}); err != nil {
 			return errors.Wrapf(err, "error updating ClusterClass %s/%s", clusterclass.identity.Namespace, clusterclass.identity.Name)
@@ -648,7 +648,7 @@ func waitReadyForMove(ctx context.Context, proxy Proxy, nodes []*node, dryRun bo
 		key := client.ObjectKeyFromObject(obj)
 
 		blockLogged := false
-		if err := retryWithExponentialBackoff(backoff, func() error {
+		if err := retryWithExponentialBackoff(ctx, backoff, func(ctx context.Context) error {
 			if err := c.Get(ctx, key, obj); err != nil {
 				return errors.Wrapf(err, "error getting %s/%s", obj.GroupVersionKind(), key)
 			}
@@ -788,7 +788,7 @@ func (o *objectMover) ensureNamespaces(ctx context.Context, graph *objectGraph,
 		}
 		namespaces.Insert(namespace)
 
-		if err := retryWithExponentialBackoff(ensureNamespaceBackoff, func() error {
+		if err := retryWithExponentialBackoff(ctx, ensureNamespaceBackoff, func(ctx context.Context) error {
 			return o.ensureNamespace(ctx, toProxy, namespace)
 		}); err != nil {
 			return err
@@ -872,7 +872,7 @@ func (o *objectMover) createGroup(ctx context.Context, group moveGroup, toProxy
 	for _, nodeToCreate := range group {
 		// Creates the Kubernetes object corresponding to the nodeToCreate.
 		// Nb. The operation is wrapped in a retry loop to make move more resilient to unexpected conditions.
-		err := retryWithExponentialBackoff(createTargetObjectBackoff, func() error {
+		err := retryWithExponentialBackoff(ctx, createTargetObjectBackoff, func(ctx context.Context) error {
 			return o.createTargetObject(ctx, nodeToCreate, toProxy, mutators, existingNamespaces)
 		})
 		if err != nil {
@@ -894,7 +894,7 @@ func (o *objectMover) backupGroup(ctx context.Context, group moveGroup, director
 	for _, nodeToBackup := range group {
 		// Backs-up the Kubernetes object corresponding to the nodeToBackup.
 		// Nb. The operation is wrapped in a retry loop to make move more resilient to unexpected conditions.
-		err := retryWithExponentialBackoff(backupTargetObjectBackoff, func() error {
+		err := retryWithExponentialBackoff(ctx, backupTargetObjectBackoff, func(ctx context.Context) error {
 			return o.backupTargetObject(ctx, nodeToBackup, directory)
 		})
 		if err != nil {
@@ -916,7 +916,7 @@ func (o *objectMover) restoreGroup(ctx context.Context, group moveGroup, toProxy
 	for _, nodeToRestore := range group {
 		// Creates the Kubernetes object corresponding to the nodeToRestore.
 		// Nb. The operation is wrapped in a retry loop to make move more resilient to unexpected conditions.
-		err := retryWithExponentialBackoff(restoreTargetObjectBackoff, func() error {
+		err := retryWithExponentialBackoff(ctx, restoreTargetObjectBackoff, func(ctx context.Context) error {
 			return o.restoreTargetObject(ctx, nodeToRestore, toProxy)
 		})
 		if err != nil {
@@ -1174,7 +1174,7 @@ func (o *objectMover) deleteGroup(ctx context.Context, group moveGroup) error {
 
 		// Delete the Kubernetes object corresponding to the current node.
 		// Nb. The operation is wrapped in a retry loop to make move more resilient to unexpected conditions.
-		err := retryWithExponentialBackoff(deleteSourceObjectBackoff, func() error {
+		err := retryWithExponentialBackoff(ctx, deleteSourceObjectBackoff, func(ctx context.Context) error {
 			return o.deleteSourceObject(ctx, nodeToDelete)
 		})
 

cmd/clusterctl/client/cluster/objectgraph.go

@@ -333,7 +333,7 @@ func (o *objectGraph) objMetaToNode(obj *unstructured.Unstructured, n *node) {
 func (o *objectGraph) getDiscoveryTypes(ctx context.Context) error {
 	crdList := &apiextensionsv1.CustomResourceDefinitionList{}
 	getDiscoveryTypesBackoff := newReadBackoff()
-	if err := retryWithExponentialBackoff(getDiscoveryTypesBackoff, func() error {
+	if err := retryWithExponentialBackoff(ctx, getDiscoveryTypesBackoff, func(ctx context.Context) error {
 		return getCRDList(ctx, o.proxy, crdList)
 	}); err != nil {
 		return err
@@ -432,7 +432,7 @@ func (o *objectGraph) Discovery(ctx context.Context, namespace string) error {
 		typeMeta := discoveryType.typeMeta
 		objList := new(unstructured.UnstructuredList)
 
-		if err := retryWithExponentialBackoff(discoveryBackoff, func() error {
+		if err := retryWithExponentialBackoff(ctx, discoveryBackoff, func(ctx context.Context) error {
 			return getObjList(ctx, o.proxy, typeMeta, selectors, objList)
 		}); err != nil {
 			return err
@@ -448,7 +448,7 @@ func (o *objectGraph) Discovery(ctx context.Context, namespace string) error {
 				if p.Type == string(clusterctlv1.InfrastructureProviderType) {
 					providerNamespaceSelector := []client.ListOption{client.InNamespace(p.Namespace)}
 					providerNamespaceSecretList := new(unstructured.UnstructuredList)
-					if err := retryWithExponentialBackoff(discoveryBackoff, func() error {
+					if err := retryWithExponentialBackoff(ctx, discoveryBackoff, func(ctx context.Context) error {
 						return getObjList(ctx, o.proxy, typeMeta, providerNamespaceSelector, providerNamespaceSecretList)
 					}); err != nil {
 						return err

cmd/clusterctl/client/cluster/ownergraph.go

@@ -76,7 +76,7 @@ func discoverOwnerGraph(ctx context.Context, namespace string, o *objectGraph) (
 		typeMeta := discoveryType.typeMeta
 		objList := new(unstructured.UnstructuredList)
 
-		if err := retryWithExponentialBackoff(discoveryBackoff, func() error {
+		if err := retryWithExponentialBackoff(ctx, discoveryBackoff, func(ctx context.Context) error {
 			return getObjList(ctx, o.proxy, typeMeta, selectors, objList)
 		}); err != nil {
 			return nil, err
@@ -92,7 +92,7 @@ func discoverOwnerGraph(ctx context.Context, namespace string, o *objectGraph) (
 				if p.Type == string(clusterctlv1.InfrastructureProviderType) {
 					providerNamespaceSelector := []client.ListOption{client.InNamespace(p.Namespace)}
 					providerNamespaceSecretList := new(unstructured.UnstructuredList)
-					if err := retryWithExponentialBackoff(discoveryBackoff, func() error {
+					if err := retryWithExponentialBackoff(ctx, discoveryBackoff, func(ctx context.Context) error {
 						return getObjList(ctx, o.proxy, typeMeta, providerNamespaceSelector, providerNamespaceSecretList)
 					}); err != nil {
 						return nil, err

cmd/clusterctl/client/cluster/proxy.go

@@ -165,7 +165,7 @@ func (k *proxy) NewClient() (client.Client, error) {
 	var c client.Client
 	// Nb. The operation is wrapped in a retry loop to make newClientSet more resilient to temporary connection problems.
 	connectBackoff := newConnectBackoff()
-	if err := retryWithExponentialBackoff(connectBackoff, func() error {
+	if err := retryWithExponentialBackoff(context.TODO(), connectBackoff, func(_ context.Context) error {
 		var err error
 		c, err = client.New(config, client.Options{Scheme: localScheme})
 		if err != nil {
@@ -191,7 +191,7 @@ func (k *proxy) CheckClusterAvailable() error {
 	}
 
 	connectBackoff := newShortConnectBackoff()
-	return retryWithExponentialBackoff(connectBackoff, func() error {
+	return retryWithExponentialBackoff(context.TODO(), connectBackoff, func(_ context.Context) error {
 		_, err := client.New(config, client.Options{Scheme: localScheme})
 		return err
 	})
@@ -224,7 +224,7 @@ func (k *proxy) ListResources(ctx context.Context, labels map[string]string, nam
 	// Get all the API resources in the cluster.
 	resourceListBackoff := newReadBackoff()
 	var resourceList []*metav1.APIResourceList
-	if err := retryWithExponentialBackoff(resourceListBackoff, func() error {
+	if err := retryWithExponentialBackoff(ctx, resourceListBackoff, func(ctx context.Context) error {
 		resourceList, err = cs.Discovery().ServerPreferredResources()
 		return err
 	}); err != nil {
@@ -236,7 +236,7 @@ func (k *proxy) ListResources(ctx context.Context, labels map[string]string, nam
 	crdsToExclude := sets.Set[string]{}
 
 	crdList := &apiextensionsv1.CustomResourceDefinitionList{}
-	if err := retryWithExponentialBackoff(newReadBackoff(), func() error {
+	if err := retryWithExponentialBackoff(ctx, newReadBackoff(), func(ctx context.Context) error {
 		return c.List(ctx, crdList)
 	}); err != nil {
 		return nil, errors.Wrap(err, "failed to list CRDs")
@@ -348,7 +348,7 @@ func listObjByGVK(ctx context.Context, c client.Client, groupVersion, kind strin
 	objList.SetKind(kind)
 
 	resourceListBackoff := newReadBackoff()
-	if err := retryWithExponentialBackoff(resourceListBackoff, func() error {
+	if err := retryWithExponentialBackoff(ctx, resourceListBackoff, func(ctx context.Context) error {
 		return c.List(ctx, objList, options...)
 	}); err != nil {
 		return nil, errors.Wrapf(err, "failed to list objects for the %q GroupVersionKind", objList.GroupVersionKind())
@@ -402,7 +402,7 @@ func (k *proxy) newClientSet() (*kubernetes.Clientset, error) {
 	var cs *kubernetes.Clientset
 	// Nb. The operation is wrapped in a retry loop to make newClientSet more resilient to temporary connection problems.
 	connectBackoff := newConnectBackoff()
-	if err := retryWithExponentialBackoff(connectBackoff, func() error {
+	if err := retryWithExponentialBackoff(context.TODO(), connectBackoff, func(_ context.Context) error {
 		var err error
 		cs, err = kubernetes.NewForConfig(config)
 		if err != nil {

cmd/clusterctl/client/cluster/upgrader.go

@@ -484,7 +484,7 @@ func (u *providerUpgrader) scaleDownProvider(ctx context.Context, provider clust
 
 // scaleDownDeployment scales down a Deployment to 0 and waits until all replicas have been deleted.
 func scaleDownDeployment(ctx context.Context, c client.Client, deploy appsv1.Deployment) error {
-	if err := retryWithExponentialBackoff(newWriteBackoff(), func() error {
+	if err := retryWithExponentialBackoff(ctx, newWriteBackoff(), func(ctx context.Context) error {
 		deployment := &appsv1.Deployment{}
 		if err := c.Get(ctx, client.ObjectKeyFromObject(&deploy), deployment); err != nil {
 			return errors.Wrapf(err, "failed to get Deployment/%s", deploy.GetName())
@@ -511,7 +511,7 @@ func scaleDownDeployment(ctx context.Context, c client.Client, deploy appsv1.Dep
 		Steps:    60,
 		Jitter:   0.4,
 	}
-	if err := retryWithExponentialBackoff(deploymentScaleToZeroBackOff, func() error {
+	if err := retryWithExponentialBackoff(ctx, deploymentScaleToZeroBackOff, func(ctx context.Context) error {
 		deployment := &appsv1.Deployment{}
 		if err := c.Get(ctx, client.ObjectKeyFromObject(&deploy), deployment); err != nil {
 			return errors.Wrapf(err, "failed to get Deployment/%s", deploy.GetName())