Properly use concurrent_transfers from MedusaConfig object

CHANGELOG/CHANGELOG-1.13.md

@@ -16,6 +16,7 @@ When cutting a new release, update the `unreleased` heading to the tag being gen
 ## unreleased
 
 * [BUGFIX] [#1235](https://github.com/k8ssandra/k8ssandra-operator/issues/1235) Prevent operator from modifying the spec when superUserRef is not set. Also, remove the injection to mount secrets to cassandra container.
+* [BUGFIX] [#1239](https://github.com/k8ssandra/k8ssandra-operator/issues/1239) Use `concurrent_transfers` setting from MedusaConfiguration object if it is actually set.
 
 ## v1.13.0 - 2024-02-24
 

apis/medusa/v1alpha1/medusa_types.go

@@ -74,8 +74,8 @@ type Storage struct {
 
 	// Number of concurrent uploads.
 	// Helps maximizing the speed of uploads but puts more pressure on the network.
-	// Defaults to 1.
-	// +kubebuilder:default=1
+	// Defaults to 0.
+	// +kubebuilder:default=0
 	// +optional
 	ConcurrentTransfers int `json:"concurrentTransfers,omitempty"`
 

charts/k8ssandra-operator/crds/k8ssandra-operator-crds.yaml

@@ -27449,10 +27449,10 @@ spec:
                         description: The name of the bucket to use for the backups.
                         type: string
                       concurrentTransfers:
-                        default: 1
+                        default: 0
                         description: Number of concurrent uploads. Helps maximizing
                           the speed of uploads but puts more pressure on the network.
-                          Defaults to 1.
+                          Defaults to 0.
                         type: integer
                       credentialsType:
                         description: Type of credentials to use for authentication.
@@ -32717,10 +32717,10 @@ spec:
                     description: The name of the bucket to use for the backups.
                     type: string
                   concurrentTransfers:
-                    default: 1
+                    default: 0
                     description: Number of concurrent uploads. Helps maximizing the
                       speed of uploads but puts more pressure on the network. Defaults
-                      to 1.
+                      to 0.
                     type: integer
                   credentialsType:
                     description: Type of credentials to use for authentication. Can

config/crd/bases/k8ssandra.io_k8ssandraclusters.yaml

@@ -27394,10 +27394,10 @@ spec:
                         description: The name of the bucket to use for the backups.
                         type: string
                       concurrentTransfers:
-                        default: 1
+                        default: 0
                         description: Number of concurrent uploads. Helps maximizing
                           the speed of uploads but puts more pressure on the network.
-                          Defaults to 1.
+                          Defaults to 0.
                         type: integer
                       credentialsType:
                         description: Type of credentials to use for authentication.

config/crd/bases/medusa.k8ssandra.io_medusaconfigurations.yaml

@@ -51,10 +51,10 @@ spec:
                     description: The name of the bucket to use for the backups.
                     type: string
                   concurrentTransfers:
-                    default: 1
+                    default: 0
                     description: Number of concurrent uploads. Helps maximizing the
                       speed of uploads but puts more pressure on the network. Defaults
-                      to 1.
+                      to 0.
                     type: integer
                   credentialsType:
                     description: Type of credentials to use for authentication. Can

controllers/k8ssandra/medusa_reconciler_test.go

@@ -26,14 +26,16 @@ import (
 )
 
 const (
-	medusaImageRepo        = "test"
-	storageSecret          = "storage-secret"
-	cassandraUserSecret    = "medusa-secret"
-	k8ssandraClusterName   = "test"
-	medusaConfigName       = "medusa-config"
-	medusaBucketSecretName = "medusa-bucket-secret"
-	prefixFromMedusaConfig = "prefix-from-medusa-config"
-	prefixFromClusterSpec  = "prefix-from-cluster-spec"
+	medusaImageRepo                     = "test"
+	storageSecret                       = "storage-secret"
+	cassandraUserSecret                 = "medusa-secret"
+	k8ssandraClusterName                = "test"
+	medusaConfigName                    = "medusa-config"
+	medusaBucketSecretName              = "medusa-bucket-secret"
+	prefixFromMedusaConfig              = "prefix-from-medusa-config"
+	prefixFromClusterSpec               = "prefix-from-cluster-spec"
+	defaultConcurrentTransfers          = 1
+	concurrentTransfersFromMedusaConfig = 2
 )
 
 func dcTemplate(dcName string, dataPlaneContext string) api.CassandraDatacenterTemplate {
@@ -62,7 +64,8 @@ func MedusaConfig(name, namespace string) *medusaapi.MedusaConfiguration {
 		},
 		Spec: medusaapi.MedusaConfigurationSpec{
 			StorageProperties: medusaapi.Storage{
-				Prefix: prefixFromMedusaConfig,
+				Prefix:              prefixFromMedusaConfig,
+				ConcurrentTransfers: concurrentTransfersFromMedusaConfig,
 			},
 		},
 	}
@@ -172,7 +175,7 @@ func createMultiDcClusterWithMedusa(t *testing.T, ctx context.Context, f *framew
 
 	t.Log("verify the config map exists and has the contents from the MedusaConfiguration object")
 	defaultPrefix := kc.Spec.Medusa.StorageProperties.Prefix
-	verifyConfigMap(require, ctx, f, namespace, defaultPrefix)
+	verifyConfigMap(require, ctx, f, namespace, defaultPrefix, defaultConcurrentTransfers)
 
 	t.Log("check that the standalone Medusa deployment was created in dc1")
 	medusaDeploymentKey1 := framework.ClusterKey{NamespacedName: types.NamespacedName{Namespace: namespace, Name: medusa.MedusaStandaloneDeploymentName(k8ssandraClusterName, "dc1")}, K8sContext: f.DataPlaneContexts[0]}
@@ -485,10 +488,10 @@ func createSingleDcClusterWithMedusaConfigRef(t *testing.T, ctx context.Context,
 	verifyReplicatedSecretReconciled(ctx, t, f, kc)
 
 	t.Log("verify the config map exists and has the contents from the MedusaConfiguration object")
-	verifyConfigMap(require, ctx, f, namespace, prefixFromClusterSpec)
+	verifyConfigMap(require, ctx, f, namespace, prefixFromClusterSpec, concurrentTransfersFromMedusaConfig)
 }
 
-func verifyConfigMap(r *require.Assertions, ctx context.Context, f *framework.Framework, namespace string, expectedPrefix string) {
+func verifyConfigMap(r *require.Assertions, ctx context.Context, f *framework.Framework, namespace string, expectedPrefix string, expectedConcurrentTransfers int) {
 	configMapName := fmt.Sprintf("%s-medusa", k8ssandraClusterName)
 	configMapKey := framework.ClusterKey{NamespacedName: types.NamespacedName{Namespace: namespace, Name: configMapName}, K8sContext: f.DataPlaneContexts[0]}
 	configMap := &corev1.ConfigMap{}
@@ -497,7 +500,9 @@ func verifyConfigMap(r *require.Assertions, ctx context.Context, f *framework.Fr
 			r.NoError(err, "failed to get Medusa ConfigMap")
 			return false
 		}
-		return strings.Contains(configMap.Data["medusa.ini"], fmt.Sprintf("prefix = %s", expectedPrefix))
+		prefixCorrect := strings.Contains(configMap.Data["medusa.ini"], fmt.Sprintf("prefix = %s", expectedPrefix))
+		concurrentTransfersCorrect := strings.Contains(configMap.Data["medusa.ini"], fmt.Sprintf("concurrent_transfers = %d", expectedConcurrentTransfers))
+		return prefixCorrect && concurrentTransfersCorrect
 	}, timeout, interval, "Medusa ConfigMap doesn't have the right content")
 }
 

pkg/medusa/reconcile.go

@@ -125,31 +125,35 @@ func CreateMedusaIni(kc *k8ss.K8ssandraCluster, dcConfig *cassandra.DatacenterCo
     use_mgmt_api = 1
     enabled = 1`
 
+	kcWithProperlyConcurrentMedusa := kc.DeepCopy()
+	if kc.Spec.Medusa.StorageProperties.ConcurrentTransfers == 0 {
+		kcWithProperlyConcurrentMedusa.Spec.Medusa.StorageProperties.ConcurrentTransfers = 1
+	}
 	t, err := template.New("ini").Parse(medusaIniTemplate)
 	if err != nil {
 		panic(err)
 	}
 	medusaIni := new(bytes.Buffer)
-	err = t.Execute(medusaIni, kc)
+	err = t.Execute(medusaIni, kcWithProperlyConcurrentMedusa)
 	if err != nil {
 		panic(err)
 	}
 
-	medusaConfiig := medusaIni.String()
+	medusaConfig := medusaIni.String()
 
 	// Create Kubernetes config here and append it
 	if dcConfig.ManagementApiAuth != nil && dcConfig.ManagementApiAuth.Manual != nil {
-		medusaConfiig += `
+		medusaConfig += `
     cassandra_url = https://127.0.0.1:8080/api/v0/ops/node/snapshots
     ca_cert = /etc/encryption/mgmt/ca.crt
     tls_cert = /etc/encryption/mgmt/tls.crt
     tls_key = /etc/encryption/mgmt/tls.key`
 	} else {
-		medusaConfiig += `
+		medusaConfig += `
     cassandra_url = http://127.0.0.1:8080/api/v0/ops/node/snapshots`
 	}
 
-	return medusaConfiig
+	return medusaConfig
 }
 
 func CreateMedusaConfigMap(namespace, k8cName, medusaIni string) *corev1.ConfigMap {

pkg/medusa/reconcile_test.go

@@ -18,6 +18,7 @@ import (
 func TestMedusaIni(t *testing.T) {
 	t.Run("Full", testMedusaIniFull)
 	t.Run("NoPrefix", testMedusaIniNoPrefix)
+	t.Run("ZeroConcurrentTransfers", testMedusaIniZeroConcurrentTransfers)
 	t.Run("Secured", testMedusaIniSecured)
 	t.Run("Unsecured", testMedusaIniUnsecured)
 	t.Run("MissingOptional", testMedusaIniMissingOptionalSettings)
@@ -158,6 +159,73 @@ func testMedusaIniNoPrefix(t *testing.T) {
 	assert.Contains(t, medusaIni, "cassandra_url = http://127.0.0.1:8080/api/v0/ops/node/snapshots")
 }
 
+func testMedusaIniZeroConcurrentTransfers(t *testing.T) {
+	kc := &api.K8ssandraCluster{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: "test",
+			Name:      "demo",
+		},
+		Spec: api.K8ssandraClusterSpec{
+			Cassandra: &api.CassandraClusterTemplate{
+				Datacenters: []api.CassandraDatacenterTemplate{
+					{
+						Meta: api.EmbeddedObjectMeta{
+							Name: "dc1",
+						},
+						K8sContext: "k8sCtx0",
+						Size:       3,
+						DatacenterOptions: api.DatacenterOptions{
+							ServerVersion: "3.11.14",
+						},
+					},
+				},
+			},
+			Medusa: &medusaapi.MedusaClusterTemplate{
+				StorageProperties: medusaapi.Storage{
+					StorageProvider: "s3",
+					StorageSecretRef: corev1.LocalObjectReference{
+						Name: "secret",
+					},
+					BucketName:               "bucket",
+					MaxBackupAge:             10,
+					MaxBackupCount:           20,
+					ApiProfile:               "default",
+					TransferMaxBandwidth:     "100MB/s",
+					ConcurrentTransfers:      0,
+					MultiPartUploadThreshold: 204857600,
+					Host:                     "192.168.0.1",
+					Region:                   "us-east-1",
+					Port:                     9001,
+					Secure:                   false,
+					BackupGracePeriodInDays:  7,
+				},
+				CassandraUserSecretRef: corev1.LocalObjectReference{
+					Name: "test-superuser",
+				},
+			},
+		},
+	}
+
+	dcConfig := cassandra.Coalesce(kc.CassClusterName(), kc.Spec.Cassandra.DeepCopy(), kc.Spec.Cassandra.Datacenters[0].DeepCopy())
+	medusaIni := CreateMedusaIni(kc, dcConfig)
+
+	assert.Contains(t, medusaIni, "storage_provider = s3")
+	assert.Contains(t, medusaIni, "bucket_name = bucket")
+	assert.Contains(t, medusaIni, "prefix = demo")
+	assert.Contains(t, medusaIni, "max_backup_age = 10")
+	assert.Contains(t, medusaIni, "max_backup_count = 20")
+	assert.Contains(t, medusaIni, "api_profile = default")
+	assert.Contains(t, medusaIni, "transfer_max_bandwidth = 100MB/s")
+	assert.Contains(t, medusaIni, "concurrent_transfers = 1")
+	assert.Contains(t, medusaIni, "multi_part_upload_threshold = 204857600")
+	assert.Contains(t, medusaIni, "host = 192.168.0.1")
+	assert.Contains(t, medusaIni, "region = us-east-1")
+	assert.Contains(t, medusaIni, "port = 9001")
+	assert.Contains(t, medusaIni, "secure = False")
+	assert.Contains(t, medusaIni, "backup_grace_period_in_days = 7")
+	assert.Contains(t, medusaIni, "cassandra_url = http://127.0.0.1:8080/api/v0/ops/node/snapshots")
+}
+
 func testMedusaIniSecured(t *testing.T) {
 	kc := &api.K8ssandraCluster{
 		ObjectMeta: metav1.ObjectMeta{
@@ -416,7 +484,7 @@ func testMedusaIniMissingOptionalSettings(t *testing.T) {
 	assert.Contains(t, medusaIni, "max_backup_count = 0")
 	assert.NotContains(t, medusaIni, "api_profile =")
 	assert.NotContains(t, medusaIni, "transfer_max_bandwidth =")
-	assert.NotContains(t, medusaIni, "concurrent_transfers =")
+	assert.Contains(t, medusaIni, "concurrent_transfers = 1")
 	assert.NotContains(t, medusaIni, "multi_part_upload_threshold =")
 	assert.NotContains(t, medusaIni, "host =")
 	assert.NotContains(t, medusaIni, "region =")

test/testdata/fixtures/multi-dc-encryption-medusa/k8ssandra.yaml

@@ -25,6 +25,7 @@ spec:
       name: global-medusa-config
     storageProperties:
       prefix: test
+      concurrentTransfers: 4
     certificatesSecretRef:
       name: client-certificates
   cassandra:

test/testdata/fixtures/multi-dc-encryption-medusa/medusa-config.yaml

@@ -23,3 +23,4 @@ spec:
     host: minio-service.minio.svc.cluster.local
     port: 9000
     secure: false
+    concurrentTransfers: 2

test/testdata/fixtures/single-dc-dse-search/k8ssandra.yaml

@@ -13,6 +13,7 @@ spec:
       host: minio-service.minio.svc.cluster.local
       port: 9000
       secure: false
+      concurrentTransfers: 8
   cassandra:
     serverVersion: 6.8.26
     serverType: dse