Extend E2E utils with machinery for ScyllaDBDatacenters

Author: rzetelskik

test/e2e/utils/helpers.go

@@ -28,6 +28,7 @@ import (
 	"github.com/scylladb/scylla-operator/pkg/scyllaclient"
 	"github.com/scylladb/scylla-operator/pkg/util/hash"
 	"github.com/scylladb/scylla-operator/test/e2e/framework"
+	utilsv1alpha1 "github.com/scylladb/scylla-operator/test/e2e/utils/v1alpha1"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -281,24 +282,7 @@ func GetStatefulSetsForScyllaCluster(ctx context.Context, client appv1client.App
 }
 
 func GetPodsForStatefulSet(ctx context.Context, client corev1client.CoreV1Interface, sts *appsv1.StatefulSet) (map[string]*corev1.Pod, error) {
-	selector, err := metav1.LabelSelectorAsSelector(sts.Spec.Selector)
-	if err != nil {
-		return nil, fmt.Errorf("can't convert StatefulSet %q selector: %w", naming.ObjRef(sts), err)
-	}
-
-	podList, err := client.Pods(sts.Namespace).List(ctx, metav1.ListOptions{
-		LabelSelector: selector.String(),
-	})
-	if err != nil {
-		return nil, fmt.Errorf("can't list Pods for StatefulSet %q: %w", naming.ObjRef(sts), err)
-	}
-
-	res := map[string]*corev1.Pod{}
-	for _, pod := range podList.Items {
-		res[pod.Name] = &pod
-	}
-
-	return res, nil
+	return utilsv1alpha1.GetPodsForStatefulSet(ctx, client, sts)
 }
 
 func GetDaemonSetsForNodeConfig(ctx context.Context, client appv1client.AppsV1Interface, nc *scyllav1alpha1.NodeConfig) ([]*appsv1.DaemonSet, error) {

test/e2e/utils/v1alpha1/config.go

@@ -0,0 +1,20 @@
+// Copyright (C) 2025 ScyllaDB
+
+package v1alpha1
+
+import "time"
+
+const (
+	// SyncTimeout is the maximum time the sync loop can last. In normal circumstances this is in the order
+	// of seconds but there are special cases we need to account for. Like when the sync loop generates new keys
+	// and signs certificates it can take several seconds. When the CI creates multiple clusters in parallel on
+	// a constrained CPU, one cert can easily take over 30s.
+	SyncTimeout        = 2 * time.Minute
+	imagePullTimeout   = 4 * time.Minute
+	joinClusterTimeout = 3 * time.Minute
+	cleanupJobTimeout  = 1 * time.Minute
+
+	// memberRolloutTimeout is the maximum amount of time it takes to start a scylla pod and become ready.
+	// It includes the time to pull the images, copy the necessary files (sidecar), join the cluster and similar.
+	memberRolloutTimeout = 30*time.Second + imagePullTimeout + joinClusterTimeout
+)

test/e2e/utils/v1alpha1/helpers.go

@@ -4,18 +4,30 @@ package v1alpha1
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"reflect"
+	"sort"
+	"strings"
+	"time"
+
+	scyllav1 "github.com/scylladb/scylla-operator/pkg/api/scylla/v1"
 	scyllav1alpha1 "github.com/scylladb/scylla-operator/pkg/api/scylla/v1alpha1"
 	"github.com/scylladb/scylla-operator/pkg/controllerhelpers"
 	"github.com/scylladb/scylla-operator/pkg/helpers"
 	"github.com/scylladb/scylla-operator/pkg/helpers/slices"
 	"github.com/scylladb/scylla-operator/pkg/naming"
 	"github.com/scylladb/scylla-operator/pkg/pointer"
 	"github.com/scylladb/scylla-operator/pkg/scyllaclient"
+	"github.com/scylladb/scylla-operator/test/e2e/framework"
+	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/util/wait"
+	appv1client "k8s.io/client-go/kubernetes/typed/apps/v1"
 	corev1client "k8s.io/client-go/kubernetes/typed/core/v1"
+	"k8s.io/klog/v2"
 )
 
 func GetScyllaClient(ctx context.Context, client corev1client.CoreV1Interface, sdc *scyllav1alpha1.ScyllaDBDatacenter) (*scyllaclient.Client, []string, error) {
@@ -160,4 +172,153 @@ func GetBroadcastAddress(ctx context.Context, client corev1client.CoreV1Interfac
 	}
 
 	return broadcastAddress, nil
+
+}
+
+func ContextForRollout(parent context.Context, sdc *scyllav1alpha1.ScyllaDBDatacenter) (context.Context, context.CancelFunc) {
+	return context.WithTimeout(parent, RolloutTimeoutForScyllaDBDatacenter(sdc))
+}
+
+func RolloutTimeoutForScyllaDBDatacenter(sdc *scyllav1alpha1.ScyllaDBDatacenter) time.Duration {
+	return SyncTimeout + time.Duration(GetNodeCount(sdc))*memberRolloutTimeout + cleanupJobTimeout
+}
+
+func GetNodeCount(sdc *scyllav1alpha1.ScyllaDBDatacenter) int32 {
+	nodes := int32(0)
+	rackTemplateNodes := int32(0)
+
+	if sdc.Spec.RackTemplate != nil && sdc.Spec.RackTemplate.Nodes != nil {
+		rackTemplateNodes = *sdc.Spec.RackTemplate.Nodes
+	}
+
+	for _, r := range sdc.Spec.Racks {
+		if r.Nodes != nil {
+			nodes += *r.Nodes
+		} else {
+			nodes += rackTemplateNodes
+		}
+	}
+
+	return nodes
+}
+
+func IsScyllaDBDatacenterRolledOut(sdc *scyllav1alpha1.ScyllaDBDatacenter) (bool, error) {
+	if !helpers.IsStatusConditionPresentAndTrue(sdc.Status.Conditions, scyllav1.AvailableCondition, sdc.Generation) {
+		return false, nil
+	}
+
+	if !helpers.IsStatusConditionPresentAndFalse(sdc.Status.Conditions, scyllav1.ProgressingCondition, sdc.Generation) {
+		return false, nil
+	}
+
+	if !helpers.IsStatusConditionPresentAndFalse(sdc.Status.Conditions, scyllav1.DegradedCondition, sdc.Generation) {
+		return false, nil
+	}
+
+	framework.Infof("ScyllaDBDatacenter %s (RV=%s) is rolled out", klog.KObj(sdc), sdc.ResourceVersion)
+
+	return true, nil
+}
+
+func IsScyllaDBManagerClusterRegistrationRolledOut(smcr *scyllav1alpha1.ScyllaDBManagerClusterRegistration) (bool, error) {
+	if !helpers.IsStatusConditionPresentAndFalse(smcr.Status.Conditions, scyllav1.ProgressingCondition, smcr.Generation) {
+		return false, nil
+	}
+
+	if !helpers.IsStatusConditionPresentAndFalse(smcr.Status.Conditions, scyllav1.DegradedCondition, smcr.Generation) {
+		return false, nil
+	}
+
+	framework.Infof("ScyllaDBManagerClusterRegistration %s (RV=%s) is rolled out", klog.KObj(smcr), smcr.ResourceVersion)
+
+	return true, nil
+}
+
+func GetStatefulSetsForScyllaDBDatacenter(ctx context.Context, client appv1client.AppsV1Interface, sdc *scyllav1alpha1.ScyllaDBDatacenter) (map[string]*appsv1.StatefulSet, error) {
+	statefulsetList, err := client.StatefulSets(sdc.Namespace).List(ctx, metav1.ListOptions{
+		LabelSelector: labels.Set{
+			naming.ClusterNameLabel: sdc.Name,
+		}.AsSelector().String(),
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	res := map[string]*appsv1.StatefulSet{}
+	for _, s := range statefulsetList.Items {
+		rackName := s.Labels[naming.RackNameLabel]
+		res[rackName] = &s
+	}
+
+	return res, nil
+}
+
+func GetPodsForStatefulSet(ctx context.Context, client corev1client.CoreV1Interface, sts *appsv1.StatefulSet) (map[string]*corev1.Pod, error) {
+	selector, err := metav1.LabelSelectorAsSelector(sts.Spec.Selector)
+	if err != nil {
+		return nil, fmt.Errorf("can't convert StatefulSet %q selector: %w", naming.ObjRef(sts), err)
+	}
+
+	podList, err := client.Pods(sts.Namespace).List(ctx, metav1.ListOptions{
+		LabelSelector: selector.String(),
+	})
+	if err != nil {
+		return nil, fmt.Errorf("can't list Pods for StatefulSet %q: %w", naming.ObjRef(sts), err)
+	}
+
+	res := map[string]*corev1.Pod{}
+	for _, pod := range podList.Items {
+		res[pod.Name] = &pod
+	}
+
+	return res, nil
+}
+
+// TODO: Should be unified with function coming from test/helpers once e2e's there starts using ScyllaDBDatacenter API.
+func WaitForFullQuorum(ctx context.Context, client corev1client.CoreV1Interface, sdc *scyllav1alpha1.ScyllaDBDatacenter, sortedExpectedHosts []string) error {
+	scyllaClient, hosts, err := GetScyllaClient(ctx, client, sdc)
+	if err != nil {
+		return fmt.Errorf("can't get scylla client: %w", err)
+	}
+	defer scyllaClient.Close()
+
+	// Wait for node status to propagate and reach consistency.
+	// This can take a while so let's set a large enough timeout to avoid flakes.
+	return wait.PollImmediateWithContext(ctx, 1*time.Second, 5*time.Minute, func(ctx context.Context) (done bool, err error) {
+		allSeeAllAsUN := true
+		infoMessages := make([]string, 0, len(hosts))
+		var errs []error
+		for _, h := range hosts {
+			var s scyllaclient.NodeStatusInfoSlice
+			s, err = scyllaClient.Status(ctx, h)
+			if err != nil {
+				return true, fmt.Errorf("can't get scylla status on node %q: %w", h, err)
+			}
+
+			sHosts := s.Hosts()
+			sort.Strings(sHosts)
+			if !reflect.DeepEqual(sHosts, sortedExpectedHosts) {
+				errs = append(errs, fmt.Errorf("node %q thinks the cluster consists of different nodes: %s", h, sHosts))
+			}
+
+			downHosts := s.DownHosts()
+			infoMessages = append(infoMessages, fmt.Sprintf("Node %q, down: %q, up: %q", h, strings.Join(downHosts, "\n"), strings.Join(s.LiveHosts(), ",")))
+
+			if len(downHosts) != 0 {
+				allSeeAllAsUN = false
+			}
+		}
+
+		if !allSeeAllAsUN {
+			framework.Infof("ScyllaDB nodes have not reached status consistency yet. Statuses:\n%s", strings.Join(infoMessages, ","))
+		}
+
+		err = errors.Join(errs...)
+		if err != nil {
+			framework.Infof("ScyllaDB nodes encountered an error. Statuses:\n%s", strings.Join(infoMessages, ","))
+			return true, err
+		}
+
+		return allSeeAllAsUN, nil
+	})
 }

test/e2e/utils/verification/scyllacluster/cql.go

@@ -9,6 +9,7 @@ import (
 	"github.com/scylladb/scylla-operator/test/e2e/framework"
 	"github.com/scylladb/scylla-operator/test/e2e/scheme"
 	"github.com/scylladb/scylla-operator/test/e2e/utils"
+	scylladbdatacenterverification "github.com/scylladb/scylla-operator/test/e2e/utils/verification/scylladbdatacenter"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	corev1client "k8s.io/client-go/kubernetes/typed/core/v1"
@@ -27,22 +28,11 @@ func WaitForFullMultiDCQuorum(ctx context.Context, dcClientMap map[string]corev1
 }
 
 func VerifyCQLData(ctx context.Context, di *utils.DataInserter) {
-	err := di.AwaitSchemaAgreement(ctx)
-	o.Expect(err).NotTo(o.HaveOccurred())
-
-	framework.By("Verifying the data")
-	data, err := di.Read()
-	o.Expect(err).NotTo(o.HaveOccurred())
-	o.Expect(data).To(o.Equal(di.GetExpected()))
+	scylladbdatacenterverification.VerifyCQLData(ctx, di)
 }
 
 func InsertAndVerifyCQLData(ctx context.Context, hosts []string, options ...utils.DataInserterOption) *utils.DataInserter {
-	framework.By("Inserting data")
-	di, err := utils.NewDataInserter(hosts, options...)
-	o.Expect(err).NotTo(o.HaveOccurred())
-
-	InsertAndVerifyCQLDataUsingDataInserter(ctx, di)
-	return di
+	return scylladbdatacenterverification.InsertAndVerifyCQLData(ctx, hosts, options...)
 }
 
 func InsertAndVerifyCQLDataByDC(ctx context.Context, hosts map[string][]string) *utils.DataInserter {
@@ -54,13 +44,7 @@ func InsertAndVerifyCQLDataByDC(ctx context.Context, hosts map[string][]string)
 }
 
 func InsertAndVerifyCQLDataUsingDataInserter(ctx context.Context, di *utils.DataInserter) *utils.DataInserter {
-	framework.By("Inserting data")
-	err := di.Insert()
-	o.Expect(err).NotTo(o.HaveOccurred())
-
-	VerifyCQLData(ctx, di)
-
-	return di
+	return scylladbdatacenterverification.InsertAndVerifyCQLDataUsingDataInserter(ctx, di)
 }
 
 type VerifyCQLConnectionConfigsOptions struct {

test/e2e/utils/verification/scylladbcluster/cql.go

@@ -4,18 +4,14 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"sort"
+
 	"github.com/scylladb/scylla-operator/pkg/api/scylla/v1alpha1"
 	"github.com/scylladb/scylla-operator/pkg/helpers/slices"
 	"github.com/scylladb/scylla-operator/pkg/naming"
 	"github.com/scylladb/scylla-operator/test/e2e/framework"
-	v1alpha2 "github.com/scylladb/scylla-operator/test/e2e/utils/v1alpha1"
+	utilsv1alpha1 "github.com/scylladb/scylla-operator/test/e2e/utils/v1alpha1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/util/wait"
-	v2 "k8s.io/client-go/kubernetes/typed/core/v1"
-	"reflect"
-	"sort"
-	"strings"
-	"time"
 )
 
 func WaitForFullQuorum(ctx context.Context, rkcClusterMap map[string]framework.ClusterInterface, sc *v1alpha1.ScyllaDBCluster) error {
@@ -46,7 +42,7 @@ func WaitForFullQuorum(ctx context.Context, rkcClusterMap map[string]framework.C
 			continue
 		}
 
-		hosts, err := v1alpha2.GetBroadcastAddresses(ctx, clusterClient.KubeAdminClient().CoreV1(), sdc)
+		hosts, err := utilsv1alpha1.GetBroadcastAddresses(ctx, clusterClient.KubeAdminClient().CoreV1(), sdc)
 		if err != nil {
 			return fmt.Errorf("can't get broadcast addresses for ScyllaDBDatacenter %q: %w", naming.ObjRef(sdc), err)
 		}
@@ -83,7 +79,7 @@ func WaitForFullQuorum(ctx context.Context, rkcClusterMap map[string]framework.C
 			continue
 		}
 
-		err = waitForFullQuorum(ctx, clusterClient.KubeAdminClient().CoreV1(), sdc, sortedAllBroadcastAddresses)
+		err = utilsv1alpha1.WaitForFullQuorum(ctx, clusterClient.KubeAdminClient().CoreV1(), sdc, sortedAllBroadcastAddresses)
 		if err != nil {
 			errs = append(errs, err)
 		}
@@ -98,51 +94,3 @@ func WaitForFullQuorum(ctx context.Context, rkcClusterMap map[string]framework.C
 
 	return nil
 }
-
-// TODO: Should be unified with function coming from test/helpers once e2e's there starts using ScyllaDBDatacenter API.
-func waitForFullQuorum(ctx context.Context, client v2.CoreV1Interface, sdc *v1alpha1.ScyllaDBDatacenter, sortedExpectedHosts []string) error {
-	scyllaClient, hosts, err := v1alpha2.GetScyllaClient(ctx, client, sdc)
-	if err != nil {
-		return fmt.Errorf("can't get scylla client: %w", err)
-	}
-	defer scyllaClient.Close()
-
-	// Wait for node status to propagate and reach consistency.
-	// This can take a while so let's set a large enough timeout to avoid flakes.
-	return wait.PollImmediateWithContext(ctx, 1*time.Second, 5*time.Minute, func(ctx context.Context) (done bool, err error) {
-		allSeeAllAsUN := true
-		infoMessages := make([]string, 0, len(hosts))
-		var errs []error
-		for _, h := range hosts {
-			s, err := scyllaClient.Status(ctx, h)
-			if err != nil {
-				return true, fmt.Errorf("can't get scylla status on node %q: %w", h, err)
-			}
-
-			sHosts := s.Hosts()
-			sort.Strings(sHosts)
-			if !reflect.DeepEqual(sHosts, sortedExpectedHosts) {
-				errs = append(errs, fmt.Errorf("node %q thinks the cluster consists of different nodes: %s", h, sHosts))
-			}
-
-			downHosts := s.DownHosts()
-			infoMessages = append(infoMessages, fmt.Sprintf("Node %q, down: %q, up: %q", h, strings.Join(downHosts, "\n"), strings.Join(s.LiveHosts(), ",")))
-
-			if len(downHosts) != 0 {
-				allSeeAllAsUN = false
-			}
-		}
-
-		if !allSeeAllAsUN {
-			framework.Infof("ScyllaDB nodes have not reached status consistency yet. Statuses:\n%s", strings.Join(infoMessages, ","))
-		}
-
-		err = errors.Join(errs...)
-		if err != nil {
-			framework.Infof("ScyllaDB nodes encountered an error. Statuses:\n%s", strings.Join(infoMessages, ","))
-			return true, err
-		}
-
-		return allSeeAllAsUN, nil
-	})
-}