[FLINK-39173][runtime] Support application-level blob storage

Author: eemario

flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarRunApplicationHandler.java

@@ -27,8 +27,12 @@
 import org.apache.flink.configuration.DeploymentOptions;
 import org.apache.flink.configuration.StateRecoveryOptions;
 import org.apache.flink.core.execution.RecoveryClaimMode;
+import org.apache.flink.runtime.blob.BlobClient;
+import org.apache.flink.runtime.blob.PermanentBlobKey;
+import org.apache.flink.runtime.client.ClientUtils;
 import org.apache.flink.runtime.dispatcher.DispatcherGateway;
 import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
+import org.apache.flink.runtime.jobmanager.HighAvailabilityMode;
 import org.apache.flink.runtime.rest.handler.AbstractRestHandler;
 import org.apache.flink.runtime.rest.handler.HandlerRequest;
 import org.apache.flink.runtime.rest.handler.RestHandlerException;
@@ -40,6 +44,8 @@
 
 import javax.annotation.Nonnull;
 
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
 import java.nio.file.Path;
 import java.time.Duration;
 import java.util.Map;
@@ -103,7 +109,18 @@ public CompletableFuture<JarRunApplicationResponseBody> handleRequest(
                 new PackagedProgramApplication(
                         applicationId, program, effectiveConfiguration, false, true, false, false);
 
-        return gateway.submitApplication(application, timeout)
+        // TODO upload user jar to blob in HA mode once application resource cleanup is supported
+        final boolean isHaEnabled =
+                HighAvailabilityMode.isHighAvailabilityModeActivated(configuration);
+        CompletableFuture<PermanentBlobKey> jarUploadFuture =
+                CompletableFuture.completedFuture(null);
+
+        return jarUploadFuture
+                .thenCompose(
+                        blobKey -> {
+                            // TODO record blob key in the application for HA recovery
+                            return gateway.submitApplication(application, timeout);
+                        })
                 .handle(
                         (acknowledge, throwable) -> {
                             if (throwable != null) {
@@ -117,6 +134,48 @@ public CompletableFuture<JarRunApplicationResponseBody> handleRequest(
                         });
     }
 
+    private CompletableFuture<PermanentBlobKey> uploadJarFile(
+            final DispatcherGateway gateway,
+            final JarHandlerContext context,
+            final ApplicationID applicationId) {
+        CompletableFuture<Integer> blobServerPortFuture = gateway.getBlobServerPort(timeout);
+        CompletableFuture<InetAddress> blobServerAddressFuture =
+                gateway.getBlobServerAddress(timeout);
+
+        return blobServerPortFuture
+                .thenCombine(
+                        blobServerAddressFuture,
+                        (blobServerPort, blobServerAddress) ->
+                                new InetSocketAddress(
+                                        blobServerAddress.getHostName(), blobServerPort))
+                .thenApply(
+                        blobSocketAddress -> {
+                            try {
+                                org.apache.flink.core.fs.Path jarPath =
+                                        new org.apache.flink.core.fs.Path(
+                                                context.getJarFile().toString());
+                                PermanentBlobKey blobKey =
+                                        ClientUtils.uploadUserJarForApplication(
+                                                jarPath,
+                                                applicationId,
+                                                () ->
+                                                        new BlobClient(
+                                                                blobSocketAddress, configuration));
+                                log.info(
+                                        "Uploaded user jar for application {} to blob server with blob key {}.",
+                                        applicationId,
+                                        blobKey);
+                                return blobKey;
+                            } catch (Exception e) {
+                                throw new CompletionException(
+                                        new RestHandlerException(
+                                                "Could not upload jar file to blob server.",
+                                                HttpResponseStatus.INTERNAL_SERVER_ERROR,
+                                                e));
+                            }
+                        });
+    }
+
     private SavepointRestoreSettings getSavepointRestoreSettings(
             final @Nonnull HandlerRequest<JarRunApplicationRequestBody> request,
             final Configuration effectiveConfiguration)

flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/utils/JarHandlerUtils.java

@@ -238,6 +238,10 @@ JobID getJobId() {
         public Optional<ApplicationID> getApplicationId() {
             return Optional.ofNullable(applicationId);
         }
+
+        public Path getJarFile() {
+            return jarFile;
+        }
     }
 
     private static List<URL> getClasspaths(Configuration configuration) {

flink-runtime/src/main/java/org/apache/flink/runtime/blob/AbstractBlobCache.java

@@ -18,6 +18,7 @@
 
 package org.apache.flink.runtime.blob;
 
+import org.apache.flink.api.common.ApplicationID;
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.configuration.BlobServerOptions;
 import org.apache.flink.configuration.Configuration;
@@ -207,6 +208,94 @@ protected File getFileInternal(@Nullable JobID jobId, BlobKey blobKey) throws IO
         }
     }
 
+    /**
+     * Returns local copy of the file for the BLOB with the given key.
+     *
+     * <p>The method will first attempt to serve the BLOB from its local cache. If the BLOB is not
+     * in the cache, the method will try to download it from this cache's BLOB server via a
+     * distributed BLOB store (if available) or direct end-to-end download.
+     *
+     * @param applicationId ID of the application this blob belongs to
+     * @param blobKey The key of the desired BLOB.
+     * @return file referring to the local storage location of the BLOB.
+     * @throws IOException Thrown if an I/O error occurs while downloading the BLOBs from the BLOB
+     *     server.
+     */
+    protected File getFileInternal(ApplicationID applicationId, BlobKey blobKey)
+            throws IOException {
+        checkArgument(blobKey != null, "BLOB key cannot be null.");
+
+        final File localFile =
+                BlobUtils.getStorageLocation(storageDir.deref(), applicationId, blobKey);
+        readWriteLock.readLock().lock();
+
+        try {
+            if (localFile.exists()) {
+                return localFile;
+            }
+        } finally {
+            readWriteLock.readLock().unlock();
+        }
+
+        // first try the distributed blob store (if available)
+        // use a temporary file (thread-safe without locking)
+        File incomingFile = createTemporaryFilename();
+        try {
+            try {
+                if (blobView.get(applicationId, blobKey, incomingFile)) {
+                    // now move the temp file to our local cache atomically
+                    readWriteLock.writeLock().lock();
+                    try {
+                        BlobUtils.moveTempFileToStore(
+                                incomingFile, applicationId, blobKey, localFile, log, null);
+                    } finally {
+                        readWriteLock.writeLock().unlock();
+                    }
+
+                    return localFile;
+                }
+            } catch (Exception e) {
+                log.info(
+                        "Failed to copy from blob store. Downloading from BLOB server instead.", e);
+            }
+
+            final InetSocketAddress currentServerAddress = serverAddress;
+
+            if (currentServerAddress != null) {
+                // fallback: download from the BlobServer
+                BlobClient.downloadFromBlobServer(
+                        applicationId,
+                        blobKey,
+                        incomingFile,
+                        currentServerAddress,
+                        blobClientConfig,
+                        numFetchRetries);
+
+                readWriteLock.writeLock().lock();
+                try {
+                    BlobUtils.moveTempFileToStore(
+                            incomingFile, applicationId, blobKey, localFile, log, null);
+                } finally {
+                    readWriteLock.writeLock().unlock();
+                }
+            } else {
+                throw new IOException(
+                        "Cannot download from BlobServer, because the server address is unknown.");
+            }
+
+            return localFile;
+        } finally {
+            // delete incomingFile from a failed download
+            if (!incomingFile.delete() && incomingFile.exists()) {
+                log.warn(
+                        "Could not delete the staging file {} for blob key {} and application {}.",
+                        incomingFile,
+                        blobKey,
+                        applicationId);
+            }
+        }
+    }
+
     /**
      * Returns the port the BLOB server is listening on.
      *