This is an automated email from the ASF dual-hosted git repository. reswqa pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit ae20292b5c151714f2f64d84398dd4e5020eb9fc Author: Zihao Chen <[email protected]> AuthorDate: Tue Jul 7 14:43:07 2026 +0800 [FLINK-40097][historyserver] Prioritize on-demand fetching for accessed jobs --- .../generated/history_server_configuration.html | 6 + .../flink/configuration/HistoryServerOptions.java | 11 + .../history/AbstractHistoryServerHandler.java | 131 +++++++++++- .../webmonitor/history/ArchiveMetaInfo.java | 12 +- .../runtime/webmonitor/history/HistoryServer.java | 77 +++++-- .../HistoryServerApplicationArchiveFetcher.java | 12 +- .../history/HistoryServerArchiveFetcher.java | 124 ++++++++++- .../history/HistoryServerRocksDBHandler.java | 10 +- .../HistoryServerStaticFileServerHandler.java | 19 +- .../history/AbstractHistoryServerHandlerTest.java | 226 +++++++++++++++++++-- ...HistoryServerApplicationArchiveFetcherTest.java | 1 + .../history/HistoryServerArchiveFetcherTest.java | 78 ++++++- .../webmonitor/utils/WebFrontendBootstrapTest.java | 8 +- 13 files changed, 657 insertions(+), 58 deletions(-) diff --git a/docs/layouts/shortcodes/generated/history_server_configuration.html b/docs/layouts/shortcodes/generated/history_server_configuration.html index 6048ea7d5ee..ef521d5f999 100644 --- a/docs/layouts/shortcodes/generated/history_server_configuration.html +++ b/docs/layouts/shortcodes/generated/history_server_configuration.html @@ -68,6 +68,12 @@ <td>Integer</td> <td>The size of the common pool for archive fetching.</td> </tr> + <tr> + <td><h5>historyserver.lazy.fetch.executor.individual.pool-size</h5></td> + <td style="word-wrap: break-word;">4</td> + <td>Integer</td> + <td>The size of the individual pool for archive fetching.</td> + </tr> <tr> <td><h5>historyserver.log.jobmanager.url-pattern</h5></td> <td style="word-wrap: break-word;">(none)</td> diff --git a/flink-core/src/main/java/org/apache/flink/configuration/HistoryServerOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/HistoryServerOptions.java index 1e34bcc3656..05de6dee6ba 100644 --- a/flink-core/src/main/java/org/apache/flink/configuration/HistoryServerOptions.java +++ b/flink-core/src/main/java/org/apache/flink/configuration/HistoryServerOptions.java @@ -284,6 +284,17 @@ public class HistoryServerOptions { .text("The size of the common pool for archive fetching.") .build()); + public static final ConfigOption<Integer> + HISTORY_SERVER_LAZY_FETCH_EXECUTOR_INDIVIDUAL_POOL_SIZE = + key("historyserver.lazy.fetch.executor.individual.pool-size") + .intType() + .defaultValue(4) + .withDescription( + Description.builder() + .text( + "The size of the individual pool for archive fetching.") + .build()); + /** The type of archive storage. */ public enum HistoryServerArchiveStorageType { /** Local file system. */ diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/AbstractHistoryServerHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/AbstractHistoryServerHandler.java index 8b93ef01349..d86191d512f 100644 --- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/AbstractHistoryServerHandler.java +++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/AbstractHistoryServerHandler.java @@ -19,12 +19,17 @@ package org.apache.flink.runtime.webmonitor.history; +import org.apache.flink.configuration.HistoryServerOptions; +import org.apache.flink.core.fs.Path; import org.apache.flink.runtime.rest.NotFoundException; import org.apache.flink.runtime.rest.handler.RestHandlerException; import org.apache.flink.runtime.rest.handler.legacy.files.StaticFileServerHandler; import org.apache.flink.runtime.rest.handler.router.RoutedRequest; import org.apache.flink.runtime.rest.handler.util.HandlerUtils; +import org.apache.flink.runtime.rest.messages.ApplicationsOverviewHeaders; import org.apache.flink.runtime.rest.messages.ErrorResponseBody; +import org.apache.flink.runtime.rest.messages.JobsOverviewHeaders; +import org.apache.flink.util.Preconditions; import org.apache.flink.shaded.netty4.io.netty.channel.ChannelFuture; import org.apache.flink.shaded.netty4.io.netty.channel.ChannelFutureListener; @@ -45,6 +50,8 @@ import org.apache.flink.shaded.netty4.io.netty.handler.stream.ChunkedFile; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import javax.annotation.Nullable; + import java.io.File; import java.io.FileNotFoundException; import java.io.IOException; @@ -57,7 +64,11 @@ import java.text.SimpleDateFormat; import java.util.Collections; import java.util.Date; import java.util.Locale; +import java.util.regex.Matcher; +import java.util.regex.Pattern; +import static org.apache.flink.configuration.HistoryServerOptions.HistoryServerArchiveLoadMode.LAZY; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerArchiveFetcher.JSON_FILE_ENDING; import static org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpHeaderNames.CONNECTION; import static org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpHeaderNames.IF_MODIFIED_SINCE; import static org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus.INTERNAL_SERVER_ERROR; @@ -73,14 +84,34 @@ public abstract class AbstractHistoryServerHandler<Entry> private static final Logger LOG = LoggerFactory.getLogger(AbstractHistoryServerHandler.class); + /** Matches jobs/{jobId} or jobs/{jobId}/... paths. */ + private static final Pattern JOB_ID_PATTERN = Pattern.compile("^jobs/([^/]+)(?:/.*)?$"); + + /** Matches applications/{appId}. */ + private static final Pattern APPLICATION_ID_PATTERN = Pattern.compile("^applications/([^/]+)$"); + /** The path in which the static documents are. */ protected final File rootPath; protected final ArchiveStorage<Entry> archiveStorage; - protected AbstractHistoryServerHandler(ArchiveStorage<Entry> archiveStorage, File rootPath) + private final HistoryServerOptions.HistoryServerArchiveLoadMode archiveLoadMode; + + // for lazy load + @Nullable protected HistoryServerArchiveFetcher<Entry> archiveFetcher; + @Nullable protected HistoryServerApplicationArchiveFetcher<Entry> applicationArchiveFetcher; + + protected AbstractHistoryServerHandler( + ArchiveStorage<Entry> archiveStorage, + HistoryServerOptions.HistoryServerArchiveLoadMode archiveLoadMode, + @Nullable HistoryServerArchiveFetcher<Entry> archiveFetcher, + @Nullable HistoryServerApplicationArchiveFetcher<Entry> applicationArchiveFetcher, + File rootPath) throws IOException { this.archiveStorage = archiveStorage; + this.archiveLoadMode = archiveLoadMode; + this.archiveFetcher = archiveFetcher; + this.applicationArchiveFetcher = applicationArchiveFetcher; this.rootPath = checkNotNull(rootPath).getCanonicalFile(); } @@ -136,11 +167,12 @@ public abstract class AbstractHistoryServerHandler<Entry> requestPath = requestPath + "index.html"; } - if (!requestPath.contains(".")) { // we assume that the path ends in either .html or .js + // we assume that the path ends in either .html or .js + if (!requestPath.contains(".")) { requestPath = requestPath + ".json"; LOG.debug("Responding to request for path {}", requestPath); - Entry resource = loadResource(requestPath); + Entry resource = loadResource(requestPath, archiveLoadMode); if (resource == null) { LOG.debug("Unable to load requested resource {}", requestPath); @@ -266,14 +298,67 @@ public abstract class AbstractHistoryServerHandler<Entry> /** * Loads the resource for the given request path from the archive storage. * + * <p>The resource has four cases: + * + * <p>1. /index.html or other web resource files, should be loaded from classloader {@link + * #tryLoadFromClassloader} + * + * <p>2. /config.json, will be created when HistoryServer started + * + * <p>3. /jobs/overview.json (and /jobs/jobid.json) or /applications/overview.json (and + * /applications/applicationid.json), will be loaded synchronously + * + * <p>4. /jobs/<jobid>/.. will be loaded asynchronously + * * @param requestPath The request path + * @param archiveLoadMode The archive load mode * @return The resource for the given request path, or null if not found */ - private Entry loadResource(String requestPath) throws Exception { + private Entry loadResource( + String requestPath, HistoryServerOptions.HistoryServerArchiveLoadMode archiveLoadMode) + throws Exception { String requestKey = requestPath.startsWith("/") ? requestPath.substring(1) : requestPath; + + if (LAZY.equals(archiveLoadMode)) { + Preconditions.checkNotNull(archiveFetcher); + Preconditions.checkNotNull(applicationArchiveFetcher); + // need to update for overview + if (requestKey.equals(JobsOverviewHeaders.URL.substring(1) + JSON_FILE_ENDING) + || requestKey.equals( + ApplicationsOverviewHeaders.URL.substring(1) + JSON_FILE_ENDING)) { + archiveFetcher.fetchArchives(archiveLoadMode); + applicationArchiveFetcher.fetchArchives(archiveLoadMode); + return archiveStorage.getEntry(requestKey); + } + // for application/${applicationId}, return directly + String applicationId = extractApplicationId(requestKey); + if (applicationId != null) { + return archiveStorage.getEntry(requestKey); + } + // for job/${jobId}... + String jobId = extractJobId(requestKey); + if (jobId != null) { + if (archiveStorage.exists(requestKey)) { + return archiveStorage.getEntry(requestKey); + } + ArchiveMetaInfo jobArchiveMetaInfo = archiveFetcher.getArchiveMetaInfo(jobId); + if (archiveFetcher.needLazyLoadIndividually(jobId)) { + Path archivePath = + jobArchiveMetaInfo == null ? null : jobArchiveMetaInfo.getArchivePath(); + archiveFetcher.lazyFetchArchiveProactively(jobId, archivePath); + } + // wait for the job archive to be loaded + if (!requestKey.endsWith(jobId + JSON_FILE_ENDING)) { + archiveFetcher.waitLazyFetchArchiveFinished(jobId); + } + return archiveStorage.getEntry(requestKey); + } + } + if (archiveStorage.exists(requestKey)) { return archiveStorage.getEntry(requestKey); } + return null; } @@ -320,4 +405,42 @@ public abstract class AbstractHistoryServerHandler<Entry> } } } + + /** + * Extracts the job ID from the request path. + * + * @param requestPath Request path, e.g., {@code /jobs/abc123.../vertices} + * @return jobId string; returns {@code null} if path does not match + */ + @Nullable + protected static String extractJobId(String requestPath) { + Matcher matcher = JOB_ID_PATTERN.matcher(requestPath); + if (matcher.matches()) { + return matcher.group(1); + } + return null; + } + + /** + * Extracts the application ID from the request path. + * + * @param requestPath Request path, e.g., {@code /applications/abc123...} + * @return applicationId string; returns {@code null} if path does not match + */ + @Nullable + protected static String extractApplicationId(String requestPath) { + Matcher matcher = APPLICATION_ID_PATTERN.matcher(requestPath); + if (matcher.matches()) { + return matcher.group(1); + } + return null; + } + + /** Factory for creating instances of {@link AbstractHistoryServerHandler}. */ + public interface HistoryServerHandlerFactory { + AbstractHistoryServerHandler<?> createHistoryServerHandler( + HistoryServerArchiveFetcher<?> archiveFetcher, + HistoryServerApplicationArchiveFetcher<?> applicationArchiveFetcher) + throws IOException; + } } diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/ArchiveMetaInfo.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/ArchiveMetaInfo.java index 710868803dd..7a520506034 100644 --- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/ArchiveMetaInfo.java +++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/ArchiveMetaInfo.java @@ -18,16 +18,22 @@ package org.apache.flink.runtime.webmonitor.history; +import org.apache.flink.core.fs.Path; + /** Meta info for archived job. */ public class ArchiveMetaInfo { private final String archiveId; private volatile HistoryServerArchiveFetcher.ArchiveEventType eventType; + private final Path archivePath; public ArchiveMetaInfo( - String archiveId, HistoryServerArchiveFetcher.ArchiveEventType eventType) { + String archiveId, + HistoryServerArchiveFetcher.ArchiveEventType eventType, + Path archivePath) { this.archiveId = archiveId; this.eventType = eventType; + this.archivePath = archivePath; } public String getArchiveId() { @@ -41,4 +47,8 @@ public class ArchiveMetaInfo { public void setEventType(HistoryServerArchiveFetcher.ArchiveEventType eventType) { this.eventType = eventType; } + + public Path getArchivePath() { + return archivePath; + } } diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServer.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServer.java index d3e77c035da..151fd43b823 100644 --- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServer.java +++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServer.java @@ -77,6 +77,8 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Consumer; import static org.apache.flink.configuration.HistoryServerOptions.HISTORY_SERVER_LAZY_FETCH_EXECUTOR_COMMON_POOL_SIZE; +import static org.apache.flink.configuration.HistoryServerOptions.HISTORY_SERVER_LAZY_FETCH_EXECUTOR_INDIVIDUAL_POOL_SIZE; +import static org.apache.flink.configuration.HistoryServerOptions.HistoryServerArchiveLoadMode.LAZY; import static org.apache.flink.runtime.webmonitor.history.HistoryServerApplicationArchiveFetcher.APPLICATIONS_SUBDIR; import static org.apache.flink.runtime.webmonitor.history.HistoryServerApplicationArchiveFetcher.APPLICATION_OVERVIEWS_SUBDIR; import static org.apache.flink.runtime.webmonitor.history.HistoryServerArchiveFetcher.JOBS_SUBDIR; @@ -259,6 +261,7 @@ public class HistoryServer { archiveLoadMode = config.get(HistoryServerOptions.HISTORY_SERVER_ARCHIVE_LOAD_MODE); HistoryServerOptions.HistoryServerArchiveStorageType archiveStorageType = config.get(HistoryServerOptions.HISTORY_SERVER_ARCHIVE_STORAGE_TYPE); + AbstractHistoryServerHandler.HistoryServerHandlerFactory historyServerHandlerFactory; switch (archiveStorageType) { case FILE: // create directories for job and application overview updates @@ -267,17 +270,15 @@ public class HistoryServer { Files.createDirectories(webDir.toPath().resolve(APPLICATIONS_SUBDIR)); Files.createDirectories(webDir.toPath().resolve(APPLICATION_OVERVIEWS_SUBDIR)); archiveStorage = new FileArchiveStorage(webDir); - historyServerHandler = - new HistoryServerStaticFileServerHandler( - (FileArchiveStorage) archiveStorage, webDir); + historyServerHandlerFactory = + createFileHandlerFactory((FileArchiveStorage) archiveStorage, webDir); break; case ROCKSDB: File dbPath = new File(webDir, "rocksdb-" + UUID.randomUUID()); Files.createDirectories(dbPath.toPath()); archiveStorage = new RocksDBArchiveStorage(dbPath, config); - historyServerHandler = - new HistoryServerRocksDBHandler( - (RocksDBArchiveStorage) archiveStorage, webDir); + historyServerHandlerFactory = + createRocksDBHandlerFactory((RocksDBArchiveStorage) archiveStorage, webDir); break; default: throw new FlinkException("Unsupported archive storage type: " + archiveStorageType); @@ -288,6 +289,8 @@ public class HistoryServer { new ConcurrentHashMap<>(); int lazyFetchExecutorCommonPoolSize = config.get(HISTORY_SERVER_LAZY_FETCH_EXECUTOR_COMMON_POOL_SIZE); + int lazyFetchExecutorIndividualPoolSize = + config.get(HISTORY_SERVER_LAZY_FETCH_EXECUTOR_INDIVIDUAL_POOL_SIZE); archiveFetcher = new HistoryServerArchiveFetcher<>( refreshDirs, @@ -297,7 +300,8 @@ public class HistoryServer { CompositeArchiveRetainedStrategy.createForJobFromConfig(config), archiveStorage, archiveMetaInfoCache, - lazyFetchExecutorCommonPoolSize); + lazyFetchExecutorCommonPoolSize, + lazyFetchExecutorIndividualPoolSize); applicationArchiveFetcher = new HistoryServerApplicationArchiveFetcher<>( refreshDirs, @@ -308,7 +312,12 @@ public class HistoryServer { archiveStorage, archiveMetaInfoCache, applicationArchiveMetaInfoCache, - lazyFetchExecutorCommonPoolSize); + lazyFetchExecutorCommonPoolSize, + lazyFetchExecutorIndividualPoolSize); + + historyServerHandler = + historyServerHandlerFactory.createHistoryServerHandler( + archiveFetcher, applicationArchiveFetcher); this.shutdownHook = ShutdownHookUtil.addShutdownHook( @@ -364,6 +373,30 @@ public class HistoryServer { } } + @SuppressWarnings("unchecked") + private AbstractHistoryServerHandler.HistoryServerHandlerFactory createFileHandlerFactory( + FileArchiveStorage fileArchiveStorage, File webDir) { + return (archiveFetcher, applicationArchiveFetcher) -> + new HistoryServerStaticFileServerHandler( + fileArchiveStorage, + archiveLoadMode, + (HistoryServerArchiveFetcher<File>) archiveFetcher, + (HistoryServerApplicationArchiveFetcher<File>) applicationArchiveFetcher, + webDir); + } + + @SuppressWarnings("unchecked") + private AbstractHistoryServerHandler.HistoryServerHandlerFactory createRocksDBHandlerFactory( + RocksDBArchiveStorage rocksDBArchiveStorage, File webDir) { + return (archiveFetcher, applicationArchiveFetcher) -> + new HistoryServerRocksDBHandler( + rocksDBArchiveStorage, + archiveLoadMode, + (HistoryServerArchiveFetcher<String>) archiveFetcher, + (HistoryServerApplicationArchiveFetcher<String>) applicationArchiveFetcher, + webDir); + } + // ------------------------------------------------------------------------ // Life-cycle // ------------------------------------------------------------------------ @@ -399,11 +432,20 @@ public class HistoryServer { createDashboardConfigFile(); router.addGet("/:*", historyServerHandler); - executor.scheduleWithFixedDelay( - getArchiveFetchingRunnable(archiveLoadMode), - 0, - refreshIntervalMillis, - TimeUnit.MILLISECONDS); + if (LAZY.equals(archiveLoadMode)) { + executor.submit(getArchiveFetchingRunnable(archiveLoadMode)); + executor.scheduleWithFixedDelay( + getArchiveCleaningRunnable(), + refreshIntervalMillis, + refreshIntervalMillis, + TimeUnit.MILLISECONDS); + } else { + executor.scheduleWithFixedDelay( + getArchiveFetchingRunnable(archiveLoadMode), + 0, + refreshIntervalMillis, + TimeUnit.MILLISECONDS); + } netty = new WebFrontendBootstrap( @@ -421,6 +463,15 @@ public class HistoryServer { FatalExitExceptionHandler.INSTANCE); } + private Runnable getArchiveCleaningRunnable() { + return Runnables.withUncaughtExceptionHandler( + () -> { + archiveFetcher.cleanUpArchives(archiveLoadMode); + applicationArchiveFetcher.cleanUpArchives(archiveLoadMode); + }, + FatalExitExceptionHandler.INSTANCE); + } + void stop() { if (shutdownRequested.compareAndSet(false, true)) { synchronized (startupShutdownLock) { diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerApplicationArchiveFetcher.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerApplicationArchiveFetcher.java index f3b479aaf16..038abc85632 100644 --- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerApplicationArchiveFetcher.java +++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerApplicationArchiveFetcher.java @@ -89,7 +89,9 @@ public class HistoryServerApplicationArchiveFetcher<Entry> ArchiveStorage<Entry> archiveStorage, ConcurrentHashMap<String, ArchiveMetaInfo> archiveMetaInfoCache, ConcurrentHashMap<String, ArchiveMetaInfo> applicationArchiveMetaInfoCache, - int lazyFetchExecutorCommonPoolSize) { + int lazyFetchExecutorCommonPoolSize, + int lazyFetchExecutorIndividualPoolSize) + throws IOException { super( refreshDirs, webDir, @@ -98,7 +100,8 @@ public class HistoryServerApplicationArchiveFetcher<Entry> retainedStrategy, archiveStorage, archiveMetaInfoCache, - lazyFetchExecutorCommonPoolSize); + lazyFetchExecutorCommonPoolSize, + lazyFetchExecutorIndividualPoolSize); this.applicationArchiveMetaInfoCache = applicationArchiveMetaInfoCache; for (HistoryServer.RefreshLocation refreshDir : refreshDirs) { @@ -186,7 +189,7 @@ public class HistoryServerApplicationArchiveFetcher<Entry> .add(jobId); ArchiveEvent processArchiveEvents = LAZY.equals(archiveLoadMode) - ? lazyProcessJobArchive(jobId, jobArchive.getPath()) + ? lazyProcessJobArchive(jobId, jobArchive.getPath(), false) : processJobArchive(jobId, jobArchive.getPath()); events.add(processArchiveEvents); } @@ -314,7 +317,8 @@ public class HistoryServerApplicationArchiveFetcher<Entry> List<ArchiveEvent> lazyProcessArchive(String archiveId, Path archivePath, Path refreshDir) throws Exception { List<ArchiveEvent> events = new ArrayList<>(); - ArchiveMetaInfo archiveMetaInfo = new ArchiveMetaInfo(archiveId, OVERVIEW_PARSING); + ArchiveMetaInfo archiveMetaInfo = + new ArchiveMetaInfo(archiveId, OVERVIEW_PARSING, archivePath); ArchiveMetaInfo existing = applicationArchiveMetaInfoCache.putIfAbsent(archiveId, archiveMetaInfo); if (existing != null) { diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java index 2de41ed8091..e667d9cd365 100644 --- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java +++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java @@ -18,6 +18,7 @@ package org.apache.flink.runtime.webmonitor.history; +import org.apache.flink.annotation.VisibleForTesting; import org.apache.flink.api.common.JobID; import org.apache.flink.api.common.JobStatus; import org.apache.flink.configuration.HistoryServerOptions; @@ -40,6 +41,8 @@ import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMap import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import javax.annotation.Nullable; + import java.io.File; import java.io.IOException; import java.io.StringWriter; @@ -135,7 +138,9 @@ public class HistoryServerArchiveFetcher<Entry> implements AutoCloseable { /** Executor for loading archives. */ private final ExecutorService commonFetchExecutor; + private final ExecutorService individualFetchExecutor; private final Map<String, Future<?>> commonFetchTasks; + private final Map<String, Future<?>> individualFetchTasks; private final ConcurrentHashMap<String, ArchiveMetaInfo> archiveMetaInfoCache; HistoryServerArchiveFetcher( @@ -146,7 +151,9 @@ public class HistoryServerArchiveFetcher<Entry> implements AutoCloseable { ArchiveRetainedStrategy retainedStrategy, ArchiveStorage<Entry> archiveStorage, ConcurrentHashMap<String, ArchiveMetaInfo> archiveMetaInfoCache, - int lazyFetchExecutorCommonPoolSize) { + int lazyFetchExecutorCommonPoolSize, + int lazyFetchExecutorIndividualPoolSize) + throws IOException { this.refreshDirs = checkNotNull(refreshDirs); this.archiveEventListener = archiveEventListener; this.processExpiredArchiveDeletion = cleanupExpiredArchives; @@ -162,7 +169,12 @@ public class HistoryServerArchiveFetcher<Entry> implements AutoCloseable { Executors.newFixedThreadPool( lazyFetchExecutorCommonPoolSize, new ExecutorThreadFactory("HistoryServer-commonFetchExecutor")); + this.individualFetchExecutor = + Executors.newFixedThreadPool( + lazyFetchExecutorIndividualPoolSize, + new ExecutorThreadFactory("HistoryServer-individualFetchExecutor")); this.commonFetchTasks = new ConcurrentHashMap<>(); + this.individualFetchTasks = new ConcurrentHashMap<>(); updateJobOverview(); if (LOG.isInfoEnabled()) { @@ -174,6 +186,12 @@ public class HistoryServerArchiveFetcher<Entry> implements AutoCloseable { void fetchArchives(HistoryServerOptions.HistoryServerArchiveLoadMode archiveLoadMode) { LOG.debug("Starting archive fetching."); + scanArchives(archiveLoadMode, true); + } + + void scanArchives( + HistoryServerOptions.HistoryServerArchiveLoadMode archiveLoadMode, boolean fetch) { + LOG.debug("Starting archive fetching."); try { List<ArchiveEvent> events = new ArrayList<>(); Map<Path, Set<String>> archivesToRemove = new HashMap<>(); @@ -211,7 +229,9 @@ public class HistoryServerArchiveFetcher<Entry> implements AutoCloseable { continue; } - fetchArchive(refreshDir, archiveId, archivePath, archiveLoadMode, events); + if (fetch) { + fetchArchive(refreshDir, archiveId, archivePath, archiveLoadMode, events); + } } } @@ -228,7 +248,7 @@ public class HistoryServerArchiveFetcher<Entry> implements AutoCloseable { updateOverview(); } events.forEach(archiveEventListener); - LOG.debug("Finished archive fetching."); + LOG.debug("Finished archive scan."); } catch (Exception e) { LOG.error("Critical failure while fetching/processing archives.", e); } @@ -510,17 +530,30 @@ public class HistoryServerArchiveFetcher<Entry> implements AutoCloseable { // -------------------------------- Lazy Load ---------------------------------------- List<ArchiveEvent> lazyProcessArchive(String archiveId, Path archivePath, Path refreshDir) throws Exception { - return Collections.singletonList(lazyProcessJobArchive(archiveId, archivePath)); + return Collections.singletonList(lazyProcessJobArchive(archiveId, archivePath, false)); } - ArchiveEvent lazyProcessJobArchive(String jobId, Path jobArchive) throws Exception { - final ArchiveMetaInfo archiveMetaInfo = new ArchiveMetaInfo(jobId, PENDING); + ArchiveEvent lazyProcessJobArchive(String jobId, Path jobArchive, boolean individual) + throws Exception { + final ArchiveMetaInfo archiveMetaInfo = new ArchiveMetaInfo(jobId, PENDING, jobArchive); ArchiveMetaInfo existing = archiveMetaInfoCache.putIfAbsent(jobId, archiveMetaInfo); if (existing != null) { return new ArchiveEvent(jobId, existing.getEventType()); } archiveMetaInfo.setEventType(ArchiveEventType.OVERVIEW_PARSING); + ExecutorService fetchExecutor; + Map<String, Future<?>> fetchTasks; + if (individual) { + fetchExecutor = individualFetchExecutor; + fetchTasks = individualFetchTasks; + } else { + fetchExecutor = commonFetchExecutor; + fetchTasks = commonFetchTasks; + } + + archiveMetaInfo.setEventType(ArchiveEventType.OVERVIEW_PARSING); + Collection<ArchivedJson> archivedJsons = FsJsonArchivist.readArchivedJsons(jobArchive); List<ArchivedJson> detailArchives = new ArrayList<>(); boolean overviewCreated = false; @@ -554,7 +587,7 @@ public class HistoryServerArchiveFetcher<Entry> implements AutoCloseable { if (!detailArchives.isEmpty()) { Future<?> future = - commonFetchExecutor.submit( + fetchExecutor.submit( () -> { try { archiveMetaInfo.setEventType(ArchiveEventType.DETAIL_PARSING); @@ -577,12 +610,15 @@ public class HistoryServerArchiveFetcher<Entry> implements AutoCloseable { } } archiveMetaInfo.setEventType(ArchiveEventType.CREATED); + if (individual) { + updateOverview(); + } LOG.debug("Async detail parsing for job {} finished.", jobId); } finally { - commonFetchTasks.remove(jobId); + fetchTasks.remove(jobId); } }); - commonFetchTasks.put(jobId, future); + fetchTasks.put(jobId, future); } ArchiveEventType archiveEventType = @@ -591,15 +627,85 @@ public class HistoryServerArchiveFetcher<Entry> implements AutoCloseable { return new ArchiveEvent(jobId, archiveEventType); } + void lazyFetchArchiveProactively(String jobId, @Nullable Path archivePath) throws Exception { + resetWhenTriggerLazyFetch(jobId); + + if (archivePath != null) { + lazyProcessJobArchive(jobId, archivePath, true); + return; + } + + for (HistoryServer.RefreshLocation refreshDir : refreshDirs) { + archivePath = new Path(refreshDir.getPath(), jobId); + if (refreshDir.getFs().exists(archivePath)) { + lazyProcessJobArchive(jobId, archivePath, true); + } + } + } + + void cleanUpArchives(HistoryServerOptions.HistoryServerArchiveLoadMode archiveLoadMode) { + LOG.debug("Starting archive cleanup."); + scanArchives(archiveLoadMode, false); + } + + boolean needLazyLoadIndividually(String jobId) { + ArchiveMetaInfo archiveMetaInfo = archiveMetaInfoCache.get(jobId); + if (archiveMetaInfo == null) { + return true; + } + + switch (archiveMetaInfo.getEventType()) { + case PENDING: + case OVERVIEW_PARSING: + case OVERVIEW_CREATED: + return commonFetchTasks.containsKey(jobId) + && !individualFetchTasks.containsKey(jobId); + default: + return false; + } + } + void cleanUpLazyFetchTask(String jobId) { Future<?> commonFetchTask = commonFetchTasks.get(jobId); if (commonFetchTask != null) { commonFetchTask.cancel(true); + commonFetchTasks.remove(jobId); + } + Future<?> individualFetchTask = individualFetchTasks.get(jobId); + if (individualFetchTask != null) { + individualFetchTask.cancel(true); + individualFetchTasks.remove(jobId); } } @Override public void close() { ExecutorUtils.gracefulShutdown(1L, TimeUnit.SECONDS, commonFetchExecutor); + ExecutorUtils.gracefulShutdown(1L, TimeUnit.SECONDS, individualFetchExecutor); + } + + @VisibleForTesting + Future<?> getCommonFetchTask(String jobId) { + return commonFetchTasks.get(jobId); + } + + void resetWhenTriggerLazyFetch(String jobId) { + archiveMetaInfoCache.remove(jobId); + cleanUpLazyFetchTask(jobId); + } + + void waitLazyFetchArchiveFinished(String jobId) throws Exception { + Future<?> commonFetchTask = commonFetchTasks.get(jobId); + if (commonFetchTask != null) { + commonFetchTask.get(); + } + Future<?> individualFetchTask = individualFetchTasks.get(jobId); + if (individualFetchTask != null) { + individualFetchTask.get(); + } + } + + ArchiveMetaInfo getArchiveMetaInfo(String jobId) { + return archiveMetaInfoCache.get(jobId); } } diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerRocksDBHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerRocksDBHandler.java index 147bc93eaf9..f6b5a67a0ce 100644 --- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerRocksDBHandler.java +++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerRocksDBHandler.java @@ -19,6 +19,7 @@ package org.apache.flink.runtime.webmonitor.history; +import org.apache.flink.configuration.HistoryServerOptions; import org.apache.flink.runtime.rest.handler.util.HandlerUtils; import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandler; @@ -40,9 +41,14 @@ import java.util.Collections; @ChannelHandler.Sharable public class HistoryServerRocksDBHandler extends AbstractHistoryServerHandler<String> { - public HistoryServerRocksDBHandler(RocksDBArchiveStorage rocksDBArchiveStorage, File rootPath) + public HistoryServerRocksDBHandler( + ArchiveStorage<String> archiveStorage, + HistoryServerOptions.HistoryServerArchiveLoadMode archiveLoadMode, + HistoryServerArchiveFetcher<String> archiveFetcher, + HistoryServerApplicationArchiveFetcher<String> applicationArchiveFetcher, + File rootPath) throws IOException { - super(rocksDBArchiveStorage, rootPath); + super(archiveStorage, archiveLoadMode, archiveFetcher, applicationArchiveFetcher, rootPath); } @Override diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerStaticFileServerHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerStaticFileServerHandler.java index 695f8edfffe..28e0e0b7a7c 100644 --- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerStaticFileServerHandler.java +++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerStaticFileServerHandler.java @@ -26,6 +26,8 @@ package org.apache.flink.runtime.webmonitor.history; * https://github.com/netty/netty/blob/4.0/example/src/main/java/io/netty/example/http/file/HttpStaticFileServerHandler.java * *************************************************************************** */ +import org.apache.flink.annotation.VisibleForTesting; +import org.apache.flink.configuration.HistoryServerOptions; import org.apache.flink.runtime.rest.handler.legacy.files.StaticFileServerHandler; import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandler; @@ -35,6 +37,8 @@ import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpRequest; import java.io.File; import java.io.IOException; +import static org.apache.flink.configuration.HistoryServerOptions.HistoryServerArchiveLoadMode.EAGER; + /** * Simple file server handler used by the {@link HistoryServer} that serves requests to web * frontend's static files, such as HTML, CSS, JS or JSON files. @@ -52,13 +56,20 @@ public class HistoryServerStaticFileServerHandler extends AbstractHistoryServerH // ------------------------------------------------------------------------ - public HistoryServerStaticFileServerHandler(File rootPath) throws IOException { - this(new FileArchiveStorage(rootPath), rootPath); + @VisibleForTesting + public HistoryServerStaticFileServerHandler(ArchiveStorage<File> archiveStorage, File rootPath) + throws IOException { + this(archiveStorage, EAGER, null, null, rootPath); } public HistoryServerStaticFileServerHandler( - FileArchiveStorage fileArchiveStorage, File rootPath) throws IOException { - super(fileArchiveStorage, rootPath); + ArchiveStorage<File> archiveStorage, + HistoryServerOptions.HistoryServerArchiveLoadMode archiveLoadMode, + HistoryServerArchiveFetcher<File> archiveFetcher, + HistoryServerApplicationArchiveFetcher<File> applicationArchiveFetcher, + File rootPath) + throws IOException { + super(archiveStorage, archiveLoadMode, archiveFetcher, applicationArchiveFetcher, rootPath); } // ------------------------------------------------------------------------ diff --git a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/AbstractHistoryServerHandlerTest.java b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/AbstractHistoryServerHandlerTest.java index 479f3dcb07b..cbff6ef4496 100644 --- a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/AbstractHistoryServerHandlerTest.java +++ b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/AbstractHistoryServerHandlerTest.java @@ -18,8 +18,10 @@ package org.apache.flink.runtime.webmonitor.history; +import org.apache.flink.api.common.JobID; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.HistoryServerOptions.HistoryServerArchiveLoadMode; import org.apache.flink.runtime.rest.handler.router.Router; import org.apache.flink.runtime.webmonitor.testutils.HttpUtils; import org.apache.flink.runtime.webmonitor.utils.WebFrontendBootstrap; @@ -35,12 +37,23 @@ import org.junit.jupiter.api.io.TempDir; import org.slf4j.LoggerFactory; import java.io.File; +import java.io.IOException; import java.nio.file.Files; import java.nio.file.Path; import java.util.Arrays; import java.util.Collection; +import java.util.Collections; +import java.util.List; import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import static org.apache.flink.configuration.HistoryServerOptions.HistoryServerArchiveLoadMode.EAGER; +import static org.apache.flink.configuration.HistoryServerOptions.HistoryServerArchiveLoadMode.LAZY; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerTestUtils.createJobArchive; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerTestUtils.createRefreshLocation; import static org.assertj.core.api.Assertions.assertThat; /** @@ -50,10 +63,37 @@ import static org.assertj.core.api.Assertions.assertThat; @ExtendWith(ParameterizedTestExtension.class) public class AbstractHistoryServerHandlerTest { + @Parameter public HandlerFactory handlerFactory; + + @TempDir private Path tmpDir; + private Path uploadDir; + private Path remoteArchiveRootPath; + + private Path webDir; + private AbstractHistoryServerHandler<?> handler; + private WebFrontendBootstrap webUI; + private String baseUrl; + /** Factory that creates a concrete handler bound to the given web directory. */ @FunctionalInterface public interface HandlerFactory { - AbstractHistoryServerHandler<?> create(File webDir) throws Exception; + AbstractHistoryServerHandler<?> create( + File webDir, + HistoryServerArchiveLoadMode archiveLoadMode, + List<HistoryServer.RefreshLocation> refreshDirs) + throws Exception; + } + + /** Constructor reference for a concrete {@link AbstractHistoryServerHandler} subclass. */ + @FunctionalInterface + private interface HandlerConstructor<T> { + AbstractHistoryServerHandler<T> create( + ArchiveStorage<T> archiveStorage, + HistoryServerArchiveLoadMode archiveLoadMode, + HistoryServerArchiveFetcher<T> archiveFetcher, + HistoryServerApplicationArchiveFetcher<T> applicationArchiveFetcher, + File webDir) + throws IOException; } /** @@ -62,32 +102,98 @@ public class AbstractHistoryServerHandlerTest { */ @Parameters(name = "handlerFactory={0}") private static Collection<HandlerFactory> handlerFactories() { - HandlerFactory staticFileServerHandlerFactory = HistoryServerStaticFileServerHandler::new; + HandlerFactory staticFileServerHandlerFactory = + (webDir, mode, refreshDirs) -> + buildHandler( + webDir, + mode, + refreshDirs, + new FileArchiveStorage(webDir), + HistoryServerStaticFileServerHandler::new); + HandlerFactory rocksDBHandlerFactory = - webDir -> - new HistoryServerRocksDBHandler( - new RocksDBArchiveStorage( - new File(webDir, "rocksdb-" + UUID.randomUUID()), - new Configuration()), - webDir); + (webDir, mode, refreshDirs) -> { + File dbPath = new File(webDir, "rocksdb-" + UUID.randomUUID()); + Files.createDirectories(dbPath.toPath()); + return buildHandler( + webDir, + mode, + refreshDirs, + new RocksDBArchiveStorage(dbPath, new Configuration()), + HistoryServerRocksDBHandler::new); + }; + return Arrays.asList(staticFileServerHandlerFactory, rocksDBHandlerFactory); } - @Parameter public HandlerFactory handlerFactory; + private static <T> AbstractHistoryServerHandler<?> buildHandler( + File webDir, + HistoryServerArchiveLoadMode mode, + List<HistoryServer.RefreshLocation> refreshDirs, + ArchiveStorage<T> baseStorage, + HandlerConstructor<T> handlerCtor) + throws Exception { + ArchiveStorage<T> storage = + LAZY == mode + ? new HistoryServerTestUtils.BlockingArchiveStorage<>( + baseStorage, "/config") + : baseStorage; - @TempDir private Path tmpDir; + ConcurrentHashMap<String, ArchiveMetaInfo> jobMetaInfoCache = new ConcurrentHashMap<>(); + ConcurrentHashMap<String, ArchiveMetaInfo> applicationMetaInfoCache = + new ConcurrentHashMap<>(); - private Path webDir; - private AbstractHistoryServerHandler<?> handler; - private WebFrontendBootstrap webUI; - private String baseUrl; + HistoryServerArchiveFetcher<T> archiveFetcher = + new HistoryServerArchiveFetcher<>( + refreshDirs, + webDir, + ignored -> {}, + false, + HistoryServerTestUtils.RETAIN_ALL, + storage, + jobMetaInfoCache, + 4, + 4); + HistoryServerApplicationArchiveFetcher<T> applicationArchiveFetcher = + new HistoryServerApplicationArchiveFetcher<>( + refreshDirs, + webDir, + ignored -> {}, + false, + HistoryServerTestUtils.RETAIN_ALL, + storage, + jobMetaInfoCache, + applicationMetaInfoCache, + 4, + 4); + + return handlerCtor.create(storage, mode, archiveFetcher, applicationArchiveFetcher, webDir); + } @BeforeEach void setUp() throws Exception { webDir = Files.createDirectory(tmpDir.resolve("webDir")); - final Path uploadDir = Files.createDirectory(tmpDir.resolve("uploadDir")); + uploadDir = Files.createDirectory(tmpDir.resolve("uploadDir")); + remoteArchiveRootPath = Files.createDirectories(tmpDir.resolve("remote")); + + // Default: eager mode + startServer(EAGER); + } + + @AfterEach + void tearDown() { + stopServer(); + } + + /** Support recreates the handler and web frontend for the given load mode. */ + private void startServer(HistoryServerArchiveLoadMode archiveLoadMode) throws Exception { + stopServer(); + + List<HistoryServer.RefreshLocation> refreshDirs = + Collections.singletonList(createRefreshLocation(remoteArchiveRootPath.toFile())); + + this.handler = handlerFactory.create(webDir.toFile(), archiveLoadMode, refreshDirs); - handler = handlerFactory.create(webDir.toFile()); Router<?> router = new Router().addGet("/:*", handler); webUI = new WebFrontendBootstrap( @@ -101,10 +207,10 @@ public class AbstractHistoryServerHandlerTest { baseUrl = "http://localhost:"; + webUI.getServerPort(); } - @AfterEach - void tearDown() { + private void stopServer() { if (webUI != null) { webUI.shutdown(); + webUI = null; } } @@ -184,4 +290,88 @@ public class AbstractHistoryServerHandlerTest { assertThat(missing.f0).isEqualTo(404); assertThat(missing.f1).contains("not found"); } + + /** + * Tests {@code AbstractHistoryServerHandler#loadResource} in {@code LAZY} mode using a {@link + * HistoryServerTestUtils.BlockingArchiveStorage} to suspend writes for the detail key {@code + * jobs/<jobId>/config}. + * + * <p>Verifies the three core lazy-load behaviours: + * + * <ul> + * <li>requesting {@code /jobs/overview} triggers a synchronous phase-1 fetch that writes the + * overview keys but leaves the detail key blocked in an asynchronous task; + * <li>once phase-1 has completed, requesting {@code /jobs/<jobId>} is served immediately from + * the archive storage without waiting for the asynchronous detail task; + * <li>requesting {@code /jobs/<jobId>/config} blocks until the asynchronous detail task is + * allowed to finish, after which the response is served successfully. + * </ul> + */ + @TestTemplate + void testLazyModeLoadResource() throws Exception { + // Default mode is EAGER, we need to recreate the handler for LAZY mode with + // BlockingArchiveStorage. + startServer(LAZY); + + JobID jobId = JobID.generate(); + createJobArchive(remoteArchiveRootPath.toFile(), jobId, true); + + // Phase 1: requesting /jobs/overview triggers a synchronous fetch + // that writes the overview keys; the detail key is queued for an + // asynchronous write that is now blocked on releaseLatch. + Tuple2<Integer, String> overviewResponse = + HttpUtils.getFromHTTP(baseUrl + "/jobs/overview"); + assertThat(overviewResponse.f0).isEqualTo(200); + assertThat(overviewResponse.f1).contains(jobId.toString()); + + // The asynchronous detail write must have been reached by now. + assertThat(handler.archiveStorage) + .isInstanceOf(HistoryServerTestUtils.BlockingArchiveStorage.class); + HistoryServerTestUtils.BlockingArchiveStorage<?> blockingArchiveStorage = + (HistoryServerTestUtils.BlockingArchiveStorage<?>) handler.archiveStorage; + boolean asyncReached = blockingArchiveStorage.asyncStartLatch.await(10, TimeUnit.SECONDS); + assertThat(asyncReached).isTrue(); + + // Phase 1 keys are present, but the detail key is still blocked. + assertThat(blockingArchiveStorage.exists("overviews/" + jobId + ".json")).isTrue(); + assertThat(blockingArchiveStorage.exists("jobs/" + jobId + ".json")).isTrue(); + assertThat(blockingArchiveStorage.exists("jobs/overview.json")).isTrue(); + assertThat(blockingArchiveStorage.exists("jobs/" + jobId + "/config.json")).isFalse(); + + // Phase 2: a request for /jobs/<jobId> can be served immediately + // from the storage even though the detail task is still blocked. + Tuple2<Integer, String> jobResponse = HttpUtils.getFromHTTP(baseUrl + "/jobs/" + jobId); + assertThat(jobResponse.f0).isEqualTo(200); + assertThat(jobResponse.f1).contains(jobId.toString()); + assertThat(blockingArchiveStorage.exists("jobs/" + jobId + "/config.json")).isFalse(); + + // Phase 3: a request for /jobs/<jobId>/config will wait for the + // asynchronous detail task to finish. Issue it on a separate + // thread, verify it is still in flight, then release the latch. + AtomicReference<Tuple2<Integer, String>> detailResponse = new AtomicReference<>(); + CompletableFuture<Void> detailFuture = + CompletableFuture.runAsync( + () -> { + try { + detailResponse.set( + HttpUtils.getFromHTTP( + baseUrl + "/jobs/" + jobId + "/config")); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + + // Give the detail request enough time to reach + // waitLazyFetchArchiveFinished and block on the still-suspended async + // detail task. It must not have completed yet. + Thread.sleep(500); + assertThat(detailFuture.isDone()).isFalse(); + + blockingArchiveStorage.releaseLatch.countDown(); + + detailFuture.get(10, TimeUnit.SECONDS); + assertThat(detailResponse.get().f0).isEqualTo(200); + assertThat(detailResponse.get().f1).contains(jobId.toString()); + assertThat(blockingArchiveStorage.exists("jobs/" + jobId + "/config.json")).isTrue(); + } } diff --git a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerApplicationArchiveFetcherTest.java b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerApplicationArchiveFetcherTest.java index 8775be5eb7a..05f5e168389 100644 --- a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerApplicationArchiveFetcherTest.java +++ b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerApplicationArchiveFetcherTest.java @@ -194,6 +194,7 @@ class HistoryServerApplicationArchiveFetcherTest { storage, jobMetaInfoCache, applicationMetaInfoCache, + 4, 4); } diff --git a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcherTest.java b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcherTest.java index ff929837fd9..612d156f244 100644 --- a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcherTest.java +++ b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcherTest.java @@ -121,6 +121,7 @@ class HistoryServerArchiveFetcherTest { RETAIN_ALL, storage, archiveMetaInfoCache, + 4, 4); } @@ -231,7 +232,8 @@ class HistoryServerArchiveFetcherTest { HistoryServerArchiveFetcher<?> fetcher = createArchiveFetcher(remoteArchiveRootPath, false, archiveStorage); - assertThatThrownBy(() -> fetcher.lazyProcessJobArchive(jobId.toString(), archivePath)) + assertThatThrownBy( + () -> fetcher.lazyProcessJobArchive(jobId.toString(), archivePath, false)) .isInstanceOf(RuntimeException.class) .hasMessage("Archive of job " + jobId + " is empty"); @@ -275,7 +277,8 @@ class HistoryServerArchiveFetcherTest { // try to call lazyProcessJobArchive again, should return DETAIL_PARSING. HistoryServerArchiveFetcher.ArchiveEvent callAgainEvent = - fetcher.lazyProcessJobArchive(jobIdWithDetail.toString(), jobWithDetailArchivePath); + fetcher.lazyProcessJobArchive( + jobIdWithDetail.toString(), jobWithDetailArchivePath, false); assertThat(callAgainEvent.getType()).isEqualTo(DETAIL_PARSING); // Phase 2: execute the asynchronous task @@ -352,4 +355,75 @@ class HistoryServerArchiveFetcherTest { assertThat(archiveStorage.exists("overviews/" + jobId + ".json")).isTrue(); assertThat(archiveStorage.exists("jobs/overview.json")).isTrue(); } + + @TestTemplate + void testScanArchivesWithoutFetch() throws Exception { + JobID jobId = JobID.generate(); + createJobArchive(remoteArchiveRootPath, jobId, true); + + HistoryServerArchiveFetcher<?> fetcher = + createArchiveFetcher(remoteArchiveRootPath, true, archiveStorage); + + fetcher.scanArchives(EAGER, false); + assertThat(archiveEvents).isEmpty(); + assertThat(archiveStorage.exists("overviews/" + jobId + ".json")).isFalse(); + } + + @TestTemplate + void testLazyFetchArchiveProactively() throws Exception { + // with explicit path + JobID jobId = JobID.generate(); + Path archivePath = createJobArchive(remoteArchiveRootPath, jobId, true); + + HistoryServerArchiveFetcher<?> fetcher = + createArchiveFetcher(remoteArchiveRootPath, false, archiveStorage); + + fetcher.lazyFetchArchiveProactively(jobId.toString(), archivePath); + waitForArchiveLoaded(archiveMetaInfoCache, jobId.toString()); + + assertThat(archiveMetaInfoCache.get(jobId.toString()).getEventType()).isEqualTo(CREATED); + assertThat(archiveStorage.exists("overviews/" + jobId + ".json")).isTrue(); + assertThat(archiveStorage.exists("jobs/" + jobId + ".json")).isTrue(); + assertThat(archiveStorage.exists("jobs/" + jobId + "/config.json")).isTrue(); + + // without explicit path + jobId = JobID.generate(); + createJobArchive(remoteArchiveRootPath, jobId, true); + + // call without explicit path; the fetcher should locate the archive in refreshDirs. + fetcher.lazyFetchArchiveProactively(jobId.toString(), null); + waitForArchiveLoaded(archiveMetaInfoCache, jobId.toString()); + + assertThat(archiveMetaInfoCache.get(jobId.toString()).getEventType()).isEqualTo(CREATED); + assertThat(archiveStorage.exists("overviews/" + jobId + ".json")).isTrue(); + assertThat(archiveStorage.exists("jobs/" + jobId + ".json")).isTrue(); + assertThat(archiveStorage.exists("jobs/" + jobId + "/config.json")).isTrue(); + } + + @TestTemplate + void testCleanUpLazyFetchTaskCancelsRunningFuture() throws Exception { + JobID jobId = JobID.generate(); + createJobArchive(remoteArchiveRootPath, jobId, true); + + HistoryServerTestUtils.BlockingArchiveStorage<Object> blockingStorage = + new HistoryServerTestUtils.BlockingArchiveStorage<>( + archiveStorage, "jobs/" + jobId + "/config"); + archiveStorage = blockingStorage; + + HistoryServerArchiveFetcher<?> fetcher = + createArchiveFetcher(remoteArchiveRootPath, false, blockingStorage); + fetcher.fetchArchives(LAZY); + // make sure the async detail task has started + assertThat(blockingStorage.asyncStartLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(fetcher.getCommonFetchTask(jobId.toString())).isNotNull(); + assertThat(fetcher.getCommonFetchTask(jobId.toString())).isNotDone(); + + // cancel the in-flight task + fetcher.cleanUpLazyFetchTask(jobId.toString()); + + assertThat(fetcher.getCommonFetchTask(jobId.toString())).isNull(); + assertThat(archiveStorage.exists("overviews/" + jobId + ".json")).isTrue(); + assertThat(archiveStorage.exists("jobs/" + jobId + ".json")).isTrue(); + assertThat(archiveStorage.exists("jobs/" + jobId + "/config.json")).isFalse(); + } } diff --git a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/utils/WebFrontendBootstrapTest.java b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/utils/WebFrontendBootstrapTest.java index 968dcf73a7d..6a4bc3f9715 100644 --- a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/utils/WebFrontendBootstrapTest.java +++ b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/utils/WebFrontendBootstrapTest.java @@ -24,6 +24,7 @@ import org.apache.flink.runtime.io.network.netty.InboundChannelHandlerFactory; import org.apache.flink.runtime.io.network.netty.Prio0InboundChannelHandlerFactory; import org.apache.flink.runtime.io.network.netty.Prio1InboundChannelHandlerFactory; import org.apache.flink.runtime.rest.handler.router.Router; +import org.apache.flink.runtime.webmonitor.history.FileArchiveStorage; import org.apache.flink.runtime.webmonitor.history.HistoryServerStaticFileServerHandler; import org.apache.flink.runtime.webmonitor.testutils.HttpUtils; import org.apache.flink.testutils.junit.extensions.ContextClassLoaderExtension; @@ -34,6 +35,7 @@ import org.junit.jupiter.api.extension.RegisterExtension; import org.junit.jupiter.api.io.TempDir; import org.slf4j.LoggerFactory; +import java.io.File; import java.nio.file.Files; import java.nio.file.Path; @@ -57,12 +59,16 @@ class WebFrontendBootstrapTest { @Test void testHandlersMustBeLoaded() throws Exception { Path webDir = Files.createDirectories(tmp.resolve("webDir")); + File webDirFile = webDir.toFile(); Configuration configuration = new Configuration(); configuration.set(Prio0InboundChannelHandlerFactory.REDIRECT_FROM_URL, "/nonExisting"); configuration.set(Prio0InboundChannelHandlerFactory.REDIRECT_TO_URL, "/index.html"); Router<?> router = new Router<>() - .addGet("/:*", new HistoryServerStaticFileServerHandler(webDir.toFile())); + .addGet( + "/:*", + new HistoryServerStaticFileServerHandler( + new FileArchiveStorage(webDirFile), webDirFile)); WebFrontendBootstrap webUI = new WebFrontendBootstrap( router,
