[PLAT-16356] Add YBC Throttle Params to the universe CRD

Summary:
Added YBC Throttle params to the universe CRD so the operator can pick
them up.
In addition to updating the throttle params, it will also validate the values
are good before attempting to update YBC.

Test Plan:
created a universe via the operator
updated some YBC params and validated they got updated
Deleted the YBC Params and validated it got set back to defaults
Set out of bounds value for ybc params and validate error
Set good value for ybc params and validated status was re-updated

Reviewers: anijhawan, vkumar

Reviewed By: vkumar

Subscribers: yugaware

Differential Revision: https://phorge.dev.yugabyte.com/D41774

managed/src/main/java/com/yugabyte/yw/common/operator/YBReconcilerFactory.java

@@ -3,6 +3,7 @@
 import com.google.inject.Inject;
 import com.yugabyte.yw.commissioner.TaskExecutor;
 import com.yugabyte.yw.common.CustomerTaskManager;
+import com.yugabyte.yw.common.backuprestore.ybc.YbcManager;
 import com.yugabyte.yw.common.config.GlobalConfKeys;
 import com.yugabyte.yw.common.config.RuntimeConfGetter;
 import com.yugabyte.yw.common.operator.utils.OperatorUtils;
@@ -23,6 +24,7 @@ public class YBReconcilerFactory {
   @Inject private CustomerTaskManager customerTaskManager;
   @Inject private OperatorUtils operatorUtils;
   @Inject private UniverseActionsHandler universeActionsHandler;
+  @Inject private YbcManager ybcManager;
 
   public YBUniverseReconciler getYBUniverseReconciler(KubernetesClient client) {
     String namespace = confGetter.getGlobalConf(GlobalConfKeys.KubernetesOperatorNamespace);
@@ -38,6 +40,7 @@ public YBUniverseReconciler getYBUniverseReconciler(KubernetesClient client) {
         confGetter,
         customerTaskManager,
         operatorUtils,
-        universeActionsHandler);
+        universeActionsHandler,
+        ybcManager);
   }
 }

managed/src/main/java/com/yugabyte/yw/common/operator/YBUniverseReconciler.java

@@ -13,7 +13,9 @@
 import com.yugabyte.yw.commissioner.tasks.UniverseTaskBase;
 import com.yugabyte.yw.commissioner.tasks.UniverseTaskBase.AllowedTasks;
 import com.yugabyte.yw.common.CustomerTaskManager;
+import com.yugabyte.yw.common.backuprestore.ybc.YbcManager;
 import com.yugabyte.yw.common.config.RuntimeConfGetter;
+import com.yugabyte.yw.common.gflags.GFlagsUtil;
 import com.yugabyte.yw.common.gflags.SpecificGFlags;
 import com.yugabyte.yw.common.operator.OperatorStatusUpdater.UniverseState;
 import com.yugabyte.yw.common.operator.utils.KubernetesEnvironmentVariables;
@@ -37,6 +39,7 @@
 import com.yugabyte.yw.forms.UniverseDefinitionTaskParams.UserIntent;
 import com.yugabyte.yw.forms.UniverseDefinitionTaskParams.UserIntent.K8SNodeResourceSpec;
 import com.yugabyte.yw.forms.UniverseResp;
+import com.yugabyte.yw.forms.YbcThrottleParametersResponse.ThrottleParamValue;
 import com.yugabyte.yw.models.Customer;
 import com.yugabyte.yw.models.CustomerTask;
 import com.yugabyte.yw.models.CustomerTask.TargetType;
@@ -55,9 +58,11 @@
 import io.fabric8.kubernetes.client.informers.cache.Lister;
 import io.yugabyte.operator.v1alpha1.Release;
 import io.yugabyte.operator.v1alpha1.YBUniverse;
+import io.yugabyte.operator.v1alpha1.ybuniversespec.YbcThrottleParameters;
 import io.yugabyte.operator.v1alpha1.ybuniversespec.YcqlPassword;
 import io.yugabyte.operator.v1alpha1.ybuniversespec.YsqlPassword;
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Base64;
 import java.util.Collections;
 import java.util.HashMap;
