Add WALG_UPLOAD_DISK_CONCURRENCY env (#504)

* Add WALG_UPLOAD_DISK_CONCURRENCY env

* Add additional params

* Try configuring the CLONE as well

* Dynamic WALG_UPLOAD_DISK_CONCURRENCY depending on CPU limits

* check for errors

controllers/postgres_controller.go

@@ -30,6 +30,7 @@ import (
 	coreosv1 "github.com/prometheus-operator/prometheus-operator/pkg/apis/monitoring/v1"
 	networkingv1 "k8s.io/api/networking/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -441,6 +442,20 @@ func (r *PostgresReconciler) updatePodEnvironmentConfigMap(ctx context.Context,
 	// region
 	region := backupConfig.S3Region
 
+	// set the WALG_UPLOAD_DISK_CONCURRENCY based on the configured cpu limits
+	q, err := resource.ParseQuantity(p.Spec.Size.CPU)
+	if err != nil {
+		return fmt.Errorf("error while parsing the postgres cpu size: %w", err)
+	}
+	uploadDiskConcurrency := "1"
+	if q.Value() > 32 {
+		uploadDiskConcurrency = "32"
+	} else if q.Value() > 1 {
+		uploadDiskConcurrency = fmt.Sprint(q.Value())
+	}
+	uploadConcurrency := "32"
+	downloadConcurrency := "32"
+
 	// use the rest as provided in the secret
 	bucketName := backupConfig.S3BucketName
 	backupSchedule := backupConfig.Schedule
@@ -452,19 +467,25 @@ func (r *PostgresReconciler) updatePodEnvironmentConfigMap(ctx context.Context,
 
 	// create updated content for pod environment configmap
 	data := map[string]string{
-		"USE_WALG_BACKUP":                  "true",
-		"USE_WALG_RESTORE":                 "true",
-		"WALE_S3_PREFIX":                   "s3://" + bucketName + "/$(SCOPE)",
-		"WALG_S3_PREFIX":                   "s3://" + bucketName + "/$(SCOPE)",
-		"CLONE_WALG_S3_PREFIX":             "s3://" + bucketName + "/$(CLONE_SCOPE)",
-		"WALE_BACKUP_THRESHOLD_PERCENTAGE": "100",
-		"AWS_ENDPOINT":                     awsEndpoint,
-		"WALE_S3_ENDPOINT":                 walES3Endpoint, // same as above, but slightly modified
-		"AWS_S3_FORCE_PATH_STYLE":          "true",
-		"AWS_REGION":                       region,         // now we can use AWS S3
-		"WALG_DISABLE_S3_SSE":              walgDisableSSE, // server side encryption
-		"BACKUP_SCHEDULE":                  backupSchedule,
-		"BACKUP_NUM_TO_RETAIN":             backupNumToRetain,
+		"USE_WALG_BACKUP":                    "true",
+		"USE_WALG_RESTORE":                   "true",
+		"WALE_S3_PREFIX":                     "s3://" + bucketName + "/$(SCOPE)",
+		"WALG_S3_PREFIX":                     "s3://" + bucketName + "/$(SCOPE)",
+		"CLONE_WALG_S3_PREFIX":               "s3://" + bucketName + "/$(CLONE_SCOPE)",
+		"WALE_BACKUP_THRESHOLD_PERCENTAGE":   "100",
+		"AWS_ENDPOINT":                       awsEndpoint,
+		"WALE_S3_ENDPOINT":                   walES3Endpoint, // same as above, but slightly modified
+		"AWS_S3_FORCE_PATH_STYLE":            "true",
+		"AWS_REGION":                         region,         // now we can use AWS S3
+		"WALG_DISABLE_S3_SSE":                walgDisableSSE, // server side encryption
+		"BACKUP_SCHEDULE":                    backupSchedule,
+		"BACKUP_NUM_TO_RETAIN":               backupNumToRetain,
+		"WALG_UPLOAD_DISK_CONCURRENCY":       uploadDiskConcurrency,
+		"CLONE_WALG_UPLOAD_DISK_CONCURRENCY": uploadDiskConcurrency,
+		"WALG_UPLOAD_CONCURRENCY":            uploadConcurrency,
+		"CLONE_WALG_UPLOAD_CONCURRENCY":      uploadConcurrency,
+		"WALG_DOWNLOAD_CONCURRENCY":          downloadConcurrency,
+		"CLONE_WALG_DOWNLOAD_CONCURRENCY":    downloadConcurrency,
 	}
 
 	cm := &corev1.ConfigMap{}