Reconcile remote datacenters independently

Refactors ScyllaDBCluster controller to reconcile each datacenter independenly.
Errors or connection issues of down datacenter no longer affects reconcilation of other datacenter.

Author: zimnx

pkg/controller/scylladbcluster/resource.go

@@ -3,6 +3,7 @@ package scylladbcluster
 import (
 	"context"
 	"fmt"
+
 	scyllav1alpha1 "github.com/scylladb/scylla-operator/pkg/api/scylla/v1alpha1"
 	"github.com/scylladb/scylla-operator/pkg/controllerhelpers"
 	"github.com/scylladb/scylla-operator/pkg/naming"
@@ -12,15 +13,16 @@ import (
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 )
 
-func (scc *Controller) syncConfigMaps(
+func (scc *Controller) syncRemoteConfigMaps(
 	ctx context.Context,
 	sc *scyllav1alpha1.ScyllaDBCluster,
-	remoteNamespaces map[string]*corev1.Namespace,
-	remoteControllers map[string]metav1.Object,
-	remoteConfigMaps map[string]map[string]*corev1.ConfigMap,
+	dc *scyllav1alpha1.ScyllaDBClusterDatacenter,
+	remoteNamespace *corev1.Namespace,
+	remoteController metav1.Object,
+	remoteConfigMaps map[string]*corev1.ConfigMap,
 	managingClusterDomain string,
 ) ([]metav1.Condition, error) {
-	progressingConditions, requiredConfigMaps, err := MakeRemoteConfigMaps(sc, remoteNamespaces, remoteControllers, scc.configMapLister, managingClusterDomain)
+	progressingConditions, requiredConfigMaps, err := MakeRemoteConfigMaps(sc, dc, remoteNamespace, remoteController, scc.configMapLister, managingClusterDomain)
 	if err != nil {
 		return progressingConditions, fmt.Errorf("can't make required configmaps: %w", err)
 	}
@@ -29,54 +31,34 @@ func (scc *Controller) syncConfigMaps(
 		return progressingConditions, nil
 	}
 
-	// Delete has to be the first action to avoid getting stuck on quota.
-	var deletionErrors []error
-	for _, dc := range sc.Spec.Datacenters {
-		ns, ok := remoteNamespaces[dc.RemoteKubernetesClusterName]
-		if !ok {
-			continue
-		}
-
-		clusterClient, err := scc.kubeRemoteClient.Cluster(dc.RemoteKubernetesClusterName)
-		if err != nil {
-			return nil, fmt.Errorf("can't get client to %q cluster: %w", dc.RemoteKubernetesClusterName, err)
-		}
-
-		err = controllerhelpers.Prune(ctx,
-			requiredConfigMaps[dc.RemoteKubernetesClusterName],
-			remoteConfigMaps[dc.RemoteKubernetesClusterName],
-			&controllerhelpers.PruneControlFuncs{
-				DeleteFunc: clusterClient.CoreV1().ConfigMaps(ns.Name).Delete,
-			},
-			scc.eventRecorder,
-		)
-		if err != nil {
-			return progressingConditions, fmt.Errorf("can't prune configmap(s) in %q Datacenter of %q ScyllaDBCluster: %w", dc.Name, naming.ObjRef(sc), err)
-		}
+	clusterClient, err := scc.kubeRemoteClient.Cluster(dc.RemoteKubernetesClusterName)
+	if err != nil {
+		return nil, fmt.Errorf("can't get client to %q cluster: %w", dc.RemoteKubernetesClusterName, err)
 	}
 
-	if err := utilerrors.NewAggregate(deletionErrors); err != nil {
-		return nil, fmt.Errorf("can't prune remote configmap(s): %w", err)
+	// Delete has to be the first action to avoid getting stuck on quota.
+	err = controllerhelpers.Prune(ctx,
+		requiredConfigMaps,
+		remoteConfigMaps,
+		&controllerhelpers.PruneControlFuncs{
+			DeleteFunc: clusterClient.CoreV1().ConfigMaps(remoteNamespace.Name).Delete,
+		},
+		scc.eventRecorder,
+	)
+	if err != nil {
+		return progressingConditions, fmt.Errorf("can't prune configmap(s) in %q Datacenter of %q ScyllaDBCluster: %w", dc.Name, naming.ObjRef(sc), err)
 	}
 
 	var errs []error
-	for _, dc := range sc.Spec.Datacenters {
-		clusterClient, err := scc.kubeRemoteClient.Cluster(dc.RemoteKubernetesClusterName)
+	for _, cm := range requiredConfigMaps {
+		_, changed, err := resourceapply.ApplyConfigMap(ctx, clusterClient.CoreV1(), scc.remoteConfigMapLister.Cluster(dc.RemoteKubernetesClusterName), scc.eventRecorder, cm, resourceapply.ApplyOptions{})
+		if changed {
+			controllerhelpers.AddGenericProgressingStatusCondition(&progressingConditions, remoteConfigMapControllerProgressingCondition, cm, "apply", sc.Generation)
+		}
 		if err != nil {
-			errs = append(errs, fmt.Errorf("can't get client to %q cluster: %w", dc.Name, err))
+			errs = append(errs, fmt.Errorf("can't apply configmap: %w", err))
 			continue
 		}
-
-		for _, cm := range requiredConfigMaps[dc.RemoteKubernetesClusterName] {
-			_, changed, err := resourceapply.ApplyConfigMap(ctx, clusterClient.CoreV1(), scc.remoteConfigMapLister.Cluster(dc.RemoteKubernetesClusterName), scc.eventRecorder, cm, resourceapply.ApplyOptions{})
-			if changed {
-				controllerhelpers.AddGenericProgressingStatusCondition(&progressingConditions, remoteConfigMapControllerProgressingCondition, cm, "apply", sc.Generation)
-			}
-			if err != nil {
-				errs = append(errs, fmt.Errorf("can't apply configmap: %w", err))
-				continue
-			}
-		}
 	}
 
 	err = utilerrors.NewAggregate(errs)

pkg/controller/scylladbcluster/resource_test.go

@@ -12,78 +12,58 @@ import (
 	"github.com/scylladb/scylla-operator/pkg/resourceapply"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 )
 
-func (scc *Controller) syncEndpoints(
+func (scc *Controller) syncRemoteEndpoints(
 	ctx context.Context,
 	sc *scyllav1alpha1.ScyllaDBCluster,
+	dc *scyllav1alpha1.ScyllaDBClusterDatacenter,
+	remoteNamespace *corev1.Namespace,
+	remoteController metav1.Object,
+	remoteEndpoints map[string]*corev1.Endpoints,
 	remoteNamespaces map[string]*corev1.Namespace,
-	remoteControllers map[string]metav1.Object,
-	remoteEndpoints map[string]map[string]*corev1.Endpoints,
 	managingClusterDomain string,
 ) ([]metav1.Condition, error) {
-	progressingConditions, requiredEndpointSlices, err := MakeRemoteEndpointSlices(sc, remoteNamespaces, remoteControllers, scc.remoteServiceLister, scc.remotePodLister, managingClusterDomain)
+	progressingConditions, requiredEndpointSlices, err := MakeRemoteEndpointSlices(sc, dc, remoteNamespace, remoteController, remoteNamespaces, scc.remoteServiceLister, scc.remotePodLister, managingClusterDomain)
 	if err != nil {
 		return progressingConditions, fmt.Errorf("can't make endpointslices: %w", err)
 	}
 
-	requiredEndpoints := make(map[string][]*corev1.Endpoints, len(requiredEndpointSlices))
+	requiredEndpoints := make([]*corev1.Endpoints, 0, len(requiredEndpointSlices))
 
-	for _, dc := range sc.Spec.Datacenters {
-		re, err := controllerhelpers.ConvertEndpointSlicesToEndpoints(requiredEndpointSlices[dc.RemoteKubernetesClusterName])
-		if err != nil {
-			return progressingConditions, fmt.Errorf("can't convert endpointslices to endpoints: %w", err)
-		}
-
-		requiredEndpoints[dc.RemoteKubernetesClusterName] = re
+	re, err := controllerhelpers.ConvertEndpointSlicesToEndpoints(requiredEndpointSlices)
+	if err != nil {
+		return progressingConditions, fmt.Errorf("can't convert endpointslices to endpoints: %w", err)
 	}
 
-	// Delete any excessive Endpoints.
-	// Delete has to be the first action to avoid getting stuck on quota.
-	var deletionErrors []error
-	for _, dc := range sc.Spec.Datacenters {
-		ns, ok := remoteNamespaces[dc.RemoteKubernetesClusterName]
-		if !ok {
-			continue
-		}
-
-		clusterClient, err := scc.kubeRemoteClient.Cluster(dc.RemoteKubernetesClusterName)
-		if err != nil {
-			return nil, fmt.Errorf("can't get client to %q cluster: %w", dc.RemoteKubernetesClusterName, err)
-		}
+	requiredEndpoints = append(requiredEndpoints, re...)
 
-		err = controllerhelpers.Prune(ctx,
-			requiredEndpoints[dc.RemoteKubernetesClusterName],
-			remoteEndpoints[dc.RemoteKubernetesClusterName],
-			&controllerhelpers.PruneControlFuncs{
-				DeleteFunc: clusterClient.CoreV1().Endpoints(ns.Name).Delete,
-			},
-			scc.eventRecorder,
-		)
-		if err != nil {
-			return progressingConditions, fmt.Errorf("can't prune endpoints in %q Datacenter of %q ScyllaDBCluster: %w", dc.Name, naming.ObjRef(sc), err)
-		}
+	clusterClient, err := scc.kubeRemoteClient.Cluster(dc.RemoteKubernetesClusterName)
+	if err != nil {
+		return nil, fmt.Errorf("can't get client to %q cluster: %w", dc.RemoteKubernetesClusterName, err)
 	}
 
-	if err := utilerrors.NewAggregate(deletionErrors); err != nil {
-		return nil, fmt.Errorf("can't prune remote endpoints: %w", err)
+	// Delete any excessive Endpoints.
+	// Delete has to be the first action to avoid getting stuck on quota.
+	err = controllerhelpers.Prune(ctx,
+		requiredEndpoints,
+		remoteEndpoints,
+		&controllerhelpers.PruneControlFuncs{
+			DeleteFunc: clusterClient.CoreV1().Endpoints(remoteNamespace.Name).Delete,
+		},
+		scc.eventRecorder,
+	)
+	if err != nil {
+		return progressingConditions, fmt.Errorf("can't prune endpoints in %q Datacenter of %q ScyllaDBCluster: %w", dc.Name, naming.ObjRef(sc), err)
 	}
 
-	for _, dc := range sc.Spec.Datacenters {
-		clusterClient, err := scc.kubeRemoteClient.Cluster(dc.RemoteKubernetesClusterName)
-		if err != nil {
-			return nil, fmt.Errorf("can't get client to %q cluster: %w", dc.Name, err)
+	for _, e := range requiredEndpoints {
+		_, changed, err := resourceapply.ApplyEndpoints(ctx, clusterClient.CoreV1(), scc.remoteEndpointsLister.Cluster(dc.RemoteKubernetesClusterName), scc.eventRecorder, e, resourceapply.ApplyOptions{})
+		if changed {
+			controllerhelpers.AddGenericProgressingStatusCondition(&progressingConditions, remoteEndpointsControllerProgressingCondition, e, "apply", sc.Generation)
 		}
-
-		for _, e := range requiredEndpoints[dc.RemoteKubernetesClusterName] {
-			_, changed, err := resourceapply.ApplyEndpoints(ctx, clusterClient.CoreV1(), scc.remoteEndpointsLister.Cluster(dc.RemoteKubernetesClusterName), scc.eventRecorder, e, resourceapply.ApplyOptions{})
-			if changed {
-				controllerhelpers.AddGenericProgressingStatusCondition(&progressingConditions, remoteEndpointsControllerProgressingCondition, e, "apply", sc.Generation)
-			}
-			if err != nil {
-				return nil, fmt.Errorf("can't apply endpoints: %w", err)
-			}
+		if err != nil {
+			return nil, fmt.Errorf("can't apply endpoints: %w", err)
 		}
 	}
 

pkg/controller/scylladbcluster/sync.go

@@ -13,70 +13,52 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	discoveryv1 "k8s.io/api/discovery/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 )
 
-func (scc *Controller) syncEndpointSlices(
+func (scc *Controller) syncRemoteEndpointSlices(
 	ctx context.Context,
 	sc *scyllav1alpha1.ScyllaDBCluster,
+	dc *scyllav1alpha1.ScyllaDBClusterDatacenter,
+	remoteNamespace *corev1.Namespace,
+	remoteController metav1.Object,
+	remoteEndpointSlices map[string]*discoveryv1.EndpointSlice,
 	remoteNamespaces map[string]*corev1.Namespace,
-	remoteControllers map[string]metav1.Object,
-	remoteEndpointSlices map[string]map[string]*discoveryv1.EndpointSlice,
 	managingClusterDomain string,
 ) ([]metav1.Condition, error) {
-	progressingConditions, requiredEndpointSlices, err := MakeRemoteEndpointSlices(sc, remoteNamespaces, remoteControllers, scc.remoteServiceLister, scc.remotePodLister, managingClusterDomain)
+	progressingConditions, requiredEndpointSlices, err := MakeRemoteEndpointSlices(sc, dc, remoteNamespace, remoteController, remoteNamespaces, scc.remoteServiceLister, scc.remotePodLister, managingClusterDomain)
 	if err != nil {
 		return progressingConditions, fmt.Errorf("can't make endpointslices: %w", err)
 	}
 	if len(progressingConditions) > 0 {
 		return progressingConditions, nil
 	}
 
-	// Delete any excessive EndpointSlices.
-	// Delete has to be the first action to avoid getting stuck on quota.
-	var deletionErrors []error
-	for _, dc := range sc.Spec.Datacenters {
-		ns, ok := remoteNamespaces[dc.RemoteKubernetesClusterName]
-		if !ok {
-			continue
-		}
-
-		clusterClient, err := scc.kubeRemoteClient.Cluster(dc.RemoteKubernetesClusterName)
-		if err != nil {
-			return nil, fmt.Errorf("can't get client to %q cluster: %w", dc.RemoteKubernetesClusterName, err)
-		}
-
-		err = controllerhelpers.Prune(ctx,
-			requiredEndpointSlices[dc.RemoteKubernetesClusterName],
-			remoteEndpointSlices[dc.RemoteKubernetesClusterName],
-			&controllerhelpers.PruneControlFuncs{
-				DeleteFunc: clusterClient.DiscoveryV1().EndpointSlices(ns.Name).Delete,
-			},
-			scc.eventRecorder,
-		)
-		if err != nil {
-			return progressingConditions, fmt.Errorf("can't prune endpointslices in %q Datacenter of %q ScyllaDBCluster: %w", dc.Name, naming.ObjRef(sc), err)
-		}
+	clusterClient, err := scc.kubeRemoteClient.Cluster(dc.RemoteKubernetesClusterName)
+	if err != nil {
+		return nil, fmt.Errorf("can't get client to %q cluster: %w", dc.RemoteKubernetesClusterName, err)
 	}
 
-	if err := utilerrors.NewAggregate(deletionErrors); err != nil {
-		return nil, fmt.Errorf("can't prune remote endpointslice(s): %w", err)
+	// Delete any excessive EndpointSlices.
+	// Delete has to be the first action to avoid getting stuck on quota.
+	err = controllerhelpers.Prune(ctx,
+		requiredEndpointSlices,
+		remoteEndpointSlices,
+		&controllerhelpers.PruneControlFuncs{
+			DeleteFunc: clusterClient.DiscoveryV1().EndpointSlices(remoteNamespace.Name).Delete,
+		},
+		scc.eventRecorder,
+	)
+	if err != nil {
+		return progressingConditions, fmt.Errorf("can't prune endpointslices in %q Datacenter of %q ScyllaDBCluster: %w", dc.Name, naming.ObjRef(sc), err)
 	}
 
-	for _, dc := range sc.Spec.Datacenters {
-		clusterClient, err := scc.kubeRemoteClient.Cluster(dc.RemoteKubernetesClusterName)
-		if err != nil {
-			return nil, fmt.Errorf("can't get client to %q region: %w", dc.Name, err)
+	for _, es := range requiredEndpointSlices {
+		_, changed, err := resourceapply.ApplyEndpointSlice(ctx, clusterClient.DiscoveryV1(), scc.remoteEndpointSliceLister.Cluster(dc.RemoteKubernetesClusterName), scc.eventRecorder, es, resourceapply.ApplyOptions{})
+		if changed {
+			controllerhelpers.AddGenericProgressingStatusCondition(&progressingConditions, remoteEndpointSliceControllerProgressingCondition, es, "apply", sc.Generation)
 		}
-
-		for _, es := range requiredEndpointSlices[dc.RemoteKubernetesClusterName] {
-			_, changed, err := resourceapply.ApplyEndpointSlice(ctx, clusterClient.DiscoveryV1(), scc.remoteEndpointSliceLister.Cluster(dc.RemoteKubernetesClusterName), scc.eventRecorder, es, resourceapply.ApplyOptions{})
-			if changed {
-				controllerhelpers.AddGenericProgressingStatusCondition(&progressingConditions, remoteEndpointSliceControllerProgressingCondition, es, "apply", sc.Generation)
-			}
-			if err != nil {
-				return nil, fmt.Errorf("can't apply endpointslice: %w", err)
-			}
+		if err != nil {
+			return nil, fmt.Errorf("can't apply endpointslice: %w", err)
 		}
 	}