Add timeout to connector pod init container command (#10592)

* add timeout to init container command

* add disk usage check into init command

* fix up disk usage checking and logs from init entrypoint

* run format

Author: pmossman

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

@@ -119,6 +119,12 @@ public class KubePodProcess extends Process implements KubePod {
   private static final int KILLED_EXIT_CODE = 143;
   private static final int STDIN_REMOTE_PORT = 9001;
 
+  // init container should fail if no new data copied into the init container within
+  // INIT_RETRY_TIMEOUT_MINUTES
+  private static final double INIT_SLEEP_PERIOD_SECONDS = 0.1;
+  private static final Duration INIT_RETRY_TIMEOUT_MINUTES = Duration.ofMinutes(1);
+  private static final int INIT_RETRY_MAX_ITERATIONS = (int) (INIT_RETRY_TIMEOUT_MINUTES.toSeconds() / INIT_SLEEP_PERIOD_SECONDS);
+
   private final KubernetesClient fabricClient;
   private final Pod podDefinition;
   // Necessary since it is not possible to retrieve the pod's actual exit code upon termination. This
@@ -152,20 +158,23 @@ public static String getPodIP(final KubernetesClient client, final String podNam
 
   private static Container getInit(final boolean usesStdin,
                                    final List<VolumeMount> mainVolumeMounts,
-                                   final String busyboxImage) {
-    var initEntrypointStr = String.format("mkfifo %s && mkfifo %s", STDOUT_PIPE_FILE, STDERR_PIPE_FILE);
-
-    if (usesStdin) {
-      initEntrypointStr = String.format("mkfifo %s && ", STDIN_PIPE_FILE) + initEntrypointStr;
-    }
+                                   final String busyboxImage)
+      throws IOException {
 
-    initEntrypointStr = initEntrypointStr + String.format(" && until [ -f %s ]; do sleep 0.1; done;", SUCCESS_FILE_NAME);
+    final var initCommand = MoreResources.readResource("entrypoints/sync/init.sh")
+        .replaceAll("USES_STDIN_VALUE", String.valueOf(usesStdin))
+        .replaceAll("STDOUT_PIPE_FILE_VALUE", STDOUT_PIPE_FILE)
+        .replaceAll("STDERR_PIPE_FILE_VALUE", STDERR_PIPE_FILE)
+        .replaceAll("STDIN_PIPE_FILE_VALUE", STDIN_PIPE_FILE)
+        .replaceAll("MAX_ITERATION_VALUE", String.valueOf(INIT_RETRY_MAX_ITERATIONS))
+        .replaceAll("SUCCESS_FILE_NAME_VALUE", SUCCESS_FILE_NAME)
+        .replaceAll("SLEEP_PERIOD_VALUE", String.valueOf(INIT_SLEEP_PERIOD_SECONDS));
 
     return new ContainerBuilder()
         .withName(INIT_CONTAINER_NAME)
         .withImage(busyboxImage)
         .withWorkingDir(CONFIG_DIR)
-        .withCommand("sh", "-c", initEntrypointStr)
+        .withCommand("sh", "-c", initCommand)
         .withVolumeMounts(mainVolumeMounts)
         .build();
   }

airbyte-workers/src/main/resources/entrypoints/sync/init.sh

@@ -0,0 +1,30 @@
+USES_STDIN=USES_STDIN_VALUE
+
+mkfifo STDOUT_PIPE_FILE_VALUE
+mkfifo STDERR_PIPE_FILE_VALUE
+
+if [ "$USES_STDIN" = true ]; then
+  mkfifo STDIN_PIPE_FILE_VALUE
+fi
+
+ITERATION=0
+MAX_ITERATION=MAX_ITERATION_VALUE
+DISK_USAGE=$(du -s /config | awk '{print $1;}')
+
+until [ -f SUCCESS_FILE_NAME_VALUE -o $ITERATION -ge $MAX_ITERATION ]; do
+  ITERATION=$((ITERATION+1))
+  LAST_DISK_USAGE=$DISK_USAGE
+  DISK_USAGE=$(du -s /config | awk '{print $1;}')
+  if [ $DISK_USAGE -gt $LAST_DISK_USAGE ]; then
+    ITERATION=0
+  fi
+  sleep SLEEP_PERIOD_VALUE
+done
+
+if [ -f SUCCESS_FILE_NAME_VALUE ]; then
+  echo "All files copied successfully, exiting with code 0..."
+  exit 0
+else
+  echo "Timeout while attempting to copy to init container, exiting with code 1..."
+  exit 1
+fi