separate replication orchestrator resource requests (#10227)

Author: jrhizor

airbyte-config/models/src/main/java/io/airbyte/config/Configs.java

@@ -353,6 +353,26 @@ public interface Configs {
    */
   String getContainerOrchestratorImage();
 
+  /**
+   * Define the replication orchestrator's minimum CPU usage. Defaults to none.
+   */
+  String getReplicationOrchestratorCpuRequest();
+
+  /**
+   * Define the replication orchestrator's maximum CPU usage. Defaults to none.
+   */
+  String getReplicationOrchestratorCpuLimit();
+
+  /**
+   * Define the replication orchestrator's minimum RAM usage. Defaults to none.
+   */
+  String getReplicationOrchestratorMemoryRequest();
+
+  /**
+   * Define the replication orchestrator's maximum RAM usage. Defaults to none.
+   */
+  String getReplicationOrchestratorMemoryLimit();
+
   /**
    * Get the longest duration of non long running activity
    */

airbyte-config/models/src/main/java/io/airbyte/config/EnvConfigs.java

@@ -129,6 +129,10 @@ public class EnvConfigs implements Configs {
   public static final long DEFAULT_MAX_SYNC_WORKERS = 5;
 
   public static final String DEFAULT_NETWORK = "host";
+  private static final String REPLICATION_ORCHESTRATOR_CPU_REQUEST = "REPLICATION_ORCHESTRATOR_CPU_REQUEST";
+  private static final String REPLICATION_ORCHESTRATOR_CPU_LIMIT = "REPLICATION_ORCHESTRATOR_CPU_LIMIT";
+  private static final String REPLICATION_ORCHESTRATOR_MEMORY_REQUEST = "REPLICATION_ORCHESTRATOR_MEMORY_REQUEST";
+  private static final String REPLICATION_ORCHESTRATOR_MEMORY_LIMIT = "REPLICATION_ORCHESTRATOR_MEMORY_LIMIT";
 
   private final Function<String, String> getEnv;
   private final Supplier<Set<String>> getAllEnvKeys;
@@ -594,6 +598,26 @@ public String getContainerOrchestratorImage() {
     return getEnvOrDefault(CONTAINER_ORCHESTRATOR_IMAGE, "airbyte/container-orchestrator:" + getAirbyteVersion().serialize());
   }
 
+  @Override
+  public String getReplicationOrchestratorCpuRequest() {
+    return getEnvOrDefault(REPLICATION_ORCHESTRATOR_CPU_REQUEST, null);
+  }
+
+  @Override
+  public String getReplicationOrchestratorCpuLimit() {
+    return getEnvOrDefault(REPLICATION_ORCHESTRATOR_CPU_LIMIT, null);
+  }
+
+  @Override
+  public String getReplicationOrchestratorMemoryRequest() {
+    return getEnvOrDefault(REPLICATION_ORCHESTRATOR_MEMORY_REQUEST, null);
+  }
+
+  @Override
+  public String getReplicationOrchestratorMemoryLimit() {
+    return getEnvOrDefault(REPLICATION_ORCHESTRATOR_MEMORY_LIMIT, null);
+  }
+
   @Override
   public int getMaxActivityTimeoutSecond() {
     return Integer.parseInt(getEnvOrDefault(ACTIVITY_MAX_TIMEOUT_SECOND, "120"));

airbyte-workers/src/main/java/io/airbyte/workers/WorkerConfigs.java

@@ -13,7 +13,13 @@
 public class WorkerConfigs {
 
   private final Configs.WorkerEnvironment workerEnvironment;
+
+  // for running source, destination, normalization, dbt, normalization orchestrator, and dbt
+  // orchestrator pods
   private final ResourceRequirements resourceRequirements;
+  // for running replication orchestrator pods
+  private final ResourceRequirements replicationOrchestratorResourceRequirements;
+
   private final List<TolerationPOJO> workerKubeTolerations;
   private final Map<String, String> workerKubeNodeSelectors;
   private final String jobImagePullSecret;
@@ -30,6 +36,11 @@ public WorkerConfigs(final Configs configs) {
         .withCpuLimit(configs.getJobMainContainerCpuLimit())
         .withMemoryRequest(configs.getJobMainContainerMemoryRequest())
         .withMemoryLimit(configs.getJobMainContainerMemoryLimit());
+    this.replicationOrchestratorResourceRequirements = new ResourceRequirements()
+        .withCpuRequest(configs.getReplicationOrchestratorCpuRequest())
+        .withCpuLimit(configs.getReplicationOrchestratorCpuLimit())
+        .withMemoryRequest(configs.getReplicationOrchestratorMemoryRequest())
+        .withMemoryLimit(configs.getReplicationOrchestratorMemoryLimit());
     this.workerKubeTolerations = configs.getJobKubeTolerations();
     this.workerKubeNodeSelectors = configs.getJobKubeNodeSelectors();
     this.jobImagePullSecret = configs.getJobKubeMainContainerImagePullSecret();
@@ -80,4 +91,8 @@ public Map<String, String> getEnvMap() {
     return envMap;
   }
 
+  public ResourceRequirements getReplicationOrchestratorResourceRequirements() {
+    return replicationOrchestratorResourceRequirements;
+  }
+
 }

airbyte-workers/src/main/java/io/airbyte/workers/temporal/sync/ReplicationLauncherWorker.java

@@ -41,7 +41,7 @@ public ReplicationLauncherWorker(final UUID connectionId,
             INIT_FILE_SOURCE_LAUNCHER_CONFIG, Jsons.serialize(sourceLauncherConfig),
             INIT_FILE_DESTINATION_LAUNCHER_CONFIG, Jsons.serialize(destinationLauncherConfig)),
         containerOrchestratorConfig,
-        workerConfigs.getResourceRequirements(),
+        workerConfigs.getReplicationOrchestratorResourceRequirements(),
         ReplicationOutput.class);
   }