🐛 Kube: Fix Source Ports not releasing.

.github/workflows/gradle.yml

@@ -23,7 +23,7 @@ jobs:
           aws-region: us-east-2
       - name: Start EC2 Runner
         id: start-ec2-runner
-        uses: machulav/ec2-github-runner@v2
+        uses: machulav/ec2-github-runner@v2.2.1
         with:
           mode: start
           github-token: ${{ secrets.SELF_RUNNER_GITHUB_ACCESS_TOKEN }}
@@ -133,7 +133,7 @@ jobs:
           aws-secret-access-key: ${{ secrets.SELF_RUNNER_AWS_SECRET_ACCESS_KEY }}
           aws-region: us-east-2
       - name: Stop EC2 runner
-        uses: machulav/ec2-github-runner@v2.1.0
+        uses: machulav/ec2-github-runner@v2.2.1
         with:
           mode: stop
           github-token: ${{ secrets.SELF_RUNNER_GITHUB_ACCESS_TOKEN }}
@@ -157,7 +157,7 @@ jobs:
           aws-region: us-east-2
       - name: Start EC2 Runner
         id: start-ec2-runner
-        uses: machulav/ec2-github-runner@v2.2.0
+        uses: machulav/ec2-github-runner@v2.2.1
         with:
           mode: start
           github-token: ${{ secrets.SELF_RUNNER_GITHUB_ACCESS_TOKEN }}
@@ -301,7 +301,7 @@ jobs:
           aws-secret-access-key: ${{ secrets.SELF_RUNNER_AWS_SECRET_ACCESS_KEY }}
           aws-region: us-east-2
       - name: Stop EC2 runner
-        uses: machulav/ec2-github-runner@v2
+        uses: machulav/ec2-github-runner@v2.2.1
         with:
           mode: stop
           github-token: ${{ secrets.SELF_RUNNER_GITHUB_ACCESS_TOKEN }}
@@ -326,7 +326,7 @@ jobs:
           aws-region: us-east-2
       - name: Start EC2 Runner
         id: start-ec2-runner
-        uses: machulav/ec2-github-runner@v2
+        uses: machulav/ec2-github-runner@v2.2.1
         with:
           mode: start
           github-token: ${{ secrets.SELF_RUNNER_GITHUB_ACCESS_TOKEN }}
@@ -381,7 +381,7 @@ jobs:
           aws-secret-access-key: ${{ secrets.SELF_RUNNER_AWS_SECRET_ACCESS_KEY }}
           aws-region: us-east-2
       - name: Stop EC2 runner
-        uses: machulav/ec2-github-runner@v2
+        uses: machulav/ec2-github-runner@v2.2.1
         with:
           mode: stop
           github-token: ${{ secrets.SELF_RUNNER_GITHUB_ACCESS_TOKEN }}
@@ -406,7 +406,7 @@ jobs:
           aws-region: us-east-2
       - name: Start EC2 runner
         id: start-ec2-runner
-        uses: machulav/ec2-github-runner@v2
+        uses: machulav/ec2-github-runner@v2.2.1
         with:
           mode: start
           github-token: ${{ secrets.SELF_RUNNER_GITHUB_ACCESS_TOKEN }}
@@ -487,7 +487,7 @@ jobs:
           aws-secret-access-key: ${{ secrets.SELF_RUNNER_AWS_SECRET_ACCESS_KEY }}
           aws-region: us-east-2
       - name: Stop EC2 runner
-        uses: machulav/ec2-github-runner@v2
+        uses: machulav/ec2-github-runner@v2.2.1
         with:
           mode: stop
           github-token: ${{ secrets.SELF_RUNNER_GITHUB_ACCESS_TOKEN }}

airbyte-integrations/connector-templates/destination-python/destination_{{snakeCase name}}/spec.json

@@ -1,6 +1,8 @@
 {
   "documentationUrl": "https://docs.airbyte.io/integrations/destinations/{{kebabCase name}}",
-  "supported_destination_sync_modes": ["TODO, available options are: 'overwrite', 'append', and 'append_dedup'"],
+  "supported_destination_sync_modes": [
+    "TODO, available options are: 'overwrite', 'append', and 'append_dedup'"
+  ],
   "supportsIncremental": true,
   "supportsDBT": false,
   "supportsNormalization": false,

airbyte-integrations/connectors/destination-scaffold-destination-python/destination_scaffold_destination_python/spec.json

@@ -1,6 +1,8 @@
 {
   "documentationUrl": "https://docs.airbyte.io/integrations/destinations/scaffold-destination-python",
-  "supported_destination_sync_modes": ["TODO, available options are: 'overwrite', 'append', and 'append_dedup'"],
+  "supported_destination_sync_modes": [
+    "TODO, available options are: 'overwrite', 'append', and 'append_dedup'"
+  ],
   "supportsIncremental": true,
   "supportsDBT": false,
   "supportsNormalization": false,

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

@@ -52,8 +52,11 @@ public static void gentleClose(final Process process, final long timeout, final
       return;
     }
 
+    LOGGER.debug("Gently closing process {}", process.info().command().get());
     try {
-      process.waitFor(timeout, timeUnit);
+      if (process.isAlive()) {
+        process.waitFor(timeout, timeUnit);
+      }
     } catch (InterruptedException e) {
       LOGGER.error("Exception while while waiting for process to finish", e);
     }
@@ -100,9 +103,10 @@ static void gentleCloseWithHeartbeat(final Process process,
                                        final Duration checkHeartbeatDuration,
                                        final Duration forcedShutdownDuration,
                                        final BiConsumer<Process, Duration> forceShutdown) {
-
+    var processName = process.info().command().get();
     while (process.isAlive() && heartbeatMonitor.isBeating()) {
       try {
+        LOGGER.debug("Gently closing process {} with heartbeat..", processName);
         process.waitFor(checkHeartbeatDuration.toMillis(), TimeUnit.MILLISECONDS);
       } catch (InterruptedException e) {
         LOGGER.error("Exception while waiting for process to finish", e);
@@ -111,6 +115,7 @@ static void gentleCloseWithHeartbeat(final Process process,
 
     if (process.isAlive()) {
       try {
+        LOGGER.debug("Gently closing process {} without heartbeat..", processName);
         process.waitFor(gracefulShutdownDuration.toMillis(), TimeUnit.MILLISECONDS);
       } catch (InterruptedException e) {
         LOGGER.error("Exception during grace period for process to finish. This can happen when cancelling jobs.");
@@ -119,6 +124,7 @@ static void gentleCloseWithHeartbeat(final Process process,
 
     // if we were unable to exist gracefully, force shutdown...
     if (process.isAlive()) {
+      LOGGER.debug("Force shutdown process {}..", processName);
       forceShutdown.accept(process, forcedShutdownDuration);
     }
   }

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

@@ -47,6 +47,7 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.lang.ProcessHandle.Info;
 import java.net.InetAddress;
 import java.net.ServerSocket;
 import java.net.Socket;
@@ -528,6 +529,11 @@ public void destroy() {
     }
   }
 
+  @Override
+  public Info info() {
+    return new KubePodProcessInfo(podDefinition.getMetadata().getName());
+  }
+
   /**
    * Close all open resource in the opposite order of resource creation.
    */
@@ -538,8 +544,13 @@ private void close() {
     Exceptions.swallow(this.stdoutServerSocket::close);
     Exceptions.swallow(this.stderrServerSocket::close);
     Exceptions.swallow(this.executorService::shutdownNow);
-    Exceptions.swallow(() -> portReleaser.accept(stdoutLocalPort));
-    Exceptions.swallow(() -> portReleaser.accept(stderrLocalPort));
+    try {
+      portReleaser.accept(stdoutLocalPort);
+      portReleaser.accept(stderrLocalPort);
+    } catch (Exception e) {
+      LOGGER.error("Error releasing ports ", e);
+    }
+    LOGGER.debug("Closed {}", podDefinition.getMetadata().getName());
   }
 
   private boolean isTerminal(Pod pod) {
@@ -595,6 +606,14 @@ private int getReturnCode(Pod pod) {
         .orElseThrow();
 
     LOGGER.info("Exit code for pod {} is {}", name, returnCode);
+    // The OS traditionally handles process resource clean up. Therefore an exit code of 0, also
+    // indicates that all kernel resources were shut down.
+    // Because this is a custom implementation, manually close all the resources.
+    // Further, since the local resources are used to talk to Kubernetes resources, shut local resources
+    // down after Kubernetes resources are shut down,
+    // regardless of termination status.
+    close();
+    LOGGER.info("Closed all resources for pod {}", name);
     return returnCode;
   }
 

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

@@ -0,0 +1,75 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2020 Airbyte
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package io.airbyte.workers.process;
+
+import java.lang.ProcessHandle.Info;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Optional;
+
+/**
+ * Minimal Process info implementation to assist with debug logging.
+ *
+ * Current implement only logs out the Kubernetes pod corresponding to the JVM process.
+ */
+public class KubePodProcessInfo implements Info {
+
+  private final String podName;
+
+  public KubePodProcessInfo(String podname) {
+    this.podName = podname;
+  }
+
+  @Override
+  public Optional<String> command() {
+    return Optional.of(podName);
+  }
+
+  @Override
+  public Optional<String> commandLine() {
+    return Optional.of(podName);
+  }
+
+  @Override
+  public Optional<String[]> arguments() {
+    return Optional.empty();
+  }
+
+  @Override
+  public Optional<Instant> startInstant() {
+    return Optional.empty();
+  }
+
+  @Override
+  public Optional<Duration> totalCpuDuration() {
+    return Optional.empty();
+  }
+
+  @Override
+  public Optional<String> user() {
+    return Optional.empty();
+  }
+
+}

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

@@ -24,6 +24,7 @@
 
 package io.airbyte.workers.process;
 
+import com.google.common.annotations.VisibleForTesting;
 import io.airbyte.config.ResourceRequirements;
 import io.airbyte.workers.WorkerException;
 import io.fabric8.kubernetes.client.KubernetesClient;
@@ -79,21 +80,20 @@ public Process create(String jobId,
       throws WorkerException {
     try {
       // used to differentiate source and destination processes with the same id and attempt
-      final String suffix = RandomStringUtils.randomAlphabetic(5).toLowerCase();
-      final String podName = "airbyte-worker-" + jobId + "-" + attempt + "-" + suffix;
+      final String podName = getPodName(imageName, jobId, attempt);
 
       final int stdoutLocalPort = workerPorts.take();
-      LOGGER.info("stdoutLocalPort = " + stdoutLocalPort);
+      LOGGER.info("{} stdoutLocalPort = {}", podName, stdoutLocalPort);
 
       final int stderrLocalPort = workerPorts.take();
-      LOGGER.info("stderrLocalPort = " + stderrLocalPort);
+      LOGGER.info("{} stderrLocalPort = {}", podName, stderrLocalPort);
 
       final Consumer<Integer> portReleaser = port -> {
         if (!workerPorts.contains(port)) {
           workerPorts.add(port);
-          LOGGER.info("Port consumer releasing: " + port);
+          LOGGER.info("{} releasing: {}", podName, port);
         } else {
-          LOGGER.info("Port consumer skipping releasing: " + port);
+          LOGGER.info("{} skipping releasing: {}", podName, port);
         }
       };
 
@@ -117,4 +117,38 @@ public Process create(String jobId,
     }
   }
 
+  /**
+   * Docker image names are by convention separated by slashes. The last portion is the image's name.
+   * This is followed by a colon and a version number. e.g. airbyte/scheduler:v1 or
+   * gcr.io/my-project/image-name:v2.
+   *
+   * Kubernetes has a maximum pod name length of 63 characters.
+   *
+   * With these two facts, attempt to construct a unique Pod name with the image name present for
+   * easier operations.
+   */
+  @VisibleForTesting
+  protected static String getPodName(String fullImagePath, String jobId, int attempt) {
+    var versionDelimiter = ":";
+    var noVersion = fullImagePath.split(versionDelimiter)[0];
+
+    var dockerDelimiter = "/";
+    var nameParts = noVersion.split(dockerDelimiter);
+    var imageName = nameParts[nameParts.length - 1];
+
+    var randSuffix = RandomStringUtils.randomAlphabetic(5).toLowerCase();
+    final String suffix = "worker-" + jobId + "-" + attempt + "-" + randSuffix;
+
+    var podName = imageName + "-" + suffix;
+
+    var podNameLenLimit = 63;
+    if (podName.length() > podNameLenLimit) {
+      var extra = podName.length() - podNameLenLimit;
+      imageName = imageName.substring(extra);
+      podName = imageName + "-" + suffix;
+    }
+
+    return podName;
+  }
+
 }

airbyte-workers/src/main/java/io/airbyte/workers/protocols/airbyte/DefaultAirbyteDestination.java

@@ -112,6 +112,7 @@ public void notifyEndOfStream() throws IOException {
   @Override
   public void close() throws Exception {
     if (destinationProcess == null) {
+      LOGGER.debug("Destination process already exited");
       return;
     }
 

airbyte-workers/src/main/java/io/airbyte/workers/protocols/airbyte/DefaultAirbyteSource.java

@@ -118,6 +118,7 @@ public Optional<AirbyteMessage> attemptRead() {
   @Override
   public void close() throws Exception {
     if (sourceProcess == null) {
+      LOGGER.debug("Source process already exited");
       return;
     }
 

airbyte-workers/src/test/java/io/airbyte/workers/process/KubeProcessFactoryTest.java

@@ -0,0 +1,46 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2020 Airbyte
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package io.airbyte.workers.process;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class KubeProcessFactoryTest {
+
+  @Test
+  void getPodNameNormal() {
+    var name = KubeProcessFactory.getPodName("airbyte/tester:1", "1", 10);
+    var withoutRandSuffix = name.substring(0, name.length() - 5);
+    Assertions.assertEquals("tester-worker-1-10-", withoutRandSuffix);
+  }
+
+  @Test
+  void getPodNameTruncated() {
+    var name = KubeProcessFactory.getPodName("airbyte/very-very-very-long-name-longer-than-63-chars:2", "1", 10);
+    var withoutRandSuffix = name.substring(0, name.length() - 5);
+    Assertions.assertEquals("very-very-very-long-name-longer-than-63-chars-worker-1-10-", withoutRandSuffix);
+  }
+
+}