move replacing nodes to aggregation

Author: rzetelskik

pkg/cmd/operator/bootstrapgate.go

@@ -12,13 +12,11 @@ import (
 	"github.com/scylladb/scylla-operator/pkg/controllerhelpers"
 	"github.com/scylladb/scylla-operator/pkg/genericclioptions"
 	"github.com/scylladb/scylla-operator/pkg/naming"
-	"github.com/scylladb/scylla-operator/pkg/pointer"
 	"github.com/scylladb/scylla-operator/pkg/signals"
 	"github.com/scylladb/scylla-operator/pkg/version"
 	"github.com/spf13/cobra"
 	corev1 "k8s.io/api/core/v1"
 	apimachineryvalidation "k8s.io/apimachinery/pkg/api/validation"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	apimachineryutilerrors "k8s.io/apimachinery/pkg/util/errors"
 	apimachineryutilsets "k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/client-go/kubernetes"
@@ -35,9 +33,6 @@ type AutoGenerateSSTables struct {
 					Bootstrapped struct {
 						Value string `json:"value"`
 					} `json:"bootstrapped"`
-					HostID struct {
-						Value string `json:"value"`
-					} `json:"host_id"`
 				} `json:"columns"`
 			} `json:"clustering_elements"`
 		} `json:"anonymous"`
@@ -53,7 +48,6 @@ type BootstrapGateOptions struct {
 	SStableSystemLocalDumpPath string
 
 	bootstrapped bool
-	hostID       *string
 
 	kubeClient   kubernetes.Interface
 	scyllaClient scyllaversionedclient.Interface
@@ -161,9 +155,6 @@ func (o *BootstrapGateOptions) Complete(args []string) error {
 			if ce.Columns.Bootstrapped.Value == "COMPLETED" {
 				o.bootstrapped = true
 			}
-			if len(ce.Columns.HostID.Value) != 0 {
-				o.hostID = pointer.Ptr(ce.Columns.HostID.Value)
-			}
 		}
 	}
 
