Support multi-datacenter clusters in test runner and script

hack/.ci/lib/e2e.sh

@@ -8,8 +8,23 @@ shopt -s inherit_errexit
 
 source "$( dirname "${BASH_SOURCE[0]}" )/../../lib/kube.sh"
 
+if [ -z "${KUBECONFIG_DIR+x}" ]; then
+  KUBECONFIGS=("${KUBECONFIG}")
+else
+  KUBECONFIGS=()
+  for f in $( find "$( realpath "${KUBECONFIG_DIR}" )" -maxdepth 1 -type f -name '*.kubeconfig' ); do
+    KUBECONFIGS+=("${f}")
+  done
+fi
+
 # gather-artifacts is a self sufficient function that collects artifacts without depending on any external objects.
+# $1- target directory
 function gather-artifacts {
+  if [ -z "${1+x}" ]; then
+    echo -e "Missing target directory.\nUsage: ${FUNCNAME[0]} target_directory" > /dev/stderr
+    exit 2
+  fi
+
   if [ -z "${SO_IMAGE+x}" ]; then
     echo "SO_IMAGE can't be empty" > /dev/stderr
     exit 2
@@ -57,8 +72,8 @@ EOF
 
   exit_code="$( wait-for-container-exit-with-logs gather-artifacts must-gather must-gather )"
 
-  kubectl -n=gather-artifacts cp --retries=42 -c=wait-for-artifacts must-gather:/tmp/artifacts "${ARTIFACTS}/must-gather"
-  ls -l "${ARTIFACTS}/must-gather"
+  kubectl -n=gather-artifacts cp --retries=42 -c=wait-for-artifacts must-gather:/tmp/artifacts "${1}"
+  ls -l "${1}"
 
   kubectl -n=gather-artifacts delete pod/must-gather --wait=false
 
@@ -68,8 +83,23 @@ EOF
   fi
 }
 
