Merge pull request #27365: Automatically use Docker Compose based transform service for PythonExternalTransform when needed.

chamikaramj · web-flow · commit f19c384af1f1 · 2023-07-06T15:07:16.000-07:00
diff --git a/sdks/java/extensions/python/src/main/java/org/apache/beam/sdk/extensions/python/PythonExternalTransform.java b/sdks/java/extensions/python/src/main/java/org/apache/beam/sdk/extensions/python/PythonExternalTransform.java
@@ -19,6 +19,7 @@
 
 import java.io.File;
 import java.io.FileOutputStream;
+import java.io.IOException;
 import java.io.OutputStreamWriter;
 import java.io.Writer;
 import java.util.ArrayList;
@@ -67,6 +68,8 @@
 import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Iterables;
 import org.checkerframework.checker.nullness.qual.NonNull;
 import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /** Wrapper for invoking external Python transforms. */
 public class PythonExternalTransform<InputT extends PInput, OutputT extends POutput>
@@ -88,6 +91,8 @@ public class PythonExternalTransform<InputT extends PInput, OutputT extends POut
 
   Map<String, Coder<?>> outputCoders;
 
+  private static final Logger LOG = LoggerFactory.getLogger(PythonExternalTransform.class);
+
   private PythonExternalTransform(String fullyQualifiedName, String expansionService) {
     this.fullyQualifiedName = fullyQualifiedName;
     this.expansionService = expansionService;
@@ -438,6 +443,29 @@ ExternalTransforms.ExternalConfigurationPayload generatePayload() {
     }
   }
 
+  private boolean isPythonAvailable() {
+    for (String executable : ImmutableList.of("python3", "python")) {
+      try {
+        new ProcessBuilder(executable, "--version").start().waitFor();
+        return true;
+      } catch (IOException | InterruptedException exn) {
+        // Ignore.
+      }
+    }
+    return false;
+  }
+
+  private boolean isDockerAvailable() {
+    String executable = "docker";
+    try {
+      new ProcessBuilder(executable, "--version").start().waitFor();
+      return true;
+    } catch (IOException | InterruptedException exn) {
+      // Ignore.
+    }
+    return false;
+  }
+
   @Override
   public OutputT expand(InputT input) {
     try {
@@ -458,14 +486,34 @@ public OutputT expand(InputT input) {
         OutputT output = null;
         int port = PythonService.findAvailablePort();
         PipelineOptionsFactory.register(PythonExternalTransformOptions.class);
-        if (input
-            .getPipeline()
-            .getOptions()
-            .as(PythonExternalTransformOptions.class)
-            .getUseTransformService()) {
+        boolean useTransformService =
+            input
+                .getPipeline()
+                .getOptions()
+                .as(PythonExternalTransformOptions.class)
+                .getUseTransformService();
+        boolean pythonAvailable = isPythonAvailable();
+        boolean dockerAvailable = isDockerAvailable();
+
+        // We use the transform service if either of the following is true.
+        // * It was explicitly requested.
+        // * Python executable is not available in the system but Docker is available.
+        if (useTransformService || (!pythonAvailable && dockerAvailable)) {
           // A unique project name ensures that this expansion gets a dedicated instance of the
           // transform service.
           String projectName = UUID.randomUUID().toString();
+
+          String messageAppend =
+              useTransformService
+                  ? "it was explicitly requested"
+                  : "a Python executable is not available in the system";
+          LOG.info(
+              "Using the Docker Compose based transform service since {}. Service will have the "
+                  + "project name {} and will be made available at the port {}",
+              messageAppend,
+              projectName,
+              port);
+
           TransformServiceLauncher service = TransformServiceLauncher.forProject(projectName, port);
           service.setBeamVersion(ReleaseInfo.getReleaseInfo().getSdkVersion());
           // TODO(https://github.com/apache/beam/issues/26833): add support for installing extra
diff --git a/sdks/java/transform-service/launcher/src/main/java/org/apache/beam/sdk/transformservice/launcher/TransformServiceLauncher.java b/sdks/java/transform-service/launcher/src/main/java/org/apache/beam/sdk/transformservice/launcher/TransformServiceLauncher.java
@@ -115,9 +115,11 @@ private TransformServiceLauncher(@Nullable String projectName, int port) throws
       if (applicationDefaultCredentialsFile.exists()) {
         Files.copy(applicationDefaultCredentialsFile, applicationDefaultCredentialsFileCopied);
       } else {
-        throw new RuntimeException(
-            "Could not find the application default file: "
-                + applicationDefaultCredentialsFile.getAbsolutePath());
+        LOG.error(
+            "GCP credentials will not be available for the transform service since the Google "
+                + "Cloud application default credentials file could not be found at the expected "
+                + "location {}.",
+            applicationDefaultFilePath);
       }
     }
 

Original file line number	Diff line number	Diff line change
`@@ -115,9 +115,11 @@ private TransformServiceLauncher(@Nullable String projectName, int port) throws`
`115`	`115`	`if (applicationDefaultCredentialsFile.exists()) {`
`116`	`116`	`Files.copy(applicationDefaultCredentialsFile, applicationDefaultCredentialsFileCopied);`
`117`	`117`	`} else {`
`118`		`- throw new RuntimeException(`
`119`		`- "Could not find the application default file: "`
`120`		`- + applicationDefaultCredentialsFile.getAbsolutePath());`
	`118`	`+ LOG.error(`
	`119`	`+ "GCP credentials will not be available for the transform service since the Google "`
	`120`	`+ + "Cloud application default credentials file could not be found at the expected "`
	`121`	`+ + "location {}.",`
	`122`	`+ applicationDefaultFilePath);`
`121`	`123`	`}`
`122`	`124`	`}`
`123`	`125`