@@ -207,43 +198,13 @@ func (o *BootstrapGateOptions) Run(originalStreams genericclioptions.IOStreams,
 
 func (o *BootstrapGateOptions) Execute(ctx context.Context, originalStreams genericclioptions.IOStreams, cmd *cobra.Command) error {
 	var err error
-	ignoredNodes := apimachineryutilsets.New[string]()
 
-	// FIXME
-	// Skip the rolling restart case for now.
 	if o.bootstrapped {
-		if o.hostID == nil {
-			return fmt.Errorf("node is bootstrapped but hostID is nil, this should never happen")
-		}
-
-		// Node is restarting, other nodes are going to consider it down.
-		// FIXME: we should add it to ignored nodes instead if we want to support blocking the rolling restart.
-		klog.V(2).InfoS("Node has already been bootstrapped, assuming it's being rolling restarted. Skipping.", "Service", naming.ManualRef(o.Namespace, o.ServiceName), "Bootstrapped", o.bootstrapped, "HostID", o.hostID)
+		klog.V(2).InfoS("Node has already been bootstrapped, assuming it's being rolling restarted. Skipping.", "Service", naming.ManualRef(o.Namespace, o.ServiceName), "Bootstrapped", o.bootstrapped)
 		return nil
 	}
-	//
-
-	// Node replace case.
-	var svc *corev1.Service
-	svc, err = o.kubeClient.CoreV1().Services(o.Namespace).Get(ctx, o.ServiceName, metav1.GetOptions{})
-	if err != nil {
-		return fmt.Errorf("can't get service %q: %w", naming.ManualRef(o.Namespace, o.ServiceName), err)
-	}
 
-	replacingHostID, ok := svc.Labels[naming.ReplacingNodeHostIDLabel]
-	if ok && len(replacingHostID) == 0 {
-		return fmt.Errorf("service %q has an empty %q label, this should never happen", naming.ManualRef(o.Namespace, o.ServiceName), naming.ReplacingNodeHostIDLabel)
-	}
-
-	if len(replacingHostID) > 0 {
-		klog.V(2).InfoS("Node is replacing another node", "Service", naming.ManualRef(o.Namespace, o.ServiceName), "Bootstrapped", o.bootstrapped, "HostID", o.hostID, "ReplacingHostID", replacingHostID)
-		ignoredNodes.Insert(replacingHostID)
-	}
-	//
-
-	// ALTERNATIVE: instead of getting the configmap with statuses, this is where we could ask all other nodes for their statuses (by using broadcastAddresses from corresponding services/pods)
-
-	klog.V(2).InfoS("Waiting for required bootstrap conditions", "Service", naming.ManualRef(o.Namespace, o.ServiceName), "Bootstrapped", o.bootstrapped, "HostID", o.hostID, "ReplacingHostID", replacingHostID, "IgnoredNodes", ignoredNodes.UnsortedList())
+	klog.V(2).InfoS("Waiting for required bootstrap conditions", "Service", naming.ManualRef(o.Namespace, o.ServiceName), "Bootstrapped", o.bootstrapped)
 	_, err = controllerhelpers.WaitForConfigMapState(
 		ctx,
 		o.kubeClient.CoreV1().ConfigMaps(o.Namespace),
@@ -255,16 +216,16 @@ func (o *BootstrapGateOptions) Execute(ctx context.Context, originalStreams gene
 				return false, fmt.Errorf("missing %s key", naming.ScyllaDBClusterStatusKey)
 			}
 
-			var nodeClusterStatuses []controllerhelpers.ClusterStatus
-			err = json.Unmarshal([]byte(data), &nodeClusterStatuses)
+			var clusterStatus controllerhelpers.ClusterStatus
+			err = json.Unmarshal([]byte(data), &clusterStatus)
 			if err != nil {
 				return false, fmt.Errorf("can't unmarshal cluster status: %w", err)
 			}
 
 			// FIXME: this could be preprocessed
 			hostIDs := apimachineryutilsets.New[string]()
 			hostIDToStatusMap := make(map[string]map[string]bool)
-			for _, nodeClusterStatus := range nodeClusterStatuses {
+			for _, nodeClusterStatus := range clusterStatus.NodeClusterStatuses {
 				if nodeClusterStatus.Error != nil {
 					// We might get errors from nodes that haven't started bootstrap yet, so we need to ignore them.
 					// If the node is already a part of the cluster, we'll see it in other nodes' statuses.
@@ -281,7 +242,10 @@ func (o *BootstrapGateOptions) Execute(ctx context.Context, originalStreams gene
 				hostIDToStatusMap[nodeClusterStatus.HostID] = otherNodesHostIDToStatusMap
 			}
 
-			klog.V(4).InfoS("Cluster status", "Service", naming.ManualRef(o.Namespace, o.ServiceName), "HostIDs", hostIDs.UnsortedList(), "HostIDToStatusMap", hostIDToStatusMap)
+			// Ignore nodes that are being replaced.
+			ignoredNodes := apimachineryutilsets.New[string](clusterStatus.ReplacingHostIDs...)
+
+			klog.V(4).InfoS("Cluster status", "Service", naming.ManualRef(o.Namespace, o.ServiceName), "HostIDs", hostIDs.UnsortedList(), "HostIDToStatusMap", hostIDToStatusMap, "IgnoredNodes", ignoredNodes.UnsortedList())
 			for hostID := range hostIDs {
 				if ignoredNodes.Has(hostID) {
 					// We don't care about this node.

pkg/cmd/operator/scylladbapistatusupdate.go

@@ -265,17 +265,17 @@ func (c *controller) sync(ctx context.Context) error {
 		return fmt.Errorf("can't get service %q: %w", naming.ManualRef(c.namespace, c.serviceName), err)
 	}
 
-	clusterStatus, err := getClusterStatus(ctx)
+	nodeClusterStatus, err := getNodeClusterStatus(ctx)
 	if err != nil {
 		return fmt.Errorf("can't get cluster status: %w", err)
 	}
 
-	statusAnnotationValue, err := makeClusterStatusAnnotationValue(clusterStatus)
+	statusAnnotationValue, err := makeNodeClusterStatusAnnotationValue(nodeClusterStatus)
 	if err != nil {
 		return fmt.Errorf("can't make cluster status annotation value: %w", err)
 	}
 
-	patch, err := controllerhelpers.PrepareSetAnnotationPatch(svc, naming.ClusterStatusAnnotation, &statusAnnotationValue)
+	patch, err := controllerhelpers.PrepareSetAnnotationPatch(svc, naming.NodeClusterStatusAnnotation, &statusAnnotationValue)
 	if err != nil {
 		return fmt.Errorf("can't prepare annotation patch: %w", err)
 	}
@@ -288,7 +288,7 @@ func (c *controller) sync(ctx context.Context) error {
 	return nil
 }
 
-func getClusterStatus(ctx context.Context) (*controllerhelpers.ClusterStatus, error) {
+func getNodeClusterStatus(ctx context.Context) (*controllerhelpers.NodeClusterStatus, error) {
 	scyllaClient, err := controllerhelpers.NewScyllaClientForLocalhost()
 	if err != nil {
 		return nil, fmt.Errorf("can't create Scylla client for localhost: %w", err)
@@ -299,32 +299,32 @@ func getClusterStatus(ctx context.Context) (*controllerhelpers.ClusterStatus, er
 	nodeStatuses, err := scyllaClient.Status(ctx, localhost)
 	if err != nil {
 		klog.V(4).InfoS("Error getting node statuses", "error", err)
-		return &controllerhelpers.ClusterStatus{
+		return &controllerhelpers.NodeClusterStatus{
 			Error: pointer.Ptr(err.Error()),
 		}, nil
 	}
 
 	hostID, err := scyllaClient.GetLocalHostId(ctx, localhost, false)
 	if err != nil {
 		klog.V(4).InfoS("Error getting local host ID", "error", err)
-		return &controllerhelpers.ClusterStatus{
+		return &controllerhelpers.NodeClusterStatus{
 			Error: pointer.Ptr(err.Error()),
 		}, nil
 	}
 
-	return &controllerhelpers.ClusterStatus{
+	return &controllerhelpers.NodeClusterStatus{
 		HostID: hostID,
 		Status: nodeStatuses,
 	}, nil
 }
 
-func makeClusterStatusAnnotationValue(status *controllerhelpers.ClusterStatus) (string, error) {
+func makeNodeClusterStatusAnnotationValue(status *controllerhelpers.NodeClusterStatus) (string, error) {
 	var err error
 	buf := bytes.Buffer{}
 
 	err = json.NewEncoder(&buf).Encode(status)
 	if err != nil {
-		return "", fmt.Errorf("can't encode status: %w", err)
+		return "", fmt.Errorf("can't encode node cluster status: %w", err)
 	}
 
 	return buf.String(), nil

pkg/controller/scylladbcluster/resource.go

@@ -1544,7 +1544,10 @@ func makeLocalClusterStatusConfigMap(sc *scyllav1alpha1.ScyllaDBCluster, remoteN
 		return progressingConditions, nil, fmt.Errorf("can't get cluster statuses configmap name for ScyllaDBCluster %q: %w", naming.ObjRef(sc), err)
 	}
 
-	var nodeClusterStatuses []controllerhelpers.ClusterStatus
+	clusterStatus := &controllerhelpers.ClusterStatus{
+		ReplacingHostIDs:    []string{},
+		NodeClusterStatuses: []controllerhelpers.NodeClusterStatus{},
+	}
 	for _, dc := range sc.Spec.Datacenters {
 		dcNamespace, ok := remoteNamespaces[dc.RemoteKubernetesClusterName]
 		if !ok {
@@ -1561,18 +1564,19 @@ func makeLocalClusterStatusConfigMap(sc *scyllav1alpha1.ScyllaDBCluster, remoteN
 
 		// TODO: move to func
 		dcServiceSelector := naming.DatacenterMemberServiceSelector(sc, &dc)
-		dcNodeClusterStatuses, err := getNodeClusterStatusesFromRemoteDCMemberServices(sc, &dc, dcNamespace, dcServiceSelector, remoteServiceLister)
+		dcClusterStatus, err := getClusterStatusFromRemoteDCMemberServices(sc, &dc, dcNamespace, dcServiceSelector, remoteServiceLister)
 		if err != nil {
-			return progressingConditions, nil, fmt.Errorf("can't get node cluster statuses for ScyllaDBCluster %q Datacenter %q remote member services: %w", naming.ObjRef(sc), dc.Name, err)
+			return progressingConditions, nil, fmt.Errorf("can't get cluster status for ScyllaDBCluster %q Datacenter %q remote member services: %w", naming.ObjRef(sc), dc.Name, err)
 		}
 
-		nodeClusterStatuses = append(nodeClusterStatuses, dcNodeClusterStatuses...)
+		clusterStatus.ReplacingHostIDs = append(clusterStatus.ReplacingHostIDs, dcClusterStatus.ReplacingHostIDs...)
+		clusterStatus.NodeClusterStatuses = append(clusterStatus.NodeClusterStatuses, dcClusterStatus.NodeClusterStatuses...)
 	}
 
 	buf := bytes.Buffer{}
-	err = json.NewEncoder(&buf).Encode(nodeClusterStatuses)
+	err = json.NewEncoder(&buf).Encode(clusterStatus)
 	if err != nil {
-		return progressingConditions, nil, fmt.Errorf("can't encode node cluster statuses: %w", err)
+		return progressingConditions, nil, fmt.Errorf("can't encode cluster status: %w", err)
 	}
 
 	cm := &corev1.ConfigMap{
@@ -1611,30 +1615,32 @@ func makeLocalClusterStatusConfigMap(sc *scyllav1alpha1.ScyllaDBCluster, remoteN
 	return progressingConditions, cm, nil
 }
 
-func getNodeClusterStatusesFromRemoteDCMemberServices(sc *scyllav1alpha1.ScyllaDBCluster, remoteDC *scyllav1alpha1.ScyllaDBClusterDatacenter, remoteDCNamespace *corev1.Namespace, remoteDCMemberServiceSelector apimachinerylabels.Selector, remoteServiceLister remotelister.GenericClusterLister[corev1listers.ServiceLister]) ([]controllerhelpers.ClusterStatus, error) {
-	var nodeClusterStatuses []controllerhelpers.ClusterStatus
+func getClusterStatusFromRemoteDCMemberServices(sc *scyllav1alpha1.ScyllaDBCluster, remoteDC *scyllav1alpha1.ScyllaDBClusterDatacenter, remoteDCNamespace *corev1.Namespace, remoteDCMemberServiceSelector apimachinerylabels.Selector, remoteServiceLister remotelister.GenericClusterLister[corev1listers.ServiceLister]) (*controllerhelpers.ClusterStatus, error) {
+	clusterStatus := &controllerhelpers.ClusterStatus{
+		ReplacingHostIDs:    []string{},
+		NodeClusterStatuses: []controllerhelpers.NodeClusterStatus{},
+	}
 
 	dcMemberServices, err := remoteServiceLister.Cluster(remoteDC.RemoteKubernetesClusterName).Services(remoteDCNamespace.Name).List(remoteDCMemberServiceSelector)
 	if err != nil {
 		return nil, fmt.Errorf("can't list member Services for %q ScyllaDBCluster %q Datacenter: %w", naming.ObjRef(sc), remoteDC.Name, err)
 	}
 
 	for _, svc := range dcMemberServices {
-		nodeClusterStatusAnnotation, ok := svc.Annotations[naming.ClusterStatusAnnotation]
-		if !ok {
-			// The node might not have reported its status yet.
-			// FIXME: should we wait? Sidecar would have to be moved to init containers.
-			continue
+		if replacingNodeHostIDLabel, ok := svc.Labels[naming.ReplacingNodeHostIDLabel]; ok {
+			clusterStatus.ReplacingHostIDs = append(clusterStatus.ReplacingHostIDs, replacingNodeHostIDLabel)
 		}
 
-		var nodeClusterStatus controllerhelpers.ClusterStatus
-		err = json.NewDecoder(strings.NewReader(nodeClusterStatusAnnotation)).Decode(&nodeClusterStatus)
-		if err != nil {
-			return nil, fmt.Errorf("can't decode annotation %q of service %q for ScyllaDBCluster %q Datacenter %q: %w", naming.ClusterStatusAnnotation, naming.ObjRef(svc), naming.ObjRef(sc), remoteDC.Name, err)
-		}
+		if nodeClusterStatusAnnotation, ok := svc.Annotations[naming.NodeClusterStatusAnnotation]; ok {
+			var nodeClusterStatus controllerhelpers.NodeClusterStatus
+			err = json.NewDecoder(strings.NewReader(nodeClusterStatusAnnotation)).Decode(&nodeClusterStatus)
+			if err != nil {
+				return nil, fmt.Errorf("can't decode annotation %q of service %q for ScyllaDBCluster %q Datacenter %q: %w", naming.NodeClusterStatusAnnotation, naming.ObjRef(svc), naming.ObjRef(sc), remoteDC.Name, err)
+			}
 
-		nodeClusterStatuses = append(nodeClusterStatuses, nodeClusterStatus)
+			clusterStatus.NodeClusterStatuses = append(clusterStatus.NodeClusterStatuses, nodeClusterStatus)
+		}
 	}
 
-	return nodeClusterStatuses, nil
+	return clusterStatus, nil
 }

pkg/controller/scylladbdatacenter/resource.go

@@ -2245,7 +2245,7 @@ func cloneMapExcludingKeysOrEmpty[M ~map[K]V, S ~[]K, K comparable, V any](m M,
 	return r
 }
 
-func MakeClusterStatusConfigMap(sdc *scyllav1alpha1.ScyllaDBDatacenter, nodeClusterStatuses []controllerhelpers.ClusterStatus) (*corev1.ConfigMap, error) {
+func MakeClusterStatusConfigMap(sdc *scyllav1alpha1.ScyllaDBDatacenter, clusterStatus *controllerhelpers.ClusterStatus) (*corev1.ConfigMap, error) {
 	var err error
 
 	name, err := naming.ScyllaDBDatacenterClusterStatusesConfigMapName(sdc)
@@ -2254,9 +2254,9 @@ func MakeClusterStatusConfigMap(sdc *scyllav1alpha1.ScyllaDBDatacenter, nodeClus
 	}
 
 	buf := bytes.Buffer{}
-	err = json.NewEncoder(&buf).Encode(nodeClusterStatuses)
+	err = json.NewEncoder(&buf).Encode(clusterStatus)
 	if err != nil {
-		return nil, fmt.Errorf("can't encode node cluster statuses: %w", err)
+		return nil, fmt.Errorf("can't encode node cluster status: %w", err)
 	}
 
 	labels := cloneMapExcludingKeysOrEmpty(sdc.Labels, nonPropagatedLabelKeys)

pkg/controller/scylladbdatacenter/sync_clusterstatus.go

@@ -25,10 +25,9 @@ func (sdcc *Controller) syncClusterStatusConfigMap(
 	var progressingConditions []metav1.Condition
 	var err error
 
-	var nodeClusterStatuses []controllerhelpers.ClusterStatus
+	var clusterStatus *controllerhelpers.ClusterStatus
 	if clusterStatusOverrideConfigMapRef, ok := sdc.Annotations[naming.ScyllaDBClusterStatusOverrideConfigMapRefAnnotation]; ok {
 		// TODO: move to func
-
 		overrideCM, err := sdcc.configMapLister.ConfigMaps(sdc.Namespace).Get(clusterStatusOverrideConfigMapRef)
 		if err != nil {
 			if !apierrors.IsNotFound(err) {
@@ -44,23 +43,26 @@ func (sdcc *Controller) syncClusterStatusConfigMap(
 			return progressingConditions, nil
 		}
 
+		var overrideClusterStatus controllerhelpers.ClusterStatus
 		data, ok := overrideCM.Data[naming.ScyllaDBClusterStatusKey]
 		if !ok {
 			return progressingConditions, fmt.Errorf("missing %s key", naming.ScyllaDBClusterStatusKey)
 		}
 
-		err = json.Unmarshal([]byte(data), &nodeClusterStatuses)
+		err = json.Unmarshal([]byte(data), &overrideClusterStatus)
 		if err != nil {
 			return progressingConditions, fmt.Errorf("can't unmarshal cluster status: %w", err)
 		}
+
+		clusterStatus = &overrideClusterStatus
 	} else {
-		nodeClusterStatuses, err = getNodeClusterStatuses(sdc, services)
+		clusterStatus, err = getClusterStatus(sdc, services)
 		if err != nil {
 			return progressingConditions, fmt.Errorf("can't get node cluster statuses: %w", err)
 		}
 	}
 
-	requiredConfigMap, err := MakeClusterStatusConfigMap(sdc, nodeClusterStatuses)
+	requiredConfigMap, err := MakeClusterStatusConfigMap(sdc, clusterStatus)
 	if err != nil {
 		return progressingConditions, fmt.Errorf("can't make cluster status configmap: %w", err)
 	}
@@ -76,8 +78,11 @@ func (sdcc *Controller) syncClusterStatusConfigMap(
 	return progressingConditions, nil
 }
 
-func getNodeClusterStatuses(sdc *scyllav1alpha1.ScyllaDBDatacenter, services map[string]*corev1.Service) ([]controllerhelpers.ClusterStatus, error) {
-	var nodeClusterStatuses []controllerhelpers.ClusterStatus
+func getClusterStatus(sdc *scyllav1alpha1.ScyllaDBDatacenter, services map[string]*corev1.Service) (*controllerhelpers.ClusterStatus, error) {
+	clusterStatus := &controllerhelpers.ClusterStatus{
+		ReplacingHostIDs:    []string{},
+		NodeClusterStatuses: []controllerhelpers.NodeClusterStatus{},
+	}
 
 	for _, rack := range sdc.Spec.Racks {
 		stsName := naming.StatefulSetNameForRack(rack, sdc)
@@ -96,22 +101,21 @@ func getNodeClusterStatuses(sdc *scyllav1alpha1.ScyllaDBDatacenter, services map
 				continue
 			}
 
-			nodeClusterStatusAnnotation, ok := svc.Annotations[naming.ClusterStatusAnnotation]
-			if !ok {
-				// The node might not have reported its status yet.
-				// FIXME: should we wait? Sidecar would have to be moved to init containers.
-				continue
+			if replacingHostIDAnnotation, ok := svc.Annotations[naming.ReplacingNodeHostIDLabel]; ok {
+				clusterStatus.ReplacingHostIDs = append(clusterStatus.ReplacingHostIDs, replacingHostIDAnnotation)
 			}
 
-			var nodeClusterStatus controllerhelpers.ClusterStatus
-			err = json.NewDecoder(strings.NewReader(nodeClusterStatusAnnotation)).Decode(&nodeClusterStatus)
-			if err != nil {
-				return nil, fmt.Errorf("can't decode annotation %q of service %q for ScyllaDBDatacenter %q: %w", naming.ClusterStatusAnnotation, naming.ManualRef(sdc.Namespace, svcName), naming.ObjRef(sdc), err)
-			}
+			if nodeClusterStatusAnnotation, ok := svc.Annotations[naming.NodeClusterStatusAnnotation]; ok {
+				var nodeClusterStatus controllerhelpers.NodeClusterStatus
+				err = json.NewDecoder(strings.NewReader(nodeClusterStatusAnnotation)).Decode(&nodeClusterStatus)
+				if err != nil {
+					return nil, fmt.Errorf("can't decode annotation %q of service %q for ScyllaDBDatacenter %q: %w", naming.NodeClusterStatusAnnotation, naming.ManualRef(sdc.Namespace, svcName), naming.ObjRef(sdc), err)
+				}
 
-			nodeClusterStatuses = append(nodeClusterStatuses, nodeClusterStatus)
+				clusterStatus.NodeClusterStatuses = append(clusterStatus.NodeClusterStatuses, nodeClusterStatus)
+			}
 		}
 	}
 
-	return nodeClusterStatuses, nil
+	return clusterStatus, nil
 }

pkg/naming/constants.go

@@ -47,7 +47,7 @@ const (
 	// CleanupJobTokenRingHashAnnotation reflects which version of token ring cleanup Job is cleaning.
 	CleanupJobTokenRingHashAnnotation = "internal.scylla-operator.scylladb.com/cleanup-token-ring-hash"
 
-	ClusterStatusAnnotation = "internal.scylla-operator.scylladb.com/cluster-status"
+	NodeClusterStatusAnnotation = "internal.scylla-operator.scylladb.com/cluster-status"
 )
 
 // Annotations used for feature backward compatibility between v1.ScyllaCluster and v1alpha1.ScyllaDBDatacenter