-function gather-artifact-on-exit {
-  gather-artifacts || "Error gathering artifacts" > /dev/stderr
+function gather-artifacts-on-exit {
+  for i in "${!KUBECONFIGS[@]}"; do
+    KUBECONFIG="${KUBECONFIGS[$i]}" gather-artifacts "${ARTIFACTS}/must-gather/${i}" &
+    gather_artifacts_bg_pids["${i}"]=$!
+  done
+
+  for pid in "${gather_artifacts_bg_pids[@]}"; do
+    wait "${pid}"
+  done
+}
+
+function unset-default-storageclass {
+  for f in "${KUBECONFIGS[@]}"; do
+    for r in $( KUBECONFIG="${f}" kubectl get storageclasses -o name ); do
+      KUBECONFIG="${f}" kubectl patch "${r}" -p '{"metadata": {"annotations":{"storageclass.kubernetes.io/is-default-class":"false"}}}'
+    done
+  done
 }
 
 function apply-e2e-workarounds {

hack/.ci/run-e2e-gke-release.sh

@@ -15,19 +15,25 @@ source "$( dirname "${BASH_SOURCE[0]}" )/../lib/kube.sh"
 source "$( dirname "${BASH_SOURCE[0]}" )/lib/e2e.sh"
 parent_dir="$( dirname "${BASH_SOURCE[0]}" )"
 
-trap gather-artifact-on-exit EXIT
-
+trap gather-artifacts-on-exit EXIT
 
 SO_NODECONFIG_PATH="${SO_NODECONFIG_PATH=${parent_dir}/manifests/cluster/nodeconfig.yaml}"
 export SO_NODECONFIG_PATH
 
 # Make sure there is no default storage class before we create our own so we always use our own provisioner from the release.
-for r in $( kubectl get storageclasses -o name ); do kubectl patch "${r}" -p '{"metadata": {"annotations":{"storageclass.kubernetes.io/is-default-class":"false"}}}'; done
+unset-default-storageclass
 
 SCYLLA_OPERATOR_FEATURE_GATES="${SCYLLA_OPERATOR_FEATURE_GATES:-AllAlpha=true,AllBeta=true}"
 export SCYLLA_OPERATOR_FEATURE_GATES
 
-timeout --foreground -v 10m "${parent_dir}/../ci-deploy-release.sh" "${SO_IMAGE}"
+for i in "${!KUBECONFIGS[@]}"; do
+  KUBECONFIG="${KUBECONFIGS[$i]}" timeout --foreground -v 10m "${parent_dir}/../ci-deploy-release.sh" "${SO_IMAGE}" &
+  ci_deploy_bg_pids["${i}"]=$!
+done
+
+for pid in "${ci_deploy_bg_pids[@]}"; do
+  wait "${pid}"
+done
 
-apply-e2e-workarounds
-run-e2e
+KUBECONFIG="${KUBECONFIGS[0]}" apply-e2e-workarounds
+KUBECONFIG="${KUBECONFIGS[0]}" run-e2e

hack/.ci/run-e2e-gke.sh

@@ -15,8 +15,7 @@ source "$( dirname "${BASH_SOURCE[0]}" )/../lib/kube.sh"
 source "$( dirname "${BASH_SOURCE[0]}" )/lib/e2e.sh"
 parent_dir="$( dirname "${BASH_SOURCE[0]}" )"
 
-trap gather-artifact-on-exit EXIT
-
+trap gather-artifacts-on-exit EXIT
 
 SO_NODECONFIG_PATH="${SO_NODECONFIG_PATH=./hack/.ci/manifests/cluster/nodeconfig.yaml}"
 export SO_NODECONFIG_PATH
@@ -29,13 +28,20 @@ if [[ "${SO_DISABLE_NODECONFIG:-false}" == "true" ]]; then
   SO_CSI_DRIVER_PATH=""
 else
   # Make sure there is no default storage class before we create our own.
-  for r in $( kubectl get storageclasses -o name ); do kubectl patch "${r}" -p '{"metadata": {"annotations":{"storageclass.kubernetes.io/is-default-class":"false"}}}'; done
+  unset-default-storageclass
 fi
 
 SCYLLA_OPERATOR_FEATURE_GATES="${SCYLLA_OPERATOR_FEATURE_GATES:-AllAlpha=true,AllBeta=true}"
 export SCYLLA_OPERATOR_FEATURE_GATES
 
-timeout --foreground -v 10m "${parent_dir}/../ci-deploy.sh" "${SO_IMAGE}"
+for i in "${!KUBECONFIGS[@]}"; do
+  KUBECONFIG="${KUBECONFIGS[$i]}" timeout --foreground -v 10m "${parent_dir}/../ci-deploy.sh" "${SO_IMAGE}" &
+  ci_deploy_bg_pids["${i}"]=$!
+done
+
+for pid in "${ci_deploy_bg_pids[@]}"; do
+  wait "${pid}"
+done
 
-apply-e2e-workarounds
-run-e2e
+KUBECONFIG="${KUBECONFIGS[0]}" apply-e2e-workarounds
+KUBECONFIG="${KUBECONFIGS[0]}" run-e2e

pkg/cmd/tests/options.go

@@ -13,6 +13,7 @@ import (
 	"github.com/scylladb/scylla-operator/test/e2e/framework"
 	"github.com/spf13/cobra"
 	apierrors "k8s.io/apimachinery/pkg/util/errors"
+	"k8s.io/client-go/rest"
 )
 
 type IngressControllerOptions struct {
@@ -38,7 +39,7 @@ var supportedBroadcastAddressTypes = []scyllav1.BroadcastAddressType{
 }
 
 type TestFrameworkOptions struct {
-	genericclioptions.ClientConfig
+	genericclioptions.ClientConfigSet
 
 	ArtifactsDir                 string
 	DeleteTestingNSPolicyUntyped string
@@ -54,7 +55,7 @@ type TestFrameworkOptions struct {
 
 func NewTestFrameworkOptions(streams genericclioptions.IOStreams, userAgent string) *TestFrameworkOptions {
 	return &TestFrameworkOptions{
-		ClientConfig:                 genericclioptions.NewClientConfig(userAgent),
+		ClientConfigSet:              genericclioptions.NewClientConfigSet(userAgent),
 		ArtifactsDir:                 "",
 		DeleteTestingNSPolicyUntyped: string(framework.DeleteTestingNSPolicyAlways),
 		IngressController:            &IngressControllerOptions{},
@@ -71,7 +72,7 @@ func NewTestFrameworkOptions(streams genericclioptions.IOStreams, userAgent stri
 }
 
 func (o *TestFrameworkOptions) AddFlags(cmd *cobra.Command) {
-	o.ClientConfig.AddFlags(cmd)
+	o.ClientConfigSet.AddFlags(cmd)
 
 	cmd.PersistentFlags().StringVarP(&o.ArtifactsDir, "artifacts-dir", "", o.ArtifactsDir, "A directory for storing test artifacts. No data is collected until set.")
 	cmd.PersistentFlags().StringVarP(&o.DeleteTestingNSPolicyUntyped, "delete-namespace-policy", "", o.DeleteTestingNSPolicyUntyped, fmt.Sprintf("Namespace deletion policy. Allowed values are [%s].", strings.Join(
@@ -104,7 +105,7 @@ func (o *TestFrameworkOptions) AddFlags(cmd *cobra.Command) {
 func (o *TestFrameworkOptions) Validate(args []string) error {
 	var errors []error
 
-	err := o.ClientConfig.Validate()
+	err := o.ClientConfigSet.Validate()
 	if err != nil {
 		errors = append(errors, err)
 	}
@@ -141,7 +142,7 @@ func (o *TestFrameworkOptions) Validate(args []string) error {
 }
 
 func (o *TestFrameworkOptions) Complete(args []string) error {
-	err := o.ClientConfig.Complete()
+	err := o.ClientConfigSet.Complete()
 	if err != nil {
 		return err
 	}
@@ -172,7 +173,9 @@ func (o *TestFrameworkOptions) Complete(args []string) error {
 	}
 
 	framework.TestContext = &framework.TestContextType{
-		RestConfig:            o.RestConfig,
+		RestConfigs: slices.ConvertSlice(o.ClientConfigs, func(cc genericclioptions.ClientConfig) *rest.Config {
+			return cc.RestConfig
+		}),
 		ArtifactsDir:          o.ArtifactsDir,
 		DeleteTestingNSPolicy: o.DeleteTestingNSPolicy,
 		ScyllaClusterOptions:  o.scyllaClusterOptions,

pkg/genericclioptions/genericclioptions.go

@@ -9,10 +9,15 @@ import (
 
 	"github.com/scylladb/scylla-operator/pkg/version"
 	"github.com/spf13/cobra"
+	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 	restclient "k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/clientcmd"
 )
 
+const (
+	defaultKubeconfig = ""
+)
+
 // IOStreams is a structure containing all standard streams.
 type IOStreams struct {
 	// In think, os.Stdin
@@ -23,13 +28,113 @@ type IOStreams struct {
 	ErrOut io.Writer
 }
 
-type ClientConfig struct {
-	Kubeconfig    string
+type ClientConfigBase struct {
 	QPS           float32
 	Burst         int
 	UserAgentName string
-	RestConfig    *restclient.Config
-	ProtoConfig   *restclient.Config
+}
+
+func NewDefaultClientConfigBase(userAgentName string) ClientConfigBase {
+	return ClientConfigBase{
+		QPS:           50,
+		Burst:         75,
+		UserAgentName: userAgentName,
+	}
+}
+
+func (ccb *ClientConfigBase) AddFlags(cmd *cobra.Command) {
+	cmd.PersistentFlags().Float32VarP(&ccb.QPS, "qps", "", ccb.QPS, "Maximum allowed number of queries per second.")
+	cmd.PersistentFlags().IntVarP(&ccb.Burst, "burst", "", ccb.Burst, "Allows extra queries to accumulate when a client is exceeding its rate.")
+}
+
+func (ccb *ClientConfigBase) Validate() error {
+	return nil
+}
+
+func (ccb *ClientConfigBase) Complete() error {
+	return nil
+}
+
+type ClientConfigSet struct {
+	ClientConfigBase
+	kubeconfigs   []string
+	ClientConfigs []ClientConfig
+}
+
+func NewClientConfigSet(userAgentName string) ClientConfigSet {
+	return ClientConfigSet{
+		ClientConfigBase: NewDefaultClientConfigBase(userAgentName),
+		kubeconfigs:      []string{defaultKubeconfig},
+	}
+}
+
+func (ccs *ClientConfigSet) AddFlags(cmd *cobra.Command) {
+	ccs.ClientConfigBase.AddFlags(cmd)
+
+	cmd.PersistentFlags().StringArrayVarP(&ccs.kubeconfigs, "kubeconfig", "", ccs.kubeconfigs, "Path to kubeconfig file(s).")
+}
+
+func (ccs *ClientConfigSet) Validate() error {
+	var errs []error
+	var err error
+
+	err = ccs.ClientConfigBase.Validate()
+	if err != nil {
+		errs = append(errs, fmt.Errorf("invalid client config base: %w", err))
+	}
+
+	if len(ccs.kubeconfigs) == 0 {
+		errs = append(errs, fmt.Errorf("at least one kubeconfig must be provided"))
+	}
+
+	for _, kubeconfig := range ccs.kubeconfigs {
+		cc := NewClientConfig(ccs.UserAgentName)
+		cc.Kubeconfig = kubeconfig
+		cc.QPS = ccs.QPS
+		cc.Burst = ccs.Burst
+
+		err = cc.Validate()
+		if err != nil {
+			errs = append(errs, fmt.Errorf("invalid client config for kubeconfig %q: %w", kubeconfig, err))
+		}
+	}
+
+	return utilerrors.NewAggregate(errs)
+}
+
+func (ccs *ClientConfigSet) Complete() error {
+	var err error
+
+	err = ccs.ClientConfigBase.Complete()
+	if err != nil {
+		return fmt.Errorf("can't complete client config base: %w", err)
+	}
+
+	clientConfigs := make([]ClientConfig, 0, len(ccs.kubeconfigs))
+	for _, kubeconfig := range ccs.kubeconfigs {
+		cc := NewClientConfig(ccs.UserAgentName)
+		cc.Kubeconfig = kubeconfig
+		cc.QPS = ccs.QPS
+		cc.Burst = ccs.Burst
+
+		err = cc.Complete()
+		if err != nil {
+			return fmt.Errorf("can't complete client config for kubeconfig %q: %w", kubeconfig, err)
+		}
+
+		clientConfigs = append(clientConfigs, cc)
+	}
+
+	ccs.ClientConfigs = clientConfigs
+
+	return nil
+}
+
+type ClientConfig struct {
+	ClientConfigBase
+	Kubeconfig  string
+	RestConfig  *restclient.Config
+	ProtoConfig *restclient.Config
 }
 
 func MakeVersionedUserAgent(baseName string) string {
@@ -45,28 +150,39 @@ func MakeVersionedUserAgent(baseName string) string {
 
 func NewClientConfig(userAgentName string) ClientConfig {
 	return ClientConfig{
-		Kubeconfig:    "",
-		QPS:           50,
-		Burst:         75,
-		UserAgentName: userAgentName,
-		RestConfig:    nil,
-		ProtoConfig:   nil,
+		ClientConfigBase: NewDefaultClientConfigBase(userAgentName),
+		Kubeconfig:       defaultKubeconfig,
+		RestConfig:       nil,
+		ProtoConfig:      nil,
 	}
 }
 
 func (cc *ClientConfig) AddFlags(cmd *cobra.Command) {
+	cc.ClientConfigBase.AddFlags(cmd)
+
 	cmd.PersistentFlags().StringVarP(&cc.Kubeconfig, "kubeconfig", "", cc.Kubeconfig, "Path to the kubeconfig file.")
-	cmd.PersistentFlags().Float32VarP(&cc.QPS, "qps", "", cc.QPS, "Maximum allowed number of queries per second.")
-	cmd.PersistentFlags().IntVarP(&cc.Burst, "burst", "", cc.Burst, "Allows extra queries to accumulate when a client is exceeding its rate.")
 }
 
 func (cc *ClientConfig) Validate() error {
-	return nil
+	var errs []error
+	var err error
+
+	err = cc.ClientConfigBase.Validate()
+	if err != nil {
+		errs = append(errs, fmt.Errorf("invalid client config base: %w", err))
+	}
+
+	return utilerrors.NewAggregate(errs)
 }
 
 func (cc *ClientConfig) Complete() error {
 	var err error
 
+	err = cc.ClientConfigBase.Complete()
+	if err != nil {
+		return fmt.Errorf("can't complete client config base: %w", err)
+	}
+
 	loader := clientcmd.NewDefaultClientConfigLoadingRules()
 	// Use explicit kubeconfig if set.
 	loader.ExplicitPath = cc.Kubeconfig