@@ -101,6 +106,7 @@ public class YBUniverseReconciler extends AbstractReconciler<YBUniverse> {
   private final RuntimeConfGetter confGetter;
   private final CustomerTaskManager customerTaskManager;
   private final UniverseActionsHandler universeActionsHandler;
+  private final YbcManager ybcManager;
   private final Set<UUID> universeReadySet;
   private final Map<String, String> universeDeletionReferenceMap;
   private final Map<String, UUID> universeTaskMap;
@@ -122,7 +128,8 @@ public YBUniverseReconciler(
       RuntimeConfGetter confGetter,
       CustomerTaskManager customerTaskManager,
       OperatorUtils operatorUtils,
-      UniverseActionsHandler universeActionsHandler) {
+      UniverseActionsHandler universeActionsHandler,
+      YbcManager ybcManager) {
     this(
         client,
         informerFactory,
@@ -136,7 +143,8 @@ public YBUniverseReconciler(
         confGetter,
         customerTaskManager,
         operatorUtils,
-        universeActionsHandler);
+        universeActionsHandler,
+        ybcManager);
   }
 
   @VisibleForTesting
@@ -153,7 +161,8 @@ protected YBUniverseReconciler(
       RuntimeConfGetter confGetter,
       CustomerTaskManager customerTaskManager,
       OperatorUtils operatorUtils,
-      UniverseActionsHandler universeActionsHandler) {
+      UniverseActionsHandler universeActionsHandler,
+      YbcManager ybcManager) {
 
     super(client, informerFactory);
     this.ybUniverseClient = client.resources(YBUniverse.class);
@@ -175,6 +184,7 @@ protected YBUniverseReconciler(
     this.universeTaskMap = new HashMap<>();
     this.operatorUtils = operatorUtils;
     this.universeActionsHandler = universeActionsHandler;
+    this.ybcManager = ybcManager;
   }
 
   private static String getWorkQueueKey(YBUniverse ybUniverse) {
@@ -802,6 +812,18 @@ protected void editUniverse(
                 getWorkQueueKey(ybUniverse), OperatorWorkQueue.ResourceAction.NO_OP, false);
             return;
           }
+          // Handle updating throttle params.
+        } else if (operatorUtils.isThrottleParamUpdate(universe, ybUniverse)) {
+          if (checkAndHandleUniverseLock(
+              ybUniverse, universe, OperatorWorkQueue.ResourceAction.NO_OP)) {
+            return;
+          }
+          // TODO: We should probably update the universe status better here, but the
+          // KubernetesOperatorStatusUpdater currently doesn't have a good way to do this - all
+          // action updates are task based right now
+          updateThrottleParams(universe, ybUniverse);
+          kubernetesStatusUpdater.updateUniverseState(
+              KubernetesResourceDetails.fromResource(ybUniverse), UniverseState.READY);
           // Case with new edits
         } else if (!StringUtils.equals(
             incomingIntent.universeOverrides, currentUserIntent.universeOverrides)) {
@@ -1102,6 +1124,174 @@ private UserIntent createUserIntent(YBUniverse ybUniverse, UUID customerUUID, bo
     }
   }
 
+  private void updateThrottleParams(Universe universe, YBUniverse ybUniverse) {
+    YbcThrottleParameters specParams = ybUniverse.getSpec().getYbcThrottleParameters();
+    Map<String, ThrottleParamValue> currentParamsMap =
+        ybcManager.getThrottleParams(universe.getUniverseUUID()).getThrottleParamsMap();
+    if (specParams == null) {
+      // create default spec params
+      specParams = new YbcThrottleParameters();
+    }
+    if (specParams.getMaxConcurrentDownloads() == null)
+      specParams.setMaxConcurrentDownloads(
+          (long)
+              currentParamsMap
+                  .get(GFlagsUtil.YBC_MAX_CONCURRENT_DOWNLOADS)
+                  .getPresetValues()
+                  .getDefaultValue());
+    if (specParams.getMaxConcurrentUploads() == null)
+      specParams.setMaxConcurrentUploads(
+          (long)
+              currentParamsMap
+                  .get(GFlagsUtil.YBC_MAX_CONCURRENT_UPLOADS)
+                  .getPresetValues()
+                  .getDefaultValue());
+    if (specParams.getPerDownloadNumObjects() == null)
+      specParams.setPerDownloadNumObjects(
+          (long)
+              currentParamsMap
+                  .get(GFlagsUtil.YBC_PER_DOWNLOAD_OBJECTS)
+                  .getPresetValues()
+                  .getDefaultValue());
+    if (specParams.getPerUploadNumObjects() == null)
+      specParams.setPerUploadNumObjects(
+          (long)
+              currentParamsMap
+                  .get(GFlagsUtil.YBC_PER_UPLOAD_OBJECTS)
+                  .getPresetValues()
+                  .getDefaultValue());
+    validateThrottleParams(specParams, currentParamsMap);
+    com.yugabyte.yw.forms.YbcThrottleParameters newParams =
+        new com.yugabyte.yw.forms.YbcThrottleParameters();
+    // We are casting a Long to an int, but this is only because the java code generated from the
+    // CRD uses Longs.
+    newParams.maxConcurrentDownloads = specParams.getMaxConcurrentDownloads().intValue();
+    newParams.maxConcurrentUploads = specParams.getMaxConcurrentUploads().intValue();
+    newParams.perDownloadNumObjects = specParams.getPerDownloadNumObjects().intValue();
+    newParams.perUploadNumObjects = specParams.getPerUploadNumObjects().intValue();
+    ybcManager.setThrottleParams(universe.getUniverseUUID(), newParams);
+  }
+
+  /**
+   * Validate the throttle parameters.
+   *
+   * <p>This method will validate the throttle parameters with the preset values from YBC. If any of
+   * the throttle parameters are out of the preset range, an error message will be added to the
+   * errors list.
+   *
+   * @param specParams the throttle parameters to validate
+   * @param currentParamsMap the map of current throttle parameters and their preset values
+   * @return a list of error messages
+   */
+  private void validateThrottleParams(
+      YbcThrottleParameters specParams, Map<String, ThrottleParamValue> currentParamsMap) {
+    List<String> errors = new ArrayList<>();
+    if (specParams.getMaxConcurrentDownloads() != null
+        && specParams.getMaxConcurrentDownloads()
+            > currentParamsMap
+                .get(GFlagsUtil.YBC_MAX_CONCURRENT_DOWNLOADS)
+                .getPresetValues()
+                .getMaxValue()) {
+      errors.add(
+          "Max concurrent downloads cannot be greater than "
+              + currentParamsMap
+                  .get(GFlagsUtil.YBC_MAX_CONCURRENT_DOWNLOADS)
+                  .getPresetValues()
+                  .getMaxValue());
+    } else if (specParams.getMaxConcurrentDownloads() != null
+        && specParams.getMaxConcurrentDownloads()
+            < currentParamsMap
+                .get(GFlagsUtil.YBC_MAX_CONCURRENT_DOWNLOADS)
+                .getPresetValues()
+                .getMinValue()) {
+      errors.add(
+          "Max concurrent downloads cannot be less than "
+              + currentParamsMap
+                  .get(GFlagsUtil.YBC_MAX_CONCURRENT_DOWNLOADS)
+                  .getPresetValues()
+                  .getMinValue());
+    }
+    if (specParams.getMaxConcurrentUploads() != null
+        && specParams.getMaxConcurrentUploads()
+            > currentParamsMap
+                .get(GFlagsUtil.YBC_MAX_CONCURRENT_UPLOADS)
+                .getPresetValues()
+                .getMaxValue()) {
+      errors.add(
+          "Max concurrent uploads cannot be greater than "
+              + currentParamsMap
+                  .get(GFlagsUtil.YBC_MAX_CONCURRENT_UPLOADS)
+                  .getPresetValues()
+                  .getMaxValue());
+    } else if (specParams.getMaxConcurrentUploads() != null
+        && specParams.getMaxConcurrentUploads()
+            < currentParamsMap
+                .get(GFlagsUtil.YBC_MAX_CONCURRENT_UPLOADS)
+                .getPresetValues()
+                .getMinValue()) {
+      errors.add(
+          "Max concurrent uploads cannot be less than "
+              + currentParamsMap
+                  .get(GFlagsUtil.YBC_MAX_CONCURRENT_UPLOADS)
+                  .getPresetValues()
+                  .getMinValue());
+    }
+    if (specParams.getPerDownloadNumObjects() != null
+        && specParams.getPerDownloadNumObjects()
+            > currentParamsMap
+                .get(GFlagsUtil.YBC_PER_DOWNLOAD_OBJECTS)
+                .getPresetValues()
+                .getMaxValue()) {
+      errors.add(
+          "Per download objects cannot be greater than "
+              + currentParamsMap
+                  .get(GFlagsUtil.YBC_PER_DOWNLOAD_OBJECTS)
+                  .getPresetValues()
+                  .getMaxValue());
+    } else if (specParams.getPerDownloadNumObjects() != null
+        && specParams.getPerDownloadNumObjects()
+            < currentParamsMap
+                .get(GFlagsUtil.YBC_PER_DOWNLOAD_OBJECTS)
+                .getPresetValues()
+                .getMinValue()) {
+      errors.add(
+          "Per download objects cannot be less than "
+              + currentParamsMap
+                  .get(GFlagsUtil.YBC_PER_DOWNLOAD_OBJECTS)
+                  .getPresetValues()
+                  .getMinValue());
+    }
+    if (specParams.getPerUploadNumObjects() != null
+        && specParams.getPerUploadNumObjects()
+            > currentParamsMap
+                .get(GFlagsUtil.YBC_PER_UPLOAD_OBJECTS)
+                .getPresetValues()
+                .getMaxValue()) {
+      errors.add(
+          "Per upload objects cannot be greater than "
+              + currentParamsMap
+                  .get(GFlagsUtil.YBC_PER_UPLOAD_OBJECTS)
+                  .getPresetValues()
+                  .getMaxValue());
+    } else if (specParams.getPerUploadNumObjects() != null
+        && specParams.getPerUploadNumObjects()
+            < currentParamsMap
+                .get(GFlagsUtil.YBC_PER_UPLOAD_OBJECTS)
+                .getPresetValues()
+                .getMinValue()) {
+      errors.add(
+          "Per upload objects cannot be less than "
+              + currentParamsMap
+                  .get(GFlagsUtil.YBC_PER_UPLOAD_OBJECTS)
+                  .getPresetValues()
+                  .getMinValue());
+    }
+    if (!errors.isEmpty()) {
+      log.error("found errors: {}", errors);
+      throw new IllegalArgumentException(String.join("\n", errors));
+    }
+  }
+
   // getSecret find a secret in the namespace an operator is listening on.
   private Secret getSecret(String name) {
     if (StringUtils.isNotBlank(namespace)) {

managed/src/main/java/com/yugabyte/yw/common/operator/resources/universeCrd.yaml

@@ -228,6 +228,24 @@ spec:
                   is resumed.
                 type: boolean
                 default: false
+              ybcThrottleParameters:
+                description: >
+                  YBC throttle parameters for the universe. These throttle parameters can be used to
+                  control speed and resource usage of taking and restoring backups.
+                type: object
+                properties:
+                  maxConcurrentUploads:
+                    description: Number of parallel uploads per node. Used during backup.
+                    type: integer
+                  perUploadNumObjects:
+                    description: Number of buffers per upload per node. Used during backup.
+                    type: integer
+                  maxConcurrentDownloads:
+                    description: Number of parallel downloads per node. Used during restore.
+                    type: integer
+                  perDownloadNumObjects:
+                    description: Number of buffers per download per node. Used during restore.
+                    type: integer
               kubernetesOverrides:
                 description: >
                   Kubernetes overrides for the universe.