🐛 Kube: Fix Source Ports not releasing.

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

@@ -25,6 +25,7 @@
 package io.airbyte.workers;
 
 import com.google.common.annotations.VisibleForTesting;
+import io.airbyte.config.Configs.WorkerEnvironment;
 import io.airbyte.config.EnvConfigs;
 import io.airbyte.config.ResourceRequirements;
 import io.airbyte.config.StandardSyncInput;
@@ -41,6 +42,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+// TODO:(Issue-4824): Figure out how to log Docker process information.
 public class WorkerUtils {
 
   public static final ResourceRequirements DEFAULT_RESOURCE_REQUIREMENTS = initResourceRequirements();
@@ -52,8 +54,14 @@ public static void gentleClose(final Process process, final long timeout, final
       return;
     }
 
+    if (new EnvConfigs().getWorkerEnvironment().equals(WorkerEnvironment.KUBERNETES)) {
+      LOGGER.debug("Gently closing process {}", process.info().commandLine().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);
     }
@@ -103,6 +111,9 @@ static void gentleCloseWithHeartbeat(final Process process,
 
     while (process.isAlive() && heartbeatMonitor.isBeating()) {
       try {
+        if (new EnvConfigs().getWorkerEnvironment().equals(WorkerEnvironment.KUBERNETES)) {
+          LOGGER.debug("Gently closing process {} with heartbeat..", process.info().commandLine().get());
+        }
         process.waitFor(checkHeartbeatDuration.toMillis(), TimeUnit.MILLISECONDS);
       } catch (InterruptedException e) {
         LOGGER.error("Exception while waiting for process to finish", e);
@@ -111,6 +122,9 @@ static void gentleCloseWithHeartbeat(final Process process,
 
     if (process.isAlive()) {
       try {
+        if (new EnvConfigs().getWorkerEnvironment().equals(WorkerEnvironment.KUBERNETES)) {
+          LOGGER.debug("Gently closing process {} without heartbeat..", process.info().commandLine().get());
+        }
         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 +133,9 @@ static void gentleCloseWithHeartbeat(final Process process,
 
     // if we were unable to exist gracefully, force shutdown...
     if (process.isAlive()) {
+      if (new EnvConfigs().getWorkerEnvironment().equals(WorkerEnvironment.KUBERNETES)) {
+        LOGGER.debug("Force shutdown process {}..", process.info().commandLine().get());
+      }
       forceShutdown.accept(process, forcedShutdownDuration);
     }
   }
@@ -133,7 +150,7 @@ static void forceShutdown(Process process, Duration lastChanceDuration) {
       LOGGER.error("Exception while while killing the process", e);
     }
     if (process.isAlive()) {
-      LOGGER.error("Couldn't kill the process. You might have a zombie ({})", process.info().commandLine());
+      LOGGER.error("Couldn't kill the process. You might have a zombie process.");
     }
   }
 

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;
@@ -489,15 +490,11 @@ public InputStream getErrorStream() {
    */
   @Override
   public int waitFor() throws InterruptedException {
-    try {
-      Pod refreshedPod =
-          fabricClient.pods().inNamespace(podDefinition.getMetadata().getNamespace()).withName(podDefinition.getMetadata().getName()).get();
-      fabricClient.resource(refreshedPod).waitUntilCondition(this::isTerminal, 10, TimeUnit.DAYS);
-      wasKilled.set(true);
-      return exitValue();
-    } finally {
-      close();
-    }
+    Pod refreshedPod =
+        fabricClient.pods().inNamespace(podDefinition.getMetadata().getNamespace()).withName(podDefinition.getMetadata().getName()).get();
+    fabricClient.resource(refreshedPod).waitUntilCondition(this::isTerminal, 10, TimeUnit.DAYS);
+    wasKilled.set(true);
+    return exitValue();
   }
 
   /**
@@ -506,11 +503,7 @@ public int waitFor() throws InterruptedException {
    */
   @Override
   public boolean waitFor(long timeout, TimeUnit unit) throws InterruptedException {
-    try {
-      return super.waitFor(timeout, unit);
-    } finally {
-      close();
-    }
+    return super.waitFor(timeout, unit);
   }
 
   /**
@@ -528,6 +521,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 +536,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) {
@@ -600,7 +603,17 @@ private int getReturnCode(Pod pod) {
 
   @Override
   public int exitValue() {
-    return getReturnCode(podDefinition);
+    // getReturnCode throws IllegalThreadException if the Kube pod has not exited;
+    // close() is only called if the Kube pod has terminated.
+    var returnCode = getReturnCode(podDefinition);
+    // 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 Kube termination status.
+    close();
+    LOGGER.info("Closed all resources for pod {}", podDefinition.getMetadata().getName());
+    return returnCode;
   }
 
   private static ResourceRequirementsBuilder getResourceRequirementsBuilder(ResourceRequirements resourceRequirements) {

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 = createPodName(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 createPodName(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/DefaultDiscoverCatalogWorkerTest.java

@@ -26,8 +26,6 @@
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
@@ -107,7 +105,7 @@ public void testDiscoverSchema() throws Exception {
       }
     });
 
-    verify(process).waitFor(anyLong(), any());
+    verify(process).exitValue();
   }
 
   @SuppressWarnings("BusyWait")
@@ -124,7 +122,7 @@ public void testDiscoverSchemaProcessFail() throws Exception {
       }
     });
 
-    verify(process).waitFor(anyLong(), any());
+    verify(process).exitValue();
   }
 
   @Test

airbyte-workers/src/test/java/io/airbyte/workers/normalization/DefaultNormalizationRunnerTest.java

@@ -99,7 +99,7 @@ public void testClose() throws Exception {
     runner.normalize(JOB_ID, JOB_ATTEMPT, jobRoot, config, catalog, WorkerUtils.DEFAULT_RESOURCE_REQUIREMENTS);
     runner.close();
 
-    verify(process).destroy();
+    verify(process).waitFor();
   }
 
   @Test