add initContainer to container-orchestrator pod definition (#19088)

* init attempt at initcontainer

* wait for init container to be up instead of main container

* copy files to init container

* Revert "Bmoric/extract webbackend api (#18988)"

This reverts commit b05a5b2.

* Revert "Revert "Bmoric/extract webbackend api (#18988)""

This reverts commit ebef6e4.

* block on initContainer status; cleanup init script

* add log messages

* add quotes to log messages

* pr feedback, add comment to bash script

Author: colesnodgrass

airbyte-commons-worker/src/main/java/io/airbyte/workers/process/AsyncOrchestratorPodProcess.java

@@ -39,10 +39,10 @@
  * application. Unlike {@link KubePodProcess} there is no heartbeat mechanism that requires the
  * launching pod and the launched pod to co-exist for the duration of execution for the launched
  * pod.
- *
+ * <p>
  * Instead, this process creates the pod and interacts with a document store on cloud storage to
  * understand the state of the created pod.
- *
+ * <p>
  * The document store is considered to be the truth when retrieving the status for an async pod
  * process. If the store isn't updated by the underlying pod, it will appear as failed.
  */
@@ -190,10 +190,12 @@ public boolean hasExited() {
   public boolean waitFor(final long timeout, final TimeUnit unit) throws InterruptedException {
     // implementation copied from Process.java since this isn't a real Process
     long remainingNanos = unit.toNanos(timeout);
-    if (hasExited())
+    if (hasExited()) {
       return true;
-    if (timeout <= 0)
+    }
+    if (timeout <= 0) {
       return false;
+    }
 
     final long deadline = System.nanoTime() + remainingNanos;
     do {
@@ -202,8 +204,9 @@ public boolean waitFor(final long timeout, final TimeUnit unit) throws Interrupt
       // We are waiting polling every 500ms for status. The trade-off here is between how often
       // we poll our status storage (GCS) and how reactive we are to detect that a process is done.
       Thread.sleep(Math.min(TimeUnit.NANOSECONDS.toMillis(remainingNanos) + 1, 500));
-      if (hasExited())
+      if (hasExited()) {
         return true;
+      }
       remainingNanos = deadline - System.nanoTime();
     } while (remainingNanos > 0);
 
@@ -236,7 +239,7 @@ private boolean checkStatus(final AsyncKubePodStatus status) {
 
   /**
    * Checks terminal states first, then running, then initialized. Defaults to not started.
-   *
+   * <p>
    * The order matters here!
    */
   public AsyncKubePodStatus getDocStoreStatus() {
@@ -298,6 +301,33 @@ public void create(final Map<String, String> allLabels,
     final List<ContainerPort> containerPorts = KubePodProcess.createContainerPortList(portMap);
     containerPorts.add(new ContainerPort(serverPort, null, null, null, null));
 
+    final var initContainer = new ContainerBuilder()
+        .withName(KubePodProcess.INIT_CONTAINER_NAME)
+        .withImage("busybox:1.35")
+        .withVolumeMounts(volumeMounts)
+        .withCommand(List.of(
+            "sh",
+            "-c",
+            String.format("""
+                          i=0
+                          until [ $i -gt 60 ]
+                          do
+                            echo "$i - waiting for config file transfer to complete..."
+                            # check if the upload-complete file exists, if so exit without error
+                            if [ -f "%s/%s" ]; then
+                              exit 0
+                            fi
+                            i=$((i+1))
+                            sleep 1
+                          done
+                          echo "config files did not transfer in time"
+                          # no upload-complete file was created in time, exit with error
+                          exit 1
+                          """,
+                KubePodProcess.CONFIG_DIR,
+                KubePodProcess.SUCCESS_FILE_NAME)))
+        .build();
+
     final var mainContainer = new ContainerBuilder()
         .withName(KubePodProcess.MAIN_CONTAINER_NAME)
         .withImage(kubePodInfo.mainContainerInfo().image())
@@ -316,9 +346,11 @@ public void create(final Map<String, String> allLabels,
         .withLabels(allLabels)
         .endMetadata()
         .withNewSpec()
-        .withServiceAccount("airbyte-admin").withAutomountServiceAccountToken(true)
+        .withServiceAccount("airbyte-admin")
+        .withAutomountServiceAccountToken(true)
         .withRestartPolicy("Never")
         .withContainers(mainContainer)
+        .withInitContainers(initContainer)
         .withVolumes(volumes)
         .endSpec()
         .build();
@@ -332,9 +364,9 @@ public void create(final Map<String, String> allLabels,
     kubernetesClient.pods()
         .inNamespace(kubePodInfo.namespace())
         .withName(kubePodInfo.name())
-        .waitUntilCondition(p -> {
-          return !p.getStatus().getContainerStatuses().isEmpty() && p.getStatus().getContainerStatuses().get(0).getState().getWaiting() == null;
-        }, 5, TimeUnit.MINUTES);
+        .waitUntilCondition(p -> !p.getStatus().getInitContainerStatuses().isEmpty()
+            && p.getStatus().getInitContainerStatuses().get(0).getState().getWaiting() == null,
+            5, TimeUnit.MINUTES);
 
     final var podStatus = kubernetesClient.pods()
         .inNamespace(kubePodInfo.namespace())
@@ -343,7 +375,7 @@ public void create(final Map<String, String> allLabels,
         .getStatus();
 
     final var containerState = podStatus
-        .getContainerStatuses()
+        .getInitContainerStatuses()
         .get(0)
         .getState();
 
@@ -378,7 +410,7 @@ public static void copyFilesToKubeConfigVolumeMain(final Pod podDefinition, fina
         // several issues with copying files. See https://github.com/airbytehq/airbyte/issues/8643 for
         // details.
         final String command = String.format("kubectl cp %s %s/%s:%s -c %s", tmpFile, podDefinition.getMetadata().getNamespace(),
-            podDefinition.getMetadata().getName(), containerPath, "main");
+            podDefinition.getMetadata().getName(), containerPath, KubePodProcess.INIT_CONTAINER_NAME);
         log.info(command);
 
         proc = Runtime.getRuntime().exec(command);

airbyte-commons-worker/src/main/java/io/airbyte/workers/process/KubePodProcess.java

@@ -67,7 +67,7 @@
  * parent process starting a Kube Pod Process needs to exist within the Kube networking space. This
  * is so the parent process can forward data into the child's stdin and read the child's stdout and
  * stderr streams and copy configuration files over.
- *
+ * <p>
  * This is made possible by:
  * <ul>
  * <li>1) An init container that creates 3 named pipes corresponding to stdin, stdout and std err on
@@ -91,7 +91,7 @@
  * </ul>
  * The docker image used for this pod process must expose a AIRBYTE_ENTRYPOINT which contains the
  * entrypoint we will wrap when creating the main container in the pod.
- *
+ * <p>
  * See the constructor for more information.
  */
 
@@ -104,7 +104,7 @@ public class KubePodProcess extends Process implements KubePod {
   private static final Logger LOGGER = LoggerFactory.getLogger(KubePodProcess.class);
 
   public static final String MAIN_CONTAINER_NAME = "main";
-  private static final String INIT_CONTAINER_NAME = "init";
+  public static final String INIT_CONTAINER_NAME = "init";
   private static final String DEFAULT_MEMORY_REQUEST = "25Mi";
   private static final String DEFAULT_MEMORY_LIMIT = "50Mi";
   private static final String DEFAULT_CPU_REQUEST = "0.1";
@@ -701,7 +701,7 @@ public KubePodInfo getInfo() {
 
   /**
    * Close all open resource in the opposite order of resource creation.
-   *
+   * <p>
    * Null checks exist because certain local Kube clusters (e.g. Docker for Desktop) back this
    * implementation with OS processes and resources, which are automatically reaped by the OS.
    */