Repository: kubernetes-sigs/cluster-api
Pull request #7544 (2022-11-15 16:22:11)

[release-1.2] 🌱 ClusterCacheTracker: non-blocking per-cluster locking
Co-authored-by: Florian Gutmann <fgutmann@amazon.com>

<!-- please add an icon to the title of this PR (see https://sigs.k8s.io/cluster-api/CONTRIBUTING.md#contributing-a-patch), and delete this line and similar ones -->
<!-- the icon will be either ⚠️ (:warning:, major or breaking changes), ✨ (:sparkles:, feature additions), 🐛 (:bug:, patch and bugfixes), 📖 (:book:, documentation or proposals), or 🌱 (:seedling:, minor or other) -->

**What this PR does / why we need it**:

Cherry-pick of: #7537

**Which issue(s) this PR fixes** *(optional, in `fixes #<issue number>(, fixes #<issue_number>, ...)` format, will close the issue(s) when PR gets merged)*:
Fixes #



Patch:
diff --git a/cmd/clusterctl/client/config/providers_client.go b/cmd/clusterctl/client/config/providers_client.go
index da752b871..7d19d8ad7 100644
--- a/cmd/clusterctl/client/config/providers_client.go
+++ b/cmd/clusterctl/client/config/providers_client.go
@@ -58,7 +58,6 @@ const (
 	VSphereProviderName        = "vsphere"
 	MAASProviderName           = "maas"
 	KubevirtProviderName       = "kubevirt"
-	KubeKeyProviderName        = "kubekey"
 	VclusterProviderName       = "vcluster"
 	VirtinkProviderName        = "virtink"
 )
@@ -224,11 +223,6 @@ func (p *providersClient) defaults() []Provider {
 			url:          "https://github.com/nutanix-cloud-native/cluster-api-provider-nutanix/releases/latest/infrastructure-components.yaml",
 			providerType: clusterctlv1.InfrastructureProviderType,
 		},
-		&provider{
-			name:         KubeKeyProviderName,
-			url:          "https://github.com/kubesphere/kubekey/releases/latest/infrastructure-components.yaml",
-			providerType: clusterctlv1.InfrastructureProviderType,
-		},
 		&provider{
 			name:         KubevirtProviderName,
 			url:          "https://github.com/kubernetes-sigs/cluster-api-provider-kubevirt/releases/latest/infrastructure-components.yaml",
diff --git a/controllers/remote/cluster_cache_tracker.go b/controllers/remote/cluster_cache_tracker.go
index 86318a477..555eb0f82 100644
--- a/controllers/remote/cluster_cache_tracker.go
+++ b/controllers/remote/cluster_cache_tracker.go
@@ -36,6 +36,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/rest"
+	"k8s.io/klog/v2"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -53,9 +54,14 @@ const (
 	healthCheckPollInterval       = 10 * time.Second
 	healthCheckRequestTimeout     = 5 * time.Second
 	healthCheckUnhealthyThreshold = 10
+	initialCacheSyncTimeout       = 5 * time.Minute
 	clusterCacheControllerName    = "cluster-cache-tracker"
 )
 
+// ErrClusterLocked is returned in methods that require cluster-level locking
+// if the cluster is already locked by another concurrent call.
+var ErrClusterLocked = errors.New("cluster is locked already")
+
 // ClusterCacheTracker manages client caches for workload clusters.
 type ClusterCacheTracker struct {
 	log                   logr.Logger
@@ -63,9 +69,15 @@ type ClusterCacheTracker struct {
 	client                client.Client
 	scheme                *runtime.Scheme
 
-	lock             sync.RWMutex
+	// clusterAccessorsLock is used to lock the access to the clusterAccessors map.
+	clusterAccessorsLock sync.RWMutex
+	// clusterAccessors is the map of clusterAccessors by cluster.
 	clusterAccessors map[client.ObjectKey]*clusterAccessor
-	indexes          []Index
+	// clusterLock is a per-cluster lock used whenever we're locking for a specific cluster.
+	// E.g. for actions like creating a client or adding watches.
+	clusterLock *keyedMutex
+
+	indexes []Index
 
 	// controllerPodMetadata is the Pod metadata of the controller using this ClusterCacheTracker.
 	// This is only set when the POD_NAMESPACE, POD_NAME and POD_UID environment variables are set.
@@ -128,16 +140,14 @@ func NewClusterCacheTracker(manager ctrl.Manager, options ClusterCacheTrackerOpt
 		client:                manager.GetClient(),
 		scheme:                manager.GetScheme(),
 		clusterAccessors:      make(map[client.ObjectKey]*clusterAccessor),
+		clusterLock:           newKeyedMutex(),
 		indexes:               options.Indexes,
 	}, nil
 }
 
 // GetClient returns a cached client for the given cluster.
 func (t *ClusterCacheTracker) GetClient(ctx context.Context, cluster client.ObjectKey) (client.Client, error) {
-	t.lock.Lock()
-	defer t.lock.Unlock()
-
-	accessor, err := t.getClusterAccessorLH(ctx, cluster, t.indexes...)
+	accessor, err := t.getClusterAccessor(ctx, cluster, t.indexes...)
 	if err != nil {
 		return nil, err
 	}
@@ -147,10 +157,7 @@ func (t *ClusterCacheTracker) GetClient(ctx context.Context, cluster client.Obje
 
 // GetRESTConfig returns a cached REST config for the given cluster.
 func (t *ClusterCacheTracker) GetRESTConfig(ctc context.Context, cluster client.ObjectKey) (*rest.Config, error) {
-	t.lock.Lock()
-	defer t.lock.Unlock()
-
-	accessor, err := t.getClusterAccessorLH(ctc, cluster, t.indexes...)
+	accessor, err := t.getClusterAccessor(ctc, cluster, t.indexes...)
 	if err != nil {
 		return nil, err
 	}
@@ -168,29 +175,68 @@ type clusterAccessor struct {
 
 // clusterAccessorExists returns true if a clusterAccessor exists for cluster.
 func (t *ClusterCacheTracker) clusterAccessorExists(cluster client.ObjectKey) bool {
-	t.lock.RLock()
-	defer t.lock.RUnlock()
+	t.clusterAccessorsLock.RLock()
+	defer t.clusterAccessorsLock.RUnlock()
 
 	_, exists := t.clusterAccessors[cluster]
 	return exists
 }
 
-// getClusterAccessorLH first tries to return an already-created clusterAccessor for cluster, falling back to creating a
-// new clusterAccessor if needed. Note, this method requires t.lock to already be held (LH=lock held).
-func (t *ClusterCacheTracker) getClusterAccessorLH(ctx context.Context, cluster client.ObjectKey, indexes ...Index) (*clusterAccessor, error) {
-	a := t.clusterAccessors[cluster]
-	if a != nil {
-		return a, nil
+// loadAccessor loads a clusterAccessor.
+func (t *ClusterCacheTracker) loadAccessor(cluster client.ObjectKey) (*clusterAccessor, bool) {
+	t.clusterAccessorsLock.RLock()
+	defer t.clusterAccessorsLock.RUnlock()
+
+	accessor, ok := t.clusterAccessors[cluster]
+	return accessor, ok
+}
+
+// storeAccessor stores a clusterAccessor.
+func (t *ClusterCacheTracker) storeAccessor(cluster client.ObjectKey, accessor *clusterAccessor) {
+	t.clusterAccessorsLock.Lock()
+	defer t.clusterAccessorsLock.Unlock()
+
+	t.clusterAccessors[cluster] = accessor
+}
+
+// getClusterAccessor returns a clusterAccessor for cluster.
+// It first tries to return an already-created clusterAccessor.
+// It then falls back to create a new clusterAccessor if needed.
+// If there is already another go routine trying to create a clusterAccessor
+// for the same cluster, an error is returned.
+func (t *ClusterCacheTracker) getClusterAccessor(ctx context.Context, cluster client.ObjectKey, indexes ...Index) (*clusterAccessor, error) {
+	log := ctrl.LoggerFrom(ctx, "cluster", klog.KRef(cluster.Namespace, cluster.Name))
+
+	// If the clusterAccessor already exists, return early.
+	if accessor, ok := t.loadAccessor(cluster); ok {
+		return accessor, nil
 	}
 
-	a, err := t.newClusterAccessor(ctx, cluster, indexes...)
-	if err != nil {
-		return nil, errors.Wrap(err, "error creating client and cache for remote cluster")
+	// clusterAccessor doesn't exist yet, we might have to initialize one.
+	// Lock on the cluster to ensure only one clusterAccessor is initialized
+	// for the cluster at the same time.
+	// Return an error if another go routine already tries to create a clusterAccessor.
+	if ok := t.clusterLock.TryLock(cluster); !ok {
+		return nil, errors.Wrapf(ErrClusterLocked, "failed to create cluster accessor: failed to get lock for cluster")
 	}
+	defer t.clusterLock.Unlock(cluster)
 
-	t.clusterAccessors[cluster] = a
+	// Until we got the cluster lock a different goroutine might have initialized the clusterAccessor
+	// for this cluster successfully already. If this is the case we return it.
+	if accessor, ok := t.loadAccessor(cluster); ok {
+		return accessor, nil
+	}
+
+	// We are the go routine who has to initialize the clusterAccessor.
+	log.V(4).Info("Creating new cluster accessor")
+	accessor, err := t.newClusterAccessor(ctx, cluster, indexes...)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to create cluster accessor")
+	}
 
-	return a, nil
+	log.V(4).Info("Storing new cluster accessor")
+	t.storeAccessor(cluster, accessor)
+	return accessor, nil
 }
 
 // newClusterAccessor creates a new clusterAccessor.
@@ -264,7 +310,12 @@ func (t *ClusterCacheTracker) newClusterAccessor(ctx context.Context, cluster cl
 
 	// Start the cache!!!
 	go cache.Start(cacheCtx) //nolint:errcheck
-	if !cache.WaitForCacheSync(cacheCtx) {
+
+	// Wait until the cache is initially synced
+	cacheSyncCtx, cacheSyncCtxCancel := context.WithTimeout(ctx, initialCacheSyncTimeout)
+	defer cacheSyncCtxCancel()
+	if !cache.WaitForCacheSync(cacheSyncCtx) {
+		cache.Stop()
 		return nil, fmt.Errorf("failed waiting for cache for remote cluster %v to sync: %w", cluster, cacheCtx.Err())
 	}
 
@@ -335,22 +386,20 @@ func (t *ClusterCacheTracker) createClient(config *rest.Config, cluster client.O
 }
 
 // deleteAccessor stops a clusterAccessor's cache and removes the clusterAccessor from the tracker.
-func (t *ClusterCacheTracker) deleteAccessor(ctx context.Context, cluster client.ObjectKey) {
-	log := ctrl.LoggerFrom(ctx)
-
-	t.lock.Lock()
-	defer t.lock.Unlock()
+func (t *ClusterCacheTracker) deleteAccessor(_ context.Context, cluster client.ObjectKey) {
+	t.clusterAccessorsLock.Lock()
+	defer t.clusterAccessorsLock.Unlock()
 
 	a, exists := t.clusterAccessors[cluster]
 	if !exists {
 		return
 	}
 
-	log.V(2).Info("Deleting clusterAccessor", "cluster", cluster.String())
-
-	log.V(4).Info("Stopping cache", "cluster", cluster.String())
+	log := t.log.WithValues("Cluster", klog.KRef(cluster.Namespace, cluster.Name))
+	log.V(2).Info("Deleting clusterAccessor")
+	log.V(4).Info("Stopping cache")
 	a.cache.Stop()
-	log.V(4).Info("Cache stopped", "cluster", cluster.String())
+	log.V(4).Info("Cache stopped")
 
 	delete(t.clusterAccessors, cluster)
 }
@@ -388,26 +437,29 @@ func (t *ClusterCacheTracker) Watch(ctx context.Context, input WatchInput) error
 		return errors.New("input.Name is required")
 	}
 
-	t.lock.Lock()
-	defer t.lock.Unlock()
-
-	a, err := t.getClusterAccessorLH(ctx, input.Cluster, t.indexes...)
+	accessor, err := t.getClusterAccessor(ctx, input.Cluster, t.indexes...)
 	if err != nil {
-		return err
+		return errors.Wrapf(err, "failed to add %s watch on cluster %s", input.Kind, klog.KRef(input.Cluster.Namespace, input.Cluster.Name))
 	}
 
-	if a.watches.Has(input.Name) {
+	// We have to lock the cluster, so that the watch is not created multiple times in parallel.
+	if ok := t.clusterLock.TryLock(input.Cluster); !ok {
+		return errors.Wrapf(ErrClusterLocked, "failed to add %s watch on cluster %s: failed to get lock for cluster", input.Kind, klog.KRef(input.Cluster.Namespace, input.Cluster.Name))
+	}
+	defer t.clusterLock.Unlock(input.Cluster)
+
+	if accessor.watches.Has(input.Name) {
 		log := ctrl.LoggerFrom(ctx)
-		log.V(6).Info("Watch already exists", "namespace", input.Cluster.Namespace, "cluster", input.Cluster.Name, "name", input.Name)
+		log.V(6).Info("Watch already exists", "Cluster", klog.KRef(input.Cluster.Namespace, input.Cluster.Name), "name", input.Name)
 		return nil
 	}
 
 	// Need to create the watch
-	if err := input.Watcher.Watch(source.NewKindWithCache(input.Kind, a.cache), input.EventHandler, input.Predicates...); err != nil {
-		return errors.Wrap(err, "error creating watch")
+	if err := input.Watcher.Watch(source.NewKindWithCache(input.Kind, accessor.cache), input.EventHandler, input.Predicates...); err != nil {
+		return errors.Wrapf(err, "failed to add %s watch on cluster %s: failed to create watch", input.Kind, klog.KRef(input.Cluster.Namespace, input.Cluster.Name))
 	}
 
-	a.watches.Insert(input.Name)
+	accessor.watches.Insert(input.Name)
 
 	return nil
 }
@@ -474,7 +526,7 @@ func (t *ClusterCacheTracker) healthCheckCluster(ctx context.Context, in *health
 			return false, nil
 		}
 
-		if !t.clusterAccessorExists(in.cluster) {
+		if _, ok := t.loadAccessor(in.cluster); !ok {
 			// Cache for this cluster has already been cleaned up.
 			// Nothing to do, so return true.
 			return true, nil
@@ -509,8 +561,7 @@ func (t *ClusterCacheTracker) healthCheckCluster(ctx context.Context, in *health
 	// NB. we are ignoring ErrWaitTimeout because this error happens when the channel is close, that in this case
 	// happens when the cache is explicitly stopped.
 	if err != nil && err != wait.ErrWaitTimeout {
-		log := ctrl.LoggerFrom(ctx)
-		log.Error(err, "Error health checking cluster", "cluster", in.cluster.String())
+		t.log.Error(err, "Error health checking cluster", "Cluster", klog.KRef(in.cluster.Namespace, in.cluster.Name))
 		t.deleteAccessor(ctx, in.cluster)
 	}
 }
diff --git a/controllers/remote/keyedmutex.go b/controllers/remote/keyedmutex.go
new file mode 100644
index 000000000..58dbd86d5
--- /dev/null
+++ b/controllers/remote/keyedmutex.go
@@ -0,0 +1,66 @@
+/*
+Copyright 2021 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package remote
+
+import (
+	"sync"
+
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+// keyedMutex is a mutex locking on the key provided to the Lock function.
+// Only one caller can hold the lock for a specific key at a time.
+// A second Lock call if the lock is already held for a key returns false.
+type keyedMutex struct {
+	locksMtx sync.Mutex
+	locks    map[client.ObjectKey]struct{}
+}
+
+// newKeyedMutex creates a new keyed mutex ready for use.
+func newKeyedMutex() *keyedMutex {
+	return &keyedMutex{
+		locks: make(map[client.ObjectKey]struct{}),
+	}
+}
+
+// TryLock locks the passed in key if it's not already locked.
+// A second Lock call if the lock is already held for a key returns false.
+// In the ClusterCacheTracker case the key is the ObjectKey for a cluster.
+func (k *keyedMutex) TryLock(key client.ObjectKey) bool {
+	k.locksMtx.Lock()
+	defer k.locksMtx.Unlock()
+
+	// Check if there is already a lock for this key (e.g. Cluster).
+	if _, ok := k.locks[key]; ok {
+		// There is already a lock, return false.
+		return false
+	}
+
+	// Lock doesn't exist yet, create the lock.
+	k.locks[key] = struct{}{}
+
+	return true
+}
+
+// Unlock unlocks the key.
+func (k *keyedMutex) Unlock(key client.ObjectKey) {
+	k.locksMtx.Lock()
+	defer k.locksMtx.Unlock()
+
+	// Remove the lock if it exists.
+	delete(k.locks, key)
+}
diff --git a/controlplane/kubeadm/internal/controllers/controller.go b/controlplane/kubeadm/internal/controllers/controller.go
index a7eadbb6a..e476d52b0 100644
--- a/controlplane/kubeadm/internal/controllers/controller.go
+++ b/controlplane/kubeadm/internal/controllers/controller.go
@@ -202,11 +202,25 @@ func (r *KubeadmControlPlaneReconciler) Reconcile(ctx context.Context, req ctrl.
 
 	if !kcp.ObjectMeta.DeletionTimestamp.IsZero() {
 		// Handle deletion reconciliation loop.
-		return r.reconcileDelete(ctx, cluster, kcp)
+		res, err = r.reconcileDelete(ctx, cluster, kcp)
+		// Requeue if the reconcile failed because the ClusterCacheTracker was locked for
+		// the current cluster because of concurrent access.
+		if errors.Is(err, remote.ErrClusterLocked) {
+			log.V(5).Info("Requeueing because another worker has the lock on the ClusterCacheTracker")
+			return ctrl.Result{Requeue: true}, nil
+		}
+		return res, err
 	}
 
 	// Handle normal reconciliation loop.
-	return r.reconcile(ctx, cluster, kcp)
+	res, err = r.reconcile(ctx, cluster, kcp)
+	// Requeue if the reconcile failed because the ClusterCacheTracker was locked for
+	// the current cluster because of concurrent access.
+	if errors.Is(err, remote.ErrClusterLocked) {
+		log.V(5).Info("Requeueing because another worker has the lock on the ClusterCacheTracker")
+		return ctrl.Result{Requeue: true}, nil
+	}
+	return res, err
 }
 
 func patchKubeadmControlPlane(ctx context.Context, patchHelper *patch.Helper, kcp *controlplanev1.KubeadmControlPlane) error {
diff --git a/docs/book/src/clusterctl/provider-contract.md b/docs/book/src/clusterctl/provider-contract.md
index 206ff7813..63ef8d7c7 100644
--- a/docs/book/src/clusterctl/provider-contract.md
+++ b/docs/book/src/clusterctl/provider-contract.md
@@ -232,33 +232,32 @@ easier transition from `kubectl apply` to `clusterctl`.
 As a reference you can consider the labels applied to the following
 providers.
 
-| Provider Name| Label                                                   |
-|--------------|---------------------------------------------------------|
-|CAPI          | cluster.x-k8s.io/provider=cluster-api                   |
-|CABPK         | cluster.x-k8s.io/provider=bootstrap-kubeadm             |
-|CACPK         | cluster.x-k8s.io/provider=control-plane-kubeadm         |
-|CACPN         | cluster.x-k8s.io/provider=control-plane-nested          |
-|CAPA          | cluster.x-k8s.io/provider=infrastructure-aws            |
-|CAPB          | cluster.x-k8s.io/provider=infrastructure-byoh           |
-|CAPC          | cluster.x-k8s.io/provider=infrastructure-cloudstack     |
-|CAPD          | cluster.x-k8s.io/provider=infrastructure-docker         |
-|CAPDO         | cluster.x-k8s.io/provider=infrastructure-digitalocean   |
-|CAPG          | cluster.x-k8s.io/provider=infrastructure-gcp            |
-|CAPH          | cluster.x-k8s.io/provider=infrastructure-hetzner        |
-|CAPIBM        | cluster.x-k8s.io/provider=infrastructure-ibmcloud       |
-|CAPKK         | cluster.x-k8s.io/provider=infrastructure-kubekey        |
-|CAPK          | cluster.x-k8s.io/provider=infrastructure-kubevirt       |
-|CAPM3         | cluster.x-k8s.io/provider=infrastructure-metal3         |
-|CAPN          | cluster.x-k8s.io/provider=infrastructure-nested         |
-|CAPO          | cluster.x-k8s.io/provider=infrastructure-openstack      |
-|CAPOCI        | cluster.x-k8s.io/provider=infrastructure-oci            |
-|CAPP          | cluster.x-k8s.io/provider=infrastructure-packet         |
-|CAPV          | cluster.x-k8s.io/provider=infrastructure-vsphere        |
-|CAPVC         | cluster.x-k8s.io/provider=infrastructure-vcluster       |
-|CAPVCD        | cluster.x-k8s.io/provider=infrastructure-vcd            |
-|CAPX          | cluster.x-k8s.io/provider=infrastructure-nutanix        |
-|CAPZ          | cluster.x-k8s.io/provider=infrastructure-azure          |
-|CAPOSC        | cluster.x-k8s.io/provider=infrastructure-outscale       |
+| Provider Name| Label                                                  |
+|--------------|--------------------------------------------------------|
+|CAPI          | cluster.x-k8s.io/provider=cluster-api                  |
+|CABPK         | cluster.x-k8s.io/provider=bootstrap-kubeadm            |
+|CACPK         | cluster.x-k8s.io/provider=control-plane-kubeadm        |
+|CACPN         | cluster.x-k8s.io/provider=control-plane-nested         |
+|CAPA          | cluster.x-k8s.io/provider=infrastructure-aws           |
+|CAPB          | cluster.x-k8s.io/provider=infrastructure-byoh          |
+|CAPC          | cluster.x-k8s.io/provider=infrastructure-cloudstack    |
+|CAPD          | cluster.x-k8s.io/provider=infrastructure-docker        |
+|CAPDO         | cluster.x-k8s.io/provider=infrastructure-digitalocean  |
+|CAPG          | cluster.x-k8s.io/provider=infrastructure-gcp           |
+|CAPH          | cluster.x-k8s.io/provider=infrastructure-hetzner       |
+|CAPIBM        | cluster.x-k8s.io/provider=infrastructure-ibmcloud      |
+|CAPK          | cluster.x-k8s.io/provider=infrastructure-kubevirt      |
+|CAPM3         | cluster.x-k8s.io/provider=infrastructure-metal3        |
+|CAPN          | cluster.x-k8s.io/provider=infrastructure-nested        |
+|CAPO          | cluster.x-k8s.io/provider=infrastructure-openstack     |
+|CAPOCI        | cluster.x-k8s.io/provider=infrastructure-oci           |
+|CAPP          | cluster.x-k8s.io/provider=infrastructure-packet        |
+|CAPV          | cluster.x-k8s.io/provider=infrastructure-vsphere       |
+|CAPVC         | cluster.x-k8s.io/provider=infrastructure-vcluster      |
+|CAPVCD        | cluster.x-k8s.io/provider=infrastructure-vcd           |
+|CAPX          | cluster.x-k8s.io/provider=infrastructure-nutanix       |
+|CAPZ          | cluster.x-k8s.io/provider=infrastructure-azure         |
+|CAPOSC        | cluster.x-k8s.io/provider=infrastructure-outscale      |
 ### Workload cluster templates
 
 An infrastructure provider could publish a **cluster templates** file to be used by `clusterctl generate cluster`.
diff --git a/docs/book/src/reference/glossary.md b/docs/book/src/reference/glossary.md
index 60cd05bc4..d3368235f 100644
--- a/docs/book/src/reference/glossary.md
+++ b/docs/book/src/reference/glossary.md
@@ -72,9 +72,6 @@ Cluster API Provider Nested
 ### CAPX
 Cluster API Provider Nutanix
 
-### CAPKK
-Cluster API Provider KubeKey
-
 ### CAPK
 Cluster API Provider Kubevirt
 
diff --git a/docs/book/src/reference/providers.md b/docs/book/src/reference/providers.md
index 20d8b062d..1a882829d 100644
--- a/docs/book/src/reference/providers.md
+++ b/docs/book/src/reference/providers.md
@@ -22,7 +22,6 @@ updated info about which API version they are supporting.
 - [Hetzner](https://github.com/syself/cluster-api-provider-hetzner)
 - [Outscale](https://github.com/outscale-dev/cluster-api-provider-outscale)
 - [IBM Cloud](https://github.com/kubernetes-sigs/cluster-api-provider-ibmcloud)
-- [KubeKey](https://github.com/kubesphere/kubekey)
 - [KubeVirt](https://github.com/kubernetes-sigs/cluster-api-provider-kubevirt)
 - [MAAS](https://github.com/spectrocloud/cluster-api-provider-maas)
 - [Metal3](https://github.com/metal3-io/cluster-api-provider-metal3)
diff --git a/docs/book/src/user/quick-start.md b/docs/book/src/user/quick-start.md
index 1af50473d..16059c277 100644
--- a/docs/book/src/user/quick-start.md
+++ b/docs/book/src/user/quick-start.md
@@ -206,7 +206,7 @@ Additional documentation about experimental features can be found in [Experiment
 Depending on the infrastructure provider you are planning to use, some additional prerequisites should be satisfied
 before getting started with Cluster API. See below for the expected settings for common providers.
 
-{{#tabs name:"tab-installation-infrastructure" tabs:"AWS,Azure,CloudStack,DigitalOcean,Docker,Equinix Metal,GCP,Hetzner,IBM Cloud,KubeKey,Kubevirt,Metal3,Nutanix,OCI,OpenStack,Outscale,VCD,vcluster,Virtink,vSphere"}}
+{{#tabs name:"tab-installation-infrastructure" tabs:"AWS,Azure,CloudStack,DigitalOcean,Docker,Equinix Metal,GCP,Hetzner,IBM Cloud,Kubevirt,Metal3,Nutanix,OCI,OpenStack,Outscale,VCD,vcluster,Virtink,vSphere"}}
 {{#tab AWS}}
 
 Download the latest binary of `clusterawsadm` from the [AWS provider releases].
@@ -437,13 +437,6 @@ export IBMCLOUD_API_KEY=<you_api_key>
 clusterctl init --infrastructure ibmcloud
 ```
 
-{{#/tab }}
-{{#tab KubeKey}}
-```bash
-# Initialize the management cluster
-clusterctl init --infrastructure kubekey
-```
-
 {{#/tab }}
 {{#tab Kubevirt}}
 
@@ -604,7 +597,7 @@ before configuring a cluster with Cluster API. Instructions are provided for com
 Otherwise, you can look at the `clusterctl generate cluster` [command][clusterctl generate cluster] documentation for details about how to
 discover the list of variables required by a cluster templates.
 
-{{#tabs name:"tab-configuration-infrastructure" tabs:"AWS,Azure,CloudStack,DigitalOcean,Docker,Equinix Metal,GCP,IBM Cloud,KubeKey,Kubevirt,Metal3,Nutanix,OpenStack,Outscale,VCD,vcluster,Virtink,vSphere"}}
+{{#tabs name:"tab-configuration-infrastructure" tabs:"AWS,Azure,CloudStack,DigitalOcean,Docker,Equinix Metal,GCP,IBM Cloud,Kubevirt,Metal3,Nutanix,OpenStack,Outscale,VCD,vcluster,Virtink,vSphere"}}
 {{#tab AWS}}
 
 ```bash
@@ -803,22 +796,6 @@ export IBMPOWERVS_NETWORK_NAME=<your-capi-network-name>
 
 Please visit the [IBM Cloud provider] for more information.
 
-{{#/tab }}
-{{#tab KubeKey}}
-```bash
-# Required environment variables
-# The KKZONE is used to specify where to download the binaries. (e.g. "", "cn")
-export KKZONE=""
-# The ssh name of the all instance Linux user. (e.g. root, ubuntu)
-export USER_NAME=<your-linux-user>
-# The ssh password of the all instance Linux user. 
-export PASSWORD=<your-linux-user-password>
-# The ssh IP address of the all instance. (e.g. "[{address: 192.168.100.3}, {address: 192.168.100.4}]")
-export INSTANCES=<your-linux-ip-address>
-# The cluster control plane VIP. (e.g. "192.168.100.100")
-export CONTROL_PLANE_ENDPOINT_IP=<your-control-plane-virtual-ip>
-```
-
 {{#/tab }}
 {{#tab Kubevirt}}
 
diff --git a/exp/addons/internal/controllers/clusterresourceset_controller.go b/exp/addons/internal/controllers/clusterresourceset_controller.go
index 77adae1d6..428504ccf 100644
--- a/exp/addons/internal/controllers/clusterresourceset_controller.go
+++ b/exp/addons/internal/controllers/clusterresourceset_controller.go
@@ -151,6 +151,12 @@ func (r *ClusterResourceSetReconciler) Reconcile(ctx context.Context, req ctrl.R
 
 	for _, cluster := range clusters {
 		if err := r.ApplyClusterResourceSet(ctx, cluster, clusterResourceSet); err != nil {
+			// Requeue if the reconcile failed because the ClusterCacheTracker was locked for
+			// the current cluster because of concurrent access.
+			if errors.Is(err, remote.ErrClusterLocked) {
+				log.V(5).Info("Requeueing because another worker has the lock on the ClusterCacheTracker")
+				return ctrl.Result{Requeue: true}, nil
+			}
 			return ctrl.Result{}, err
 		}
 	}
diff --git a/internal/controllers/machine/machine_controller.go b/internal/controllers/machine/machine_controller.go
index cf892cbb8..41721e639 100644
--- a/internal/controllers/machine/machine_controller.go
+++ b/internal/controllers/machine/machine_controller.go
@@ -197,11 +197,25 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (_ ctrl.Re
 
 	// Handle deletion reconciliation loop.
 	if !m.ObjectMeta.DeletionTimestamp.IsZero() {
-		return r.reconcileDelete(ctx, cluster, m)
+		res, err := r.reconcileDelete(ctx, cluster, m)
+		// Requeue if the reconcile failed because the ClusterCacheTracker was locked for
+		// the current cluster because of concurrent access.
+		if errors.Is(err, remote.ErrClusterLocked) {
+			log.V(5).Info("Requeueing because another worker has the lock on the ClusterCacheTracker")
+			return ctrl.Result{Requeue: true}, nil
+		}
+		return res, err
 	}
 
 	// Handle normal reconciliation loop.
-	return r.reconcile(ctx, cluster, m)
+	res, err := r.reconcile(ctx, cluster, m)
+	// Requeue if the reconcile failed because the ClusterCacheTracker was locked for
+	// the current cluster because of concurrent access.
+	if errors.Is(err, remote.ErrClusterLocked) {
+		log.V(5).Info("Requeueing because another worker has the lock on the ClusterCacheTracker")
+		return ctrl.Result{Requeue: true}, nil
+	}
+	return res, err
 }
 
 func patchMachine(ctx context.Context, patchHelper *patch.Helper, machine *clusterv1.Machine, options ...patch.Option) error {
@@ -243,11 +257,8 @@ func patchMachine(ctx context.Context, patchHelper *patch.Helper, machine *clust
 }
 
 func (r *Reconciler) reconcile(ctx context.Context, cluster *clusterv1.Cluster, m *clusterv1.Machine) (ctrl.Result, error) {
-	log := ctrl.LoggerFrom(ctx)
-
 	if conditions.IsTrue(cluster, clusterv1.ControlPlaneInitializedCondition) {
 		if err := r.watchClusterNodes(ctx, cluster); err != nil {
-			log.Error(err, "error watching nodes on target cluster")
 			return ctrl.Result{}, err
 		}
 	}
diff --git a/internal/controllers/machinehealthcheck/machinehealthcheck_controller.go b/internal/controllers/machinehealthcheck/machinehealthcheck_controller.go
index be39ea2cc..7382135ba 100644
--- a/internal/controllers/machinehealthcheck/machinehealthcheck_controller.go
+++ b/internal/controllers/machinehealthcheck/machinehealthcheck_controller.go
@@ -174,6 +174,12 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (_ ctrl.Re
 
 	result, err := r.reconcile(ctx, log, cluster, m)
 	if err != nil {
+		// Requeue if the reconcile failed because the ClusterCacheTracker was locked for
+		// the current cluster because of concurrent access.
+		if errors.Is(err, remote.ErrClusterLocked) {
+			log.V(5).Info("Requeueing because another worker has the lock on the ClusterCacheTracker")
+			return ctrl.Result{Requeue: true}, nil
+		}
 		log.Error(err, "Failed to reconcile MachineHealthCheck")
 		r.recorder.Eventf(m, corev1.EventTypeWarning, "ReconcileError", "%v", err)
 
@@ -201,7 +207,6 @@ func (r *Reconciler) reconcile(ctx context.Context, logger logr.Logger, cluster
 	}
 
 	if err := r.watchClusterNodes(ctx, cluster); err != nil {
-		logger.Error(err, "error watching nodes on target cluster")
 		return ctrl.Result{}, err
 	}
 
diff --git a/internal/controllers/machineset/machineset_controller.go b/internal/controllers/machineset/machineset_controller.go
index 5b15d3f1c..1f85108f3 100644
--- a/internal/controllers/machineset/machineset_controller.go
+++ b/internal/controllers/machineset/machineset_controller.go
@@ -162,6 +162,12 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (_ ctrl.Re
 
 	result, err := r.reconcile(ctx, cluster, machineSet)
 	if err != nil {
+		// Requeue if the reconcile failed because the ClusterCacheTracker was locked for
+		// the current cluster because of concurrent access.
+		if errors.Is(err, remote.ErrClusterLocked) {
+			log.V(5).Info("Requeueing because another worker has the lock on the ClusterCacheTracker")
+			return ctrl.Result{Requeue: true}, nil
+		}
 		log.Error(err, "Failed to reconcile MachineSet")
 		r.recorder.Eventf(machineSet, corev1.EventTypeWarning, "ReconcileError", "%v", err)
 	}


Test patch:
diff --git a/cmd/clusterctl/client/config_test.go b/cmd/clusterctl/client/config_test.go
index 8c3ee2471..66eb25596 100644
--- a/cmd/clusterctl/client/config_test.go
+++ b/cmd/clusterctl/client/config_test.go
@@ -70,7 +70,6 @@ func Test_clusterctlClient_GetProvidersConfig(t *testing.T) {
 				config.GCPProviderName,
 				config.HetznerProviderName,
 				config.IBMCloudProviderName,
-				config.KubeKeyProviderName,
 				config.KubevirtProviderName,
 				config.MAASProviderName,
 				config.Metal3ProviderName,
@@ -111,7 +110,6 @@ func Test_clusterctlClient_GetProvidersConfig(t *testing.T) {
 				config.GCPProviderName,
 				config.HetznerProviderName,
 				config.IBMCloudProviderName,
-				config.KubeKeyProviderName,
 				config.KubevirtProviderName,
 				config.MAASProviderName,
 				config.Metal3ProviderName,
diff --git a/cmd/clusterctl/cmd/config_repositories_test.go b/cmd/clusterctl/cmd/config_repositories_test.go
index 85a8ad410..84a0b865d 100644
--- a/cmd/clusterctl/cmd/config_repositories_test.go
+++ b/cmd/clusterctl/cmd/config_repositories_test.go
@@ -116,7 +116,6 @@ docker              InfrastructureProvider   https://github.com/kubernetes-sigs/
 gcp                 InfrastructureProvider   https://github.com/kubernetes-sigs/cluster-api-provider-gcp/releases/latest/              infrastructure-components.yaml
 hetzner             InfrastructureProvider   https://github.com/syself/cluster-api-provider-hetzner/releases/latest/                   infrastructure-components.yaml
 ibmcloud            InfrastructureProvider   https://github.com/kubernetes-sigs/cluster-api-provider-ibmcloud/releases/latest/         infrastructure-components.yaml
-kubekey             InfrastructureProvider   https://github.com/kubesphere/kubekey/releases/latest/                                    infrastructure-components.yaml
 kubevirt            InfrastructureProvider   https://github.com/kubernetes-sigs/cluster-api-provider-kubevirt/releases/latest/         infrastructure-components.yaml
 maas                InfrastructureProvider   https://github.com/spectrocloud/cluster-api-provider-maas/releases/latest/                infrastructure-components.yaml
 metal3              InfrastructureProvider   https://github.com/metal3-io/cluster-api-provider-metal3/releases/latest/                 infrastructure-components.yaml
@@ -198,10 +197,6 @@ var expectedOutputYaml = `- File: core_components.yaml
   Name: ibmcloud
   ProviderType: InfrastructureProvider
   URL: https://github.com/kubernetes-sigs/cluster-api-provider-ibmcloud/releases/latest/
-- File: infrastructure-components.yaml
-  Name: kubekey
-  ProviderType: InfrastructureProvider
-  URL: https://github.com/kubesphere/kubekey/releases/latest/
 - File: infrastructure-components.yaml
   Name: kubevirt
   ProviderType: InfrastructureProvider
diff --git a/controllers/remote/cluster_cache_healthcheck_test.go b/controllers/remote/cluster_cache_healthcheck_test.go
index 7e46428e9..f4e3e5a9e 100644
--- a/controllers/remote/cluster_cache_healthcheck_test.go
+++ b/controllers/remote/cluster_cache_healthcheck_test.go
@@ -140,7 +140,10 @@ func TestClusterCacheHealthCheck(t *testing.T) {
 			})
 
 			// Make sure this passes for at least for some seconds, to give the health check goroutine time to run.
-			g.Consistently(func() bool { return cct.clusterAccessorExists(testClusterKey) }, 5*time.Second, 1*time.Second).Should(BeTrue())
+			g.Consistently(func() bool {
+				_, ok := cct.loadAccessor(testClusterKey)
+				return ok
+			}, 5*time.Second, 1*time.Second).Should(BeTrue())
 		})
 
 		t.Run("with an invalid path", func(t *testing.T) {
@@ -162,7 +165,10 @@ func TestClusterCacheHealthCheck(t *testing.T) {
 				})
 
 			// This should succeed after N consecutive failed requests.
-			g.Eventually(func() bool { return cct.clusterAccessorExists(testClusterKey) }, 5*time.Second, 1*time.Second).Should(BeFalse())
+			g.Eventually(func() bool {
+				_, ok := cct.loadAccessor(testClusterKey)
+				return ok
+			}, 5*time.Second, 1*time.Second).Should(BeFalse())
 		})
 
 		t.Run("with an invalid config", func(t *testing.T) {
@@ -193,7 +199,10 @@ func TestClusterCacheHealthCheck(t *testing.T) {
 			})
 
 			// This should succeed after N consecutive failed requests.
-			g.Eventually(func() bool { return cct.clusterAccessorExists(testClusterKey) }, 5*time.Second, 1*time.Second).Should(BeFalse())
+			g.Eventually(func() bool {
+				_, ok := cct.loadAccessor(testClusterKey)
+				return ok
+			}, 5*time.Second, 1*time.Second).Should(BeFalse())
 		})
 	})
 }
diff --git a/controllers/remote/keyedmutex_test.go b/controllers/remote/keyedmutex_test.go
new file mode 100644
index 000000000..c982bf093
--- /dev/null
+++ b/controllers/remote/keyedmutex_test.go
@@ -0,0 +1,82 @@
+/*
+Copyright 2021 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package remote
+
+import (
+	"testing"
+
+	. "github.com/onsi/gomega"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+func TestKeyedMutex(t *testing.T) {
+	t.Run("Lock a Cluster and ensures the second Lock on the same Cluster returns false", func(t *testing.T) {
+		t.Parallel()
+		g := NewWithT(t)
+
+		cluster1 := client.ObjectKey{Namespace: metav1.NamespaceDefault, Name: "Cluster1"}
+		km := newKeyedMutex()
+
+		// Try to lock cluster1.
+		// Should work as nobody currently holds the lock for cluster1.
+		g.Expect(km.TryLock(cluster1)).To(BeTrue())
+
+		// Try to lock cluster1 again.
+		// Shouldn't work as cluster1 is already locked.
+		g.Expect(km.TryLock(cluster1)).To(BeFalse())
+
+		// Unlock cluster1.
+		km.Unlock(cluster1)
+
+		// Ensure that the lock was cleaned up from the internal map.
+		g.Expect(km.locks).To(HaveLen(0))
+	})
+
+	t.Run("Can lock different Clusters in parallel but each one only once", func(t *testing.T) {
+		g := NewWithT(t)
+		km := newKeyedMutex()
+		clusters := []client.ObjectKey{
+			{Namespace: metav1.NamespaceDefault, Name: "Cluster1"},
+			{Namespace: metav1.NamespaceDefault, Name: "Cluster2"},
+			{Namespace: metav1.NamespaceDefault, Name: "Cluster3"},
+			{Namespace: metav1.NamespaceDefault, Name: "Cluster4"},
+		}
+
+		// Run this twice to ensure Clusters can be locked again
+		// after they have been unlocked.
+		for i := 0; i < 2; i++ {
+			// Lock all Clusters (should work).
+			for _, key := range clusters {
+				g.Expect(km.TryLock(key)).To(BeTrue())
+			}
+
+			// Ensure Clusters can't be locked again.
+			for _, key := range clusters {
+				g.Expect(km.TryLock(key)).To(BeFalse())
+			}
+
+			// Unlock all Clusters.
+			for _, key := range clusters {
+				km.Unlock(key)
+			}
+		}
+
+		// Ensure that the lock was cleaned up from the internal map.
+		g.Expect(km.locks).To(HaveLen(0))
+	})
+}
diff --git a/test/infrastructure/docker/exp/internal/controllers/dockermachinepool_controller.go b/test/infrastructure/docker/exp/internal/controllers/dockermachinepool_controller.go
index 65c137e0f..8164c2c80 100644
--- a/test/infrastructure/docker/exp/internal/controllers/dockermachinepool_controller.go
+++ b/test/infrastructure/docker/exp/internal/controllers/dockermachinepool_controller.go
@@ -34,6 +34,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/source"
 
 	clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1"
+	"sigs.k8s.io/cluster-api/controllers/remote"
 	expv1 "sigs.k8s.io/cluster-api/exp/api/v1beta1"
 	utilexp "sigs.k8s.io/cluster-api/exp/util"
 	"sigs.k8s.io/cluster-api/test/infrastructure/container"
@@ -123,7 +124,14 @@ func (r *DockerMachinePoolReconciler) Reconcile(ctx context.Context, req ctrl.Re
 	}
 
 	// Handle non-deleted machines
-	return r.reconcileNormal(ctx, cluster, machinePool, dockerMachinePool)
+	res, err = r.reconcileNormal(ctx, cluster, machinePool, dockerMachinePool)
+	// Requeue if the reconcile failed because the ClusterCacheTracker was locked for
+	// the current cluster because of concurrent access.
+	if errors.Is(err, remote.ErrClusterLocked) {
+		log.V(5).Info("Requeueing because another worker has the lock on the ClusterCacheTracker")
+		return ctrl.Result{Requeue: true}, nil
+	}
+	return res, err
 }
 
 // SetupWithManager will add watches for this controller.
diff --git a/test/infrastructure/docker/internal/controllers/dockermachine_controller.go b/test/infrastructure/docker/internal/controllers/dockermachine_controller.go
index fd0204b55..37e2b332d 100644
--- a/test/infrastructure/docker/internal/controllers/dockermachine_controller.go
+++ b/test/infrastructure/docker/internal/controllers/dockermachine_controller.go
@@ -35,6 +35,7 @@ import (
 
 	clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1"
 	bootstrapv1 "sigs.k8s.io/cluster-api/bootstrap/kubeadm/api/v1beta1"
+	"sigs.k8s.io/cluster-api/controllers/remote"
 	"sigs.k8s.io/cluster-api/test/infrastructure/container"
 	infrav1 "sigs.k8s.io/cluster-api/test/infrastructure/docker/api/v1beta1"
 	"sigs.k8s.io/cluster-api/test/infrastructure/docker/internal/docker"
@@ -156,7 +157,14 @@ func (r *DockerMachineReconciler) Reconcile(ctx context.Context, req ctrl.Reques
 	}
 
 	// Handle non-deleted machines
-	return r.reconcileNormal(ctx, cluster, machine, dockerMachine, externalMachine, externalLoadBalancer)
+	res, err := r.reconcileNormal(ctx, cluster, machine, dockerMachine, externalMachine, externalLoadBalancer)
+	// Requeue if the reconcile failed because the ClusterCacheTracker was locked for
+	// the current cluster because of concurrent access.
+	if errors.Is(err, remote.ErrClusterLocked) {
+		log.V(5).Info("Requeueing because another worker has the lock on the ClusterCacheTracker")
+		return ctrl.Result{Requeue: true}, nil
+	}
+	return res, err
 }
 
 func patchDockerMachine(ctx context.Context, patchHelper *patch.Helper, dockerMachine *infrav1.DockerMachine) error {