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 db8a7f74fd93cb1d28823a292d57207ffccb2f88 Author: Zihao Chen <[email protected]> AuthorDate: Mon Jul 6 11:17:05 2026 +0800 [FLINK-40097][historyserver] Lazily load archives to expose job overviews earlier --- .../generated/history_server_configuration.html | 12 + .../flink/configuration/HistoryServerOptions.java | 40 ++ .../webmonitor/history/ArchiveMetaInfo.java | 44 +++ .../runtime/webmonitor/history/HistoryServer.java | 48 ++- .../HistoryServerApplicationArchiveFetcher.java | 67 +++- .../history/HistoryServerArchiveFetcher.java | 188 ++++++++-- ...HistoryServerApplicationArchiveFetcherTest.java | 406 +++++++++++++++++++++ .../history/HistoryServerArchiveFetcherTest.java | 355 ++++++++++++++++++ .../webmonitor/history/HistoryServerTest.java | 104 +----- .../webmonitor/history/HistoryServerTestUtils.java | 295 +++++++++++++++ 10 files changed, 1422 insertions(+), 137 deletions(-) diff --git a/docs/layouts/shortcodes/generated/history_server_configuration.html b/docs/layouts/shortcodes/generated/history_server_configuration.html index 896c6813563..6048ea7d5ee 100644 --- a/docs/layouts/shortcodes/generated/history_server_configuration.html +++ b/docs/layouts/shortcodes/generated/history_server_configuration.html @@ -32,6 +32,12 @@ <td>Duration</td> <td>Interval for refreshing the archived job directories.</td> </tr> + <tr> + <td><h5>historyserver.archive.load.mode</h5></td> + <td style="word-wrap: break-word;">EAGER</td> + <td><p>Enum</p></td> + <td>The mode that HistoryServer loads archives.<br /><br />Possible values:<ul><li>"EAGER"</li><li>"LAZY"</li></ul></td> + </tr> <tr> <td><h5>historyserver.archive.retained-applications</h5></td> <td style="word-wrap: break-word;">-1</td> @@ -56,6 +62,12 @@ <td><p>Enum</p></td> <td>The type of archive storage.<br /><br />Possible values:<ul><li>"FILE"</li><li>"ROCKSDB"</li></ul></td> </tr> + <tr> + <td><h5>historyserver.lazy.fetch.executor.common.pool-size</h5></td> + <td style="word-wrap: break-word;">4</td> + <td>Integer</td> + <td>The size of the common 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 c90f92ccc3d..1e34bcc3656 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 @@ -257,6 +257,33 @@ public class HistoryServerOptions { .text("The type of archive storage.") .build()); + /** + * The mode that HistoryServer loads archives. + * + * <ul> + * <li>EAGER: Loads all archives by scheduled executor. + * <li>LAZY: Loads archives asynchronously only when requested. + * </ul> + */ + public static final ConfigOption<HistoryServerArchiveLoadMode> + HISTORY_SERVER_ARCHIVE_LOAD_MODE = + key("historyserver.archive.load.mode") + .enumType(HistoryServerArchiveLoadMode.class) + .defaultValue(HistoryServerArchiveLoadMode.EAGER) + .withDescription( + Description.builder() + .text("The mode that HistoryServer loads archives.") + .build()); + + public static final ConfigOption<Integer> HISTORY_SERVER_LAZY_FETCH_EXECUTOR_COMMON_POOL_SIZE = + key("historyserver.lazy.fetch.executor.common.pool-size") + .intType() + .defaultValue(4) + .withDescription( + Description.builder() + .text("The size of the common pool for archive fetching.") + .build()); + /** The type of archive storage. */ public enum HistoryServerArchiveStorageType { /** Local file system. */ @@ -266,5 +293,18 @@ public class HistoryServerOptions { ROCKSDB } + /** The mode that HistoryServer loads archives. */ + public enum HistoryServerArchiveLoadMode { + + /** + * Eager mode (default). Archive files will be downloaded and persisted with the default + * retention behavior. + */ + EAGER, + + /** Lazy mode. Archive files will be downloaded and persisted if necessary. */ + LAZY + } + private HistoryServerOptions() {} } 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 new file mode 100644 index 00000000000..710868803dd --- /dev/null +++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/ArchiveMetaInfo.java @@ -0,0 +1,44 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.runtime.webmonitor.history; + +/** Meta info for archived job. */ +public class ArchiveMetaInfo { + + private final String archiveId; + private volatile HistoryServerArchiveFetcher.ArchiveEventType eventType; + + public ArchiveMetaInfo( + String archiveId, HistoryServerArchiveFetcher.ArchiveEventType eventType) { + this.archiveId = archiveId; + this.eventType = eventType; + } + + public String getArchiveId() { + return archiveId; + } + + public HistoryServerArchiveFetcher.ArchiveEventType getEventType() { + return eventType; + } + + public void setEventType(HistoryServerArchiveFetcher.ArchiveEventType eventType) { + this.eventType = eventType; + } +} 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 ed59bf2db4c..d3e77c035da 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 @@ -68,6 +68,7 @@ import java.util.List; import java.util.UUID; import java.util.concurrent.Callable; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; @@ -75,6 +76,7 @@ import java.util.concurrent.TimeUnit; 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.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; @@ -141,6 +143,7 @@ public class HistoryServer { private final Thread shutdownHook; private final ArchiveStorage<?> archiveStorage; + private final HistoryServerOptions.HistoryServerArchiveLoadMode archiveLoadMode; private final AbstractHistoryServerHandler<?> historyServerHandler; public static void main(String[] args) throws Exception { @@ -250,10 +253,10 @@ public class HistoryServer { throw new FlinkException( "Failed to validate any of the configured directories to monitor."); } - refreshIntervalMillis = config.get(HistoryServerOptions.HISTORY_SERVER_ARCHIVE_REFRESH_INTERVAL).toMillis(); + archiveLoadMode = config.get(HistoryServerOptions.HISTORY_SERVER_ARCHIVE_LOAD_MODE); HistoryServerOptions.HistoryServerArchiveStorageType archiveStorageType = config.get(HistoryServerOptions.HISTORY_SERVER_ARCHIVE_STORAGE_TYPE); switch (archiveStorageType) { @@ -280,6 +283,11 @@ public class HistoryServer { throw new FlinkException("Unsupported archive storage type: " + archiveStorageType); } + ConcurrentHashMap<String, ArchiveMetaInfo> archiveMetaInfoCache = new ConcurrentHashMap<>(); + ConcurrentHashMap<String, ArchiveMetaInfo> applicationArchiveMetaInfoCache = + new ConcurrentHashMap<>(); + int lazyFetchExecutorCommonPoolSize = + config.get(HISTORY_SERVER_LAZY_FETCH_EXECUTOR_COMMON_POOL_SIZE); archiveFetcher = new HistoryServerArchiveFetcher<>( refreshDirs, @@ -287,7 +295,9 @@ public class HistoryServer { jobArchiveEventListener, cleanupExpiredJobs, CompositeArchiveRetainedStrategy.createForJobFromConfig(config), - archiveStorage); + archiveStorage, + archiveMetaInfoCache, + lazyFetchExecutorCommonPoolSize); applicationArchiveFetcher = new HistoryServerApplicationArchiveFetcher<>( refreshDirs, @@ -295,7 +305,10 @@ public class HistoryServer { applicationArchiveEventListener, cleanupExpiredApplications, CompositeArchiveRetainedStrategy.createForApplicationFromConfig(config), - archiveStorage); + archiveStorage, + archiveMetaInfoCache, + applicationArchiveMetaInfoCache, + lazyFetchExecutorCommonPoolSize); this.shutdownHook = ShutdownHookUtil.addShutdownHook( @@ -337,7 +350,7 @@ public class HistoryServer { @VisibleForTesting void fetchArchives() { - executor.execute(getArchiveFetchingRunnable()); + executor.execute(getArchiveFetchingRunnable(archiveLoadMode)); } public void run() { @@ -384,11 +397,13 @@ public class HistoryServer { CompletableFuture.completedFuture(pattern)))); createDashboardConfigFile(); - router.addGet("/:*", historyServerHandler); executor.scheduleWithFixedDelay( - getArchiveFetchingRunnable(), 0, refreshIntervalMillis, TimeUnit.MILLISECONDS); + getArchiveFetchingRunnable(archiveLoadMode), + 0, + refreshIntervalMillis, + TimeUnit.MILLISECONDS); netty = new WebFrontendBootstrap( @@ -396,11 +411,12 @@ public class HistoryServer { } } - private Runnable getArchiveFetchingRunnable() { + private Runnable getArchiveFetchingRunnable( + HistoryServerOptions.HistoryServerArchiveLoadMode archiveLoadMode) { return Runnables.withUncaughtExceptionHandler( () -> { - archiveFetcher.fetchArchives(); - applicationArchiveFetcher.fetchArchives(); + archiveFetcher.fetchArchives(archiveLoadMode); + applicationArchiveFetcher.fetchArchives(archiveLoadMode); }, FatalExitExceptionHandler.INSTANCE); } @@ -424,6 +440,18 @@ public class HistoryServer { LOG.warn("Error while closing archive storage.", t); } + try { + archiveFetcher.close(); + } catch (Throwable t) { + LOG.warn("Error while closing archive fetcher.", t); + } + + try { + applicationArchiveFetcher.close(); + } catch (Throwable t) { + LOG.warn("Error while closing application archive fetcher.", t); + } + try { LOG.info("Removing web dashboard root cache directory {}", webDir); FileUtils.deleteDirectory(webDir); @@ -466,7 +494,7 @@ public class HistoryServer { private final Path path; private final FileSystem fs; - private RefreshLocation(Path path, FileSystem fs) { + RefreshLocation(Path path, FileSystem fs) { this.path = path; this.fs = fs; } 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 10b8161b95b..f3b479aaf16 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 @@ -42,8 +42,13 @@ import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; import java.util.function.Consumer; +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.HistoryServerArchiveFetcher.ArchiveEventType.OVERVIEW_PARSING; + /** * This class is used by the {@link HistoryServer} to fetch the application and job archives that * are located at {@link HistoryServerOptions#HISTORY_SERVER_ARCHIVE_DIRS}. The directories are @@ -73,21 +78,29 @@ public class HistoryServerApplicationArchiveFetcher<Entry> private final Map<Path, Map<String, Set<String>>> cachedApplicationIdsToJobIds = new HashMap<>(); + private final ConcurrentHashMap<String, ArchiveMetaInfo> applicationArchiveMetaInfoCache; + HistoryServerApplicationArchiveFetcher( List<HistoryServer.RefreshLocation> refreshDirs, File webDir, - Consumer<HistoryServerApplicationArchiveFetcher.ArchiveEvent> archiveEventListener, + Consumer<ArchiveEvent> archiveEventListener, boolean cleanupExpiredArchives, ArchiveRetainedStrategy retainedStrategy, - ArchiveStorage<Entry> archiveStorage) { + ArchiveStorage<Entry> archiveStorage, + ConcurrentHashMap<String, ArchiveMetaInfo> archiveMetaInfoCache, + ConcurrentHashMap<String, ArchiveMetaInfo> applicationArchiveMetaInfoCache, + int lazyFetchExecutorCommonPoolSize) { super( refreshDirs, webDir, archiveEventListener, cleanupExpiredArchives, retainedStrategy, - archiveStorage); + archiveStorage, + archiveMetaInfoCache, + lazyFetchExecutorCommonPoolSize); + this.applicationArchiveMetaInfoCache = applicationArchiveMetaInfoCache; for (HistoryServer.RefreshLocation refreshDir : refreshDirs) { cachedApplicationIdsToJobIds.put(refreshDir.getPath(), new HashMap<>()); } @@ -143,7 +156,16 @@ public class HistoryServerApplicationArchiveFetcher<Entry> @Override List<ArchiveEvent> processArchive(String archiveId, Path archivePath, Path refreshDir) - throws IOException { + throws Exception { + return processArchive(archiveId, archivePath, refreshDir, EAGER); + } + + List<ArchiveEvent> processArchive( + String archiveId, + Path archivePath, + Path refreshDir, + HistoryServerOptions.HistoryServerArchiveLoadMode archiveLoadMode) + throws Exception { FileSystem fs = archivePath.getFileSystem(); Path applicationArchive = new Path(archivePath, ArchivePathUtils.APPLICATION_ARCHIVE_NAME); if (!fs.exists(applicationArchive)) { @@ -162,7 +184,11 @@ public class HistoryServerApplicationArchiveFetcher<Entry> .get(refreshDir) .computeIfAbsent(archiveId, k -> new HashSet<>()) .add(jobId); - events.add(processJobArchive(jobId, jobArchive.getPath())); + ArchiveEvent processArchiveEvents = + LAZY.equals(archiveLoadMode) + ? lazyProcessJobArchive(jobId, jobArchive.getPath()) + : processJobArchive(jobId, jobArchive.getPath()); + events.add(processArchiveEvents); } return events; @@ -234,6 +260,7 @@ public class HistoryServerApplicationArchiveFetcher<Entry> LOG.warn("Could not delete file from application directory.", ioe); } + applicationArchiveMetaInfoCache.remove(applicationId); return new ArchiveEvent(applicationId, ArchiveEventType.DELETED); } @@ -282,4 +309,34 @@ public class HistoryServerApplicationArchiveFetcher<Entry> LOG.error("Failed to update application overview.", e); } } + + @Override + List<ArchiveEvent> lazyProcessArchive(String archiveId, Path archivePath, Path refreshDir) + throws Exception { + List<ArchiveEvent> events = new ArrayList<>(); + ArchiveMetaInfo archiveMetaInfo = new ArchiveMetaInfo(archiveId, OVERVIEW_PARSING); + ArchiveMetaInfo existing = + applicationArchiveMetaInfoCache.putIfAbsent(archiveId, archiveMetaInfo); + if (existing != null) { + events.add(new ArchiveEvent(archiveId, existing.getEventType())); + return events; + } + + events.addAll(processArchive(archiveId, archivePath, refreshDir, LAZY)); + + archiveMetaInfo.setEventType(ArchiveEventType.CREATED); + return events; + } + + @Override + void cleanUpLazyFetchTask(String archiveId) { + for (HistoryServer.RefreshLocation refreshDir : refreshDirs) { + Path refreshDirPath = refreshDir.getPath(); + if (cachedApplicationIdsToJobIds.get(refreshDirPath).containsKey(archiveId)) { + Set<String> jobIds = + cachedApplicationIdsToJobIds.get(refreshDirPath).get(archiveId); + jobIds.forEach(super::cleanUpLazyFetchTask); + } + } + } } 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 e64114ef3da..2de41ed8091 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 @@ -30,6 +30,8 @@ import org.apache.flink.runtime.messages.webmonitor.JobDetails; import org.apache.flink.runtime.messages.webmonitor.MultipleJobsDetails; import org.apache.flink.runtime.rest.messages.JobsOverviewHeaders; import org.apache.flink.runtime.webmonitor.history.retaining.ArchiveRetainedStrategy; +import org.apache.flink.util.ExecutorUtils; +import org.apache.flink.util.concurrent.ExecutorThreadFactory; import org.apache.flink.util.jackson.JacksonMapperFactory; import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode; @@ -50,8 +52,15 @@ import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; import java.util.function.Consumer; +import static org.apache.flink.configuration.HistoryServerOptions.HistoryServerArchiveLoadMode.LAZY; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerArchiveFetcher.ArchiveEventType.PENDING; import static org.apache.flink.util.Preconditions.checkNotNull; /** @@ -66,10 +75,18 @@ import static org.apache.flink.util.Preconditions.checkNotNull; * * @param <Entry> the type of entries returned by the underlying {@link ArchiveStorage}. */ -public class HistoryServerArchiveFetcher<Entry> { +public class HistoryServerArchiveFetcher<Entry> implements AutoCloseable { /** Possible archive operations in history-server. */ public enum ArchiveEventType { + /** Archive is pending to be processed. */ + PENDING, + /** Overview content is currently parsing. */ + OVERVIEW_PARSING, + /** Overview content of archive was parsed and created in history server successfully. */ + OVERVIEW_CREATED, + /** Detail content of archive is currently parsing. */ + DETAIL_PARSING, /** Archive was found in one refresh location and created in history server. */ CREATED, /** Archive was deleted from one of refresh locations and deleted from history server. */ @@ -115,13 +132,21 @@ public class HistoryServerArchiveFetcher<Entry> { protected final ArchiveStorage<Entry> archiveStorage; + /** Executor for loading archives. */ + private final ExecutorService commonFetchExecutor; + + private final Map<String, Future<?>> commonFetchTasks; + private final ConcurrentHashMap<String, ArchiveMetaInfo> archiveMetaInfoCache; + HistoryServerArchiveFetcher( List<HistoryServer.RefreshLocation> refreshDirs, File webDir, Consumer<ArchiveEvent> archiveEventListener, boolean cleanupExpiredArchives, ArchiveRetainedStrategy retainedStrategy, - ArchiveStorage<Entry> archiveStorage) { + ArchiveStorage<Entry> archiveStorage, + ConcurrentHashMap<String, ArchiveMetaInfo> archiveMetaInfoCache, + int lazyFetchExecutorCommonPoolSize) { this.refreshDirs = checkNotNull(refreshDirs); this.archiveEventListener = archiveEventListener; this.processExpiredArchiveDeletion = cleanupExpiredArchives; @@ -132,6 +157,12 @@ public class HistoryServerArchiveFetcher<Entry> { } checkNotNull(webDir); this.archiveStorage = archiveStorage; + this.archiveMetaInfoCache = archiveMetaInfoCache; + this.commonFetchExecutor = + Executors.newFixedThreadPool( + lazyFetchExecutorCommonPoolSize, + new ExecutorThreadFactory("HistoryServer-commonFetchExecutor")); + this.commonFetchTasks = new ConcurrentHashMap<>(); updateJobOverview(); if (LOG.isInfoEnabled()) { @@ -141,9 +172,9 @@ public class HistoryServerArchiveFetcher<Entry> { } } - void fetchArchives() { + void fetchArchives(HistoryServerOptions.HistoryServerArchiveLoadMode archiveLoadMode) { + LOG.debug("Starting archive fetching."); try { - LOG.debug("Starting archive fetching."); List<ArchiveEvent> events = new ArrayList<>(); Map<Path, Set<String>> archivesToRemove = new HashMap<>(); cachedArchivesPerRefreshDirectory.forEach( @@ -180,28 +211,16 @@ public class HistoryServerArchiveFetcher<Entry> { continue; } - if (cachedArchivesPerRefreshDirectory.get(refreshDir).contains(archiveId)) { - LOG.trace( - "Ignoring archive {} because it was already fetched.", archivePath); - } else { - LOG.info("Processing archive {}.", archivePath); - try { - events.addAll(processArchive(archiveId, archivePath, refreshDir)); - cachedArchivesPerRefreshDirectory.get(refreshDir).add(archiveId); - LOG.info("Processing archive {} finished.", archivePath); - } catch (IOException e) { - LOG.error( - "Failure while fetching/processing archive {}.", archiveId, e); - deleteCachedArchives(archiveId, refreshDir); - } - } + fetchArchive(refreshDir, archiveId, archivePath, archiveLoadMode, events); } } + // clean local if (archivesToRemove.values().stream().flatMap(Set::stream).findAny().isPresent() && processExpiredArchiveDeletion) { events.addAll(cleanupExpiredArchives(archivesToRemove)); } + // clean remote and local if (!archivesBeyondRetainedLimit.isEmpty()) { events.addAll(cleanupArchivesBeyondRetainedLimit(archivesBeyondRetainedLimit)); } @@ -215,6 +234,32 @@ public class HistoryServerArchiveFetcher<Entry> { } } + private void fetchArchive( + Path refreshDir, + String archiveId, + Path archivePath, + HistoryServerOptions.HistoryServerArchiveLoadMode archiveLoadMode, + List<ArchiveEvent> events) + throws Exception { + if (cachedArchivesPerRefreshDirectory.get(refreshDir).contains(archiveId)) { + LOG.trace("Ignoring archive {} because it was already fetched.", archivePath); + } else { + LOG.info("Processing archive {}.", archivePath); + try { + List<ArchiveEvent> processArchiveEvents = + LAZY.equals(archiveLoadMode) + ? lazyProcessArchive(archiveId, archivePath, refreshDir) + : processArchive(archiveId, archivePath, refreshDir); + events.addAll(processArchiveEvents); + cachedArchivesPerRefreshDirectory.get(refreshDir).add(archiveId); + LOG.info("Processing archive {} finished.", archivePath); + } catch (Exception e) { + LOG.error("Failure while fetching/processing archive {}.", archiveId, e); + deleteCachedArchives(archiveId, refreshDir); + } + } + } + List<FileStatus> listValidArchives(FileSystem refreshFS, Path refreshDir) throws IOException { return listValidJobArchives(refreshFS, refreshDir); } @@ -255,7 +300,7 @@ public class HistoryServerArchiveFetcher<Entry> { } List<ArchiveEvent> processArchive(String archiveId, Path archivePath, Path refreshDir) - throws IOException { + throws Exception { return Collections.singletonList(processJobArchive(archiveId, archivePath)); } @@ -314,8 +359,10 @@ public class HistoryServerArchiveFetcher<Entry> { (refreshDir, archives) -> { cachedArchivesPerRefreshDirectory.get(refreshDir).removeAll(archives); archives.forEach( - archiveId -> - deleteLog.addAll(deleteCachedArchives(archiveId, refreshDir))); + archiveId -> { + cleanUpLazyFetchTask(archiveId); + deleteLog.addAll(deleteCachedArchives(archiveId, refreshDir)); + }); }); return deleteLog; @@ -348,6 +395,7 @@ public class HistoryServerArchiveFetcher<Entry> { LOG.warn("Could not delete file from job directory.", ioe); } + archiveMetaInfoCache.remove(jobId); return new ArchiveEvent(jobId, ArchiveEventType.DELETED); } @@ -458,4 +506,100 @@ public class HistoryServerArchiveFetcher<Entry> { LOG.error("Failed to update job overview.", e); } } + + // -------------------------------- Lazy Load ---------------------------------------- + List<ArchiveEvent> lazyProcessArchive(String archiveId, Path archivePath, Path refreshDir) + throws Exception { + return Collections.singletonList(lazyProcessJobArchive(archiveId, archivePath)); + } + + ArchiveEvent lazyProcessJobArchive(String jobId, Path jobArchive) throws Exception { + final ArchiveMetaInfo archiveMetaInfo = new ArchiveMetaInfo(jobId, PENDING); + ArchiveMetaInfo existing = archiveMetaInfoCache.putIfAbsent(jobId, archiveMetaInfo); + if (existing != null) { + return new ArchiveEvent(jobId, existing.getEventType()); + } + archiveMetaInfo.setEventType(ArchiveEventType.OVERVIEW_PARSING); + + Collection<ArchivedJson> archivedJsons = FsJsonArchivist.readArchivedJsons(jobArchive); + List<ArchivedJson> detailArchives = new ArrayList<>(); + boolean overviewCreated = false; + + for (ArchivedJson archive : archivedJsons) { + String path = archive.getPath(); + String json = archive.getJson(); + + if (path.equals(JobsOverviewHeaders.URL)) { + String key = JOB_OVERVIEWS_KEY_PREFIX + jobId + JSON_FILE_ENDING; + archiveStorage.putArchiveContent(key, json); + overviewCreated = true; + } else if (path.equals("/joboverview")) { // legacy path + LOG.debug("Migrating legacy archive {}", jobArchive); + json = convertLegacyJobOverview(json); + String key = JOB_OVERVIEWS_KEY_PREFIX + jobId + JSON_FILE_ENDING; + archiveStorage.putArchiveContent(key, json); + overviewCreated = true; + } else if (path.equals("/jobs/" + jobId)) { + String key = JOBS_KEY_PREFIX + jobId + JSON_FILE_ENDING; + archiveStorage.putArchiveContent(key, json); + } else { + detailArchives.add(archive); + } + } + + if (!overviewCreated && detailArchives.isEmpty()) { + archiveMetaInfoCache.remove(jobId); + throw new RuntimeException("Archive of job " + jobId + " is empty"); + } + + if (!detailArchives.isEmpty()) { + Future<?> future = + commonFetchExecutor.submit( + () -> { + try { + archiveMetaInfo.setEventType(ArchiveEventType.DETAIL_PARSING); + for (ArchivedJson archive : detailArchives) { + String path = archive.getPath(); + String json = archive.getJson(); + // this implicitly writes into webJobDir; strip the leading + // '/' from the + // REST path so that the key is a relative sub-path under + // the storage root + String key = path.substring(1) + JSON_FILE_ENDING; + try { + archiveStorage.putArchiveContent(key, json); + } catch (IOException e) { + LOG.error( + "Failed to write detail archive file for job {}, path {}.", + jobId, + path, + e); + } + } + archiveMetaInfo.setEventType(ArchiveEventType.CREATED); + LOG.debug("Async detail parsing for job {} finished.", jobId); + } finally { + commonFetchTasks.remove(jobId); + } + }); + commonFetchTasks.put(jobId, future); + } + + ArchiveEventType archiveEventType = + overviewCreated ? ArchiveEventType.OVERVIEW_CREATED : ArchiveEventType.CREATED; + archiveMetaInfo.setEventType(archiveEventType); + return new ArchiveEvent(jobId, archiveEventType); + } + + void cleanUpLazyFetchTask(String jobId) { + Future<?> commonFetchTask = commonFetchTasks.get(jobId); + if (commonFetchTask != null) { + commonFetchTask.cancel(true); + } + } + + @Override + public void close() { + ExecutorUtils.gracefulShutdown(1L, TimeUnit.SECONDS, commonFetchExecutor); + } } 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 new file mode 100644 index 00000000000..8775be5eb7a --- /dev/null +++ b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerApplicationArchiveFetcherTest.java @@ -0,0 +1,406 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.runtime.webmonitor.history; + +import org.apache.flink.api.common.ApplicationID; +import org.apache.flink.api.common.JobID; +import org.apache.flink.core.fs.Path; +import org.apache.flink.runtime.history.ArchivePathUtils; +import org.apache.flink.runtime.history.FsJsonArchivist; +import org.apache.flink.runtime.messages.webmonitor.ApplicationDetails; +import org.apache.flink.runtime.messages.webmonitor.MultipleApplicationsDetails; +import org.apache.flink.runtime.rest.messages.ApplicationsOverviewHeaders; +import org.apache.flink.testutils.junit.extensions.parameterized.Parameter; +import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension; +import org.apache.flink.testutils.junit.extensions.parameterized.Parameters; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.TestTemplate; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.api.io.TempDir; + +import java.io.File; +import java.nio.file.Files; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; + +import static org.apache.flink.configuration.ClusterOptions.CLUSTER_ID; +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.HistoryServerArchiveFetcher.ArchiveEventType.CREATED; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerArchiveFetcher.ArchiveEventType.DETAIL_PARSING; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerArchiveFetcher.ArchiveEventType.OVERVIEW_CREATED; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerTestUtils.OBJECT_MAPPER; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerTestUtils.RETAIN_ALL; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerTestUtils.createJobArchive; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerTestUtils.createRefreshLocation; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerTestUtils.waitForArchiveLoaded; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Tests for {@link HistoryServerApplicationArchiveFetcher}. Only application-specific behaviours + * that are NOT covered by {@link HistoryServerArchiveFetcherTest} are tested here. + */ +@ExtendWith(ParameterizedTestExtension.class) +class HistoryServerApplicationArchiveFetcherTest { + + private static final String DEFAULT_CLUSTER_ID = CLUSTER_ID.defaultValue(); + + @TempDir File remoteArchiveRootPath; + @TempDir File localArchiveRootPath; + + @Parameter public ArchiveStorageFactory<Object> storageFactory; + + private ArchiveStorage<Object> archiveStorage; + + private ConcurrentHashMap<String, ArchiveMetaInfo> jobMetaInfoCache; + private ConcurrentHashMap<String, ArchiveMetaInfo> applicationMetaInfoCache; + private List<HistoryServerArchiveFetcher.ArchiveEvent> archiveEvents; + + @Parameters(name = "storageFactory={0}") + private static Collection<ArchiveStorageFactory<?>> storageFactories() { + ArchiveStorageFactory<File> fileArchiveStorageFactory = FileArchiveStorage::new; + ArchiveStorageFactory<String> rocksDBStorageFactory = RocksDBArchiveStorage::new; + return List.of(fileArchiveStorageFactory, rocksDBStorageFactory); + } + + /** Creates an {@link ArchiveStorage} instance under the given temporary directory. */ + @FunctionalInterface + interface ArchiveStorageFactory<T> { + ArchiveStorage<T> create(File tempDir) throws Exception; + } + + @BeforeEach + void setUp() throws Exception { + jobMetaInfoCache = new ConcurrentHashMap<>(); + applicationMetaInfoCache = new ConcurrentHashMap<>(); + archiveEvents = new ArrayList<>(); + archiveStorage = storageFactory.create(localArchiveRootPath); + } + + @AfterEach + void tearDown() throws Exception { + archiveEvents.clear(); + if (archiveStorage != null) { + archiveStorage.close(); + archiveStorage = null; + } + } + + /** + * Create an application archive at {@code remoteArchiveRootPath/<CLUSTER_ID>/applications/ + * <applicationId>/} which contains an {@code application-summary} archive (representing the + * application overview) and one job archive per job in {@code jobs/}. + * + * <pre>{@code + * remoteArchiveRootPath/ + * ├── <cluster-id-1>/ + * │ └── applications/ + * │ ├── <application-id-1>/ + * │ │ ├── application-summary + * │ │ └── jobs/ + * │ │ ├── <job-id-1> + * │ │ └── ... + * │ └── ... + * }</pre> + * + * @param applicationId application id (must be a hex string parseable by {@link + * ApplicationID#fromHexString}) + * @param jobIds job ids whose archives are to be placed under the {@code jobs/} subdir + * @return the application archive directory path + */ + private Path createApplicationArchive(String applicationId, List<JobID> jobIds) + throws Exception { + File applicationDir = + new File( + remoteArchiveRootPath, + DEFAULT_CLUSTER_ID + + "/" + + ArchivePathUtils.APPLICATIONS_DIR + + "/" + + applicationId); + Files.createDirectories(applicationDir.toPath()); + + // Write application-summary archive (the application overview entry). + ApplicationID appId = ApplicationID.fromHexString(applicationId); + Map<String, Integer> jobInfo = new HashMap<>(); + jobInfo.put("FINISHED", jobIds.size()); + ApplicationDetails applicationDetails = + new ApplicationDetails(appId, "test-app", 0L, 1L, 1L, "FINISHED", jobInfo); + String applicationOverviewJson = + OBJECT_MAPPER.writeValueAsString( + new MultipleApplicationsDetails(Collections.singleton(applicationDetails))); + ArchivedJson applicationOverviewArchive = + new ArchivedJson(ApplicationsOverviewHeaders.URL, applicationOverviewJson); + + // mock a simple /applications/<applicationid>.json + String applicationJson = "{\"id\":\"" + applicationId + "\"}"; + List<ArchivedJson> archives = new ArrayList<>(); + archives.add(applicationOverviewArchive); + archives.add(new ArchivedJson("/applications/" + applicationId, applicationJson)); + + Path applicationSummaryPath = + new Path( + applicationDir.toURI().toString(), + ArchivePathUtils.APPLICATION_ARCHIVE_NAME); + FsJsonArchivist.writeArchivedJsons(applicationSummaryPath, archives); + + // Write job archives under jobs/ subdir. + File jobsDir = new File(applicationDir, ArchivePathUtils.JOBS_DIR); + Files.createDirectories(jobsDir.toPath()); + for (JobID jobId : jobIds) { + createJobArchive(jobsDir, jobId, true); + } + + return new Path(applicationDir.toURI().toString()); + } + + private HistoryServerApplicationArchiveFetcher<Object> createApplicationArchiveFetcher( + boolean cleanupExpired, ArchiveStorage<Object> storage) throws Exception { + List<HistoryServer.RefreshLocation> refreshDirs = + Collections.singletonList(createRefreshLocation(remoteArchiveRootPath)); + return new HistoryServerApplicationArchiveFetcher<>( + refreshDirs, + localArchiveRootPath, + event -> archiveEvents.add(event), + cleanupExpired, + RETAIN_ALL, + storage, + jobMetaInfoCache, + applicationMetaInfoCache, + 4); + } + + private static String newApplicationId() { + return new ApplicationID().toHexString(); + } + + // ========================================================================= + // EAGER MODE TESTS + // + // localArchiveRootPath/ + // ├── application-overviews/ + // │ └── application-id-1.json + // │ └── ... + // ├── applications/ + // │ └── overview.json + // │ └── application-id-1.json + // │ └── ... + // ├── overviews/ + // │ └── job-id-1.json + // │ └── ... + // ├── jobs/ + // │ └── overview.json + // │ └── job-id-1.json + // │ └── job-id-1/ + // │ ├── detail.json + // │ └── ... + // ========================================================================= + + @TestTemplate + void testEagerModeLoadsAllAppsAndJobsSync() throws Exception { + int numApps = 2; + int numJobsPerApp = 2; + List<String> appIds = new ArrayList<>(); + Map<String, List<JobID>> appToJobs = new HashMap<>(); + for (int i = 0; i < numApps; i++) { + String appId = newApplicationId(); + appIds.add(appId); + List<JobID> jobIds = new ArrayList<>(); + for (int j = 0; j < numJobsPerApp; j++) { + jobIds.add(JobID.generate()); + } + appToJobs.put(appId, jobIds); + createApplicationArchive(appId, jobIds); + } + + HistoryServerApplicationArchiveFetcher<Object> fetcher = + createApplicationArchiveFetcher(false, archiveStorage); + fetcher.fetchArchives(EAGER); + + // N (apps) + N*M (jobs) events, all CREATED + assertThat(archiveEvents).hasSize(numApps + numApps * numJobsPerApp); + for (HistoryServerArchiveFetcher.ArchiveEvent event : archiveEvents) { + assertThat(event.getType()).isEqualTo(CREATED); + } + + for (String appId : appIds) { + assertThat(archiveStorage.exists("application-overviews/" + appId + ".json")).isTrue(); + assertThat(archiveStorage.exists("applications/" + appId + ".json")).isTrue(); + for (JobID jobId : appToJobs.get(appId)) { + assertThat(archiveStorage.exists("overviews/" + jobId + ".json")).isTrue(); + assertThat(archiveStorage.exists("jobs/" + jobId + ".json")).isTrue(); + assertThat(archiveStorage.exists("jobs/" + jobId + "/config.json")).isTrue(); + } + } + assertThat(archiveStorage.exists("jobs/overview.json")).isTrue(); + assertThat(archiveStorage.exists("applications/overview.json")).isTrue(); + } + + // ========================================================================= + // LAZY MODE TESTS + // ========================================================================= + + @TestTemplate + void testLazyModeApplicationSyncJobDetailAsync() throws Exception { + String appId = newApplicationId(); + JobID jobId = JobID.generate(); + createApplicationArchive(appId, Collections.singletonList(jobId)); + + HistoryServerTestUtils.BlockingArchiveStorage<Object> blockingStorage = + new HistoryServerTestUtils.BlockingArchiveStorage<>( + archiveStorage, "jobs/" + jobId + "/config"); + // Replace the field so that close() in tearDown() releases the underlying storage too. + archiveStorage = blockingStorage; + + HistoryServerApplicationArchiveFetcher<Object> fetcher = + createApplicationArchiveFetcher(false, blockingStorage); + fetcher.fetchArchives(LAZY); + + // The application-level event is CREATED (application is loaded synchronously); + // the embedded job-level event is OVERVIEW_CREATED (detail is loaded asynchronously). + List<HistoryServerArchiveFetcher.ArchiveEventType> eventTypes = + archiveEvents.stream() + .map(HistoryServerArchiveFetcher.ArchiveEvent::getType) + .collect(Collectors.toList()); + assertThat(eventTypes).containsExactlyInAnyOrder(CREATED, OVERVIEW_CREATED); + + // localArchiveRootPath/ + // ├── application-overviews/ + // │ └── application-id-1.json + // │ └── ... + // ├── applications/ + // │ └── overview.json + // │ └── application-id-1.json + // │ └── ... + // ├── overviews/ + // │ └── job-id-1.json + // │ └── ... + // ├── jobs/ + // │ └── overview.json + // │ └── job-id-1.json + // │ └── job-id-1/ + // │ ├── detail.json + // │ └── ... + + // Phase 1: detail not yet written + boolean asyncReached = blockingStorage.asyncStartLatch.await(10, TimeUnit.SECONDS); + assertThat(asyncReached).isTrue(); + assertThat(blockingStorage.exists("jobs/" + jobId + "/config.json")).isFalse(); + // application-summary content has been written synchronously + assertThat(blockingStorage.exists("application-overviews/" + appId + ".json")).isTrue(); + assertThat(blockingStorage.exists("applications/" + appId + ".json")).isTrue(); + assertThat(blockingStorage.exists("applications/overview.json")).isTrue(); + assertThat(blockingStorage.exists("overviews/" + jobId + ".json")).isTrue(); + assertThat(blockingStorage.exists("jobs/" + jobId + ".json")).isTrue(); + assertThat(blockingStorage.exists("jobs/overview.json")).isTrue(); + + // Phase 2: release async task and wait for completion + blockingStorage.releaseLatch.countDown(); + waitForArchiveLoaded(jobMetaInfoCache, jobId.toString()); + assertThat(blockingStorage.exists("jobs/" + jobId + "/config.json")).isTrue(); + assertThat(jobMetaInfoCache.get(jobId.toString()).getEventType()).isEqualTo(CREATED); + } + + @TestTemplate + void testMissingApplicationSummaryFileThrows() throws Exception { + String appId = newApplicationId(); + // Create the directory layout WITHOUT writing the application-summary file. + File applicationDir = + new File( + remoteArchiveRootPath, + DEFAULT_CLUSTER_ID + "/" + ArchivePathUtils.APPLICATIONS_DIR + "/" + appId); + Files.createDirectories(new File(applicationDir, ArchivePathUtils.JOBS_DIR).toPath()); + + HistoryServerApplicationArchiveFetcher<Object> fetcher = + createApplicationArchiveFetcher(false, archiveStorage); + Path refreshPath = new Path(remoteArchiveRootPath.toURI().toString()); + Path applicationPath = new Path(applicationDir.toURI().toString()); + + assertThatThrownBy(() -> fetcher.processArchive(appId, applicationPath, refreshPath)) + .hasMessageContaining( + "Application archive " + + new Path( + applicationPath, ArchivePathUtils.APPLICATION_ARCHIVE_NAME) + + " does not exist."); + + // Cache should NOT contain a successful entry for this application. + assertThat(applicationMetaInfoCache.containsKey(appId)).isFalse(); + } + + // ========================================================================= + // ApplicationArchiveMetaInfoCache STATE TRANSITION TESTS + // ========================================================================= + + @TestTemplate + void testApplicationArchiveStateTransition() throws Exception { + String appId = newApplicationId(); + JobID jobId = JobID.generate(); + Path applicationArchivePath = + createApplicationArchive(appId, Collections.singletonList(jobId)); + + HistoryServerTestUtils.BlockingArchiveStorage<Object> blockingStorage = + new HistoryServerTestUtils.BlockingArchiveStorage<>( + archiveStorage, "jobs/" + jobId + "/config"); + archiveStorage = blockingStorage; + + HistoryServerApplicationArchiveFetcher<Object> fetcher = + createApplicationArchiveFetcher(false, blockingStorage); + Path refreshPath = new Path(remoteArchiveRootPath.toURI().toString()); + + fetcher.fetchArchives(LAZY); + + // Phase 1: application is loaded synchronously -> CREATED; + assertThat(applicationMetaInfoCache.get(appId).getEventType()).isEqualTo(CREATED); + + // the embedded job's detail loading is blocked at the storage -> DETAIL_PARSING. + boolean asyncReached = blockingStorage.asyncStartLatch.await(10, TimeUnit.SECONDS); + assertThat(asyncReached).isTrue(); + assertThat(jobMetaInfoCache.get(jobId.toString()).getEventType()).isEqualTo(DETAIL_PARSING); + + // try to call lazyProcessJobArchive again, should return CREATED. + List<HistoryServerArchiveFetcher.ArchiveEvent> callAgainEvent = + fetcher.lazyProcessArchive(appId, applicationArchivePath, refreshPath); + assertThat(callAgainEvent.get(0).getType()).isEqualTo(CREATED); + assertThat(applicationMetaInfoCache.get(appId).getEventType()).isEqualTo(CREATED); + + // Phase 2: execute the asynchronous task + blockingStorage.releaseLatch.countDown(); + waitForArchiveLoaded(jobMetaInfoCache, jobId.toString()); + assertThat(jobMetaInfoCache.get(jobId.toString()).getEventType()).isEqualTo(CREATED); + + // Phase 3: cleanup all archives + Map<Path, Set<String>> archivesToRemove = new HashMap<>(); + archivesToRemove.put(refreshPath, Collections.singleton(appId)); + fetcher.cleanupExpiredArchives(archivesToRemove); + + assertThat(applicationMetaInfoCache.containsKey(appId)).isFalse(); + assertThat(jobMetaInfoCache.containsKey(jobId.toString())).isFalse(); + } +} 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 new file mode 100644 index 00000000000..ff929837fd9 --- /dev/null +++ b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcherTest.java @@ -0,0 +1,355 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.runtime.webmonitor.history; + +import org.apache.flink.api.common.JobID; +import org.apache.flink.core.fs.Path; +import org.apache.flink.runtime.history.FsJsonArchivist; +import org.apache.flink.runtime.messages.webmonitor.JobDetails; +import org.apache.flink.runtime.messages.webmonitor.MultipleJobsDetails; +import org.apache.flink.testutils.junit.extensions.parameterized.Parameter; +import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension; +import org.apache.flink.testutils.junit.extensions.parameterized.Parameters; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.TestTemplate; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.api.io.TempDir; + +import java.io.File; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; + +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.HistoryServerArchiveFetcher.ArchiveEventType.CREATED; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerArchiveFetcher.ArchiveEventType.DETAIL_PARSING; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerArchiveFetcher.ArchiveEventType.OVERVIEW_CREATED; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerTestUtils.OBJECT_MAPPER; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerTestUtils.RETAIN_ALL; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerTestUtils.createJobArchive; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerTestUtils.createLegacyArchive; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerTestUtils.createRefreshLocation; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerTestUtils.waitForArchiveLoaded; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for {@link HistoryServerArchiveFetcher}. */ +@ExtendWith(ParameterizedTestExtension.class) +class HistoryServerArchiveFetcherTest { + + @TempDir File remoteArchiveRootPath; + @TempDir File localArchiveRootPath; + + @Parameter public ArchiveStorageFactory<Object> storageFactory; + + private ArchiveStorage<Object> archiveStorage; + private ConcurrentHashMap<String, ArchiveMetaInfo> archiveMetaInfoCache; + private List<HistoryServerArchiveFetcher.ArchiveEvent> archiveEvents; + + @Parameters(name = "storageFactory={0}") + private static Collection<ArchiveStorageFactory<?>> storageFactories() { + ArchiveStorageFactory<File> fileArchiveStorageFactory = FileArchiveStorage::new; + ArchiveStorageFactory<String> rocksDBStorageFactory = RocksDBArchiveStorage::new; + return List.of(fileArchiveStorageFactory, rocksDBStorageFactory); + } + + /** Creates an {@link ArchiveStorage} instance under the given temporary directory. */ + @FunctionalInterface + interface ArchiveStorageFactory<T> { + ArchiveStorage<T> create(File tempDir) throws Exception; + } + + @BeforeEach + void setUp() throws Exception { + archiveMetaInfoCache = new ConcurrentHashMap<>(); + archiveEvents = new ArrayList<>(); + archiveStorage = storageFactory.create(localArchiveRootPath); + } + + @AfterEach + void tearDown() throws Exception { + archiveEvents.clear(); + if (archiveStorage != null) { + archiveStorage.close(); + archiveStorage = null; + } + } + + /** + * Create {@link HistoryServerArchiveFetcher} instance with a custom {@link ArchiveStorage}. + * + * @param refreshDir archive scan directory + * @param cleanupExpiredJobs whether to enable expired archive cleanup + * @param storage the archive storage to use + */ + private HistoryServerArchiveFetcher<?> createArchiveFetcher( + File refreshDir, boolean cleanupExpiredJobs, ArchiveStorage<?> storage) + throws Exception { + List<HistoryServer.RefreshLocation> refreshDirs = + Collections.singletonList(createRefreshLocation(refreshDir)); + return new HistoryServerArchiveFetcher<>( + refreshDirs, + localArchiveRootPath, + event -> archiveEvents.add(event), + cleanupExpiredJobs, + RETAIN_ALL, + storage, + archiveMetaInfoCache, + 4); + } + + // ========================================================================= + // EAGER MODE TESTS + // ========================================================================= + + @TestTemplate + void testEagerModeLoadsAllFilesSync() throws Exception { + int numJobs = 3; + List<JobID> jobIds = new ArrayList<>(); + for (int i = 0; i < numJobs; i++) { + JobID jobId = JobID.generate(); + jobIds.add(jobId); + createJobArchive(remoteArchiveRootPath, jobId, true); + } + + HistoryServerArchiveFetcher<?> fetcher = + createArchiveFetcher(remoteArchiveRootPath, false, archiveStorage); + fetcher.fetchArchives(EAGER); + + assertThat(archiveEvents).hasSize(numJobs); + for (HistoryServerArchiveFetcher.ArchiveEvent event : archiveEvents) { + assertThat(event.getType()).isEqualTo(CREATED); + } + for (JobID jobId : jobIds) { + assertThat(archiveStorage.exists("overviews/" + jobId + ".json")).isTrue(); + assertThat(archiveStorage.exists("jobs/" + jobId + ".json")).isTrue(); + assertThat(archiveStorage.exists("jobs/" + jobId + "/config.json")).isTrue(); + } + assertThat(archiveStorage.exists("jobs/overview.json")).isTrue(); + } + + // ========================================================================= + // LAZY MODE TESTS + // ========================================================================= + + @TestTemplate + void testLazyModeDetailNotWrittenBeforeAsyncTaskCompleted() throws Exception { + JobID jobId = JobID.generate(); + createJobArchive(remoteArchiveRootPath, jobId, true); + + HistoryServerTestUtils.BlockingArchiveStorage<Object> blockingStorage = + new HistoryServerTestUtils.BlockingArchiveStorage<>( + archiveStorage, "jobs/" + jobId + "/config"); + // Replace the field so that close() in tearDown() releases the underlying storage too. + archiveStorage = blockingStorage; + + HistoryServerArchiveFetcher<?> fetcher = + createArchiveFetcher(remoteArchiveRootPath, false, blockingStorage); + + fetcher.fetchArchives(LAZY); + + // Phase 1: overview archive is loaded synchronously, but detail archive haven't loaded yet. + boolean asyncReached = blockingStorage.asyncStartLatch.await(10, TimeUnit.SECONDS); + assertThat(asyncReached).isTrue(); + assertThat(archiveEvents.get(0).getType()).isEqualTo(OVERVIEW_CREATED); + assertThat(blockingStorage.exists("overviews/" + jobId + ".json")).isTrue(); + assertThat(blockingStorage.exists("jobs/" + jobId + ".json")).isTrue(); + assertThat(blockingStorage.exists("jobs/overview.json")).isTrue(); + assertThat(blockingStorage.exists("jobs/" + jobId + "/config.json")).isFalse(); + + // Phase 2: execute the asynchronous task + blockingStorage.releaseLatch.countDown(); + waitForArchiveLoaded(archiveMetaInfoCache, jobId.toString()); + + assertThat(archiveMetaInfoCache.get(jobId.toString()).getEventType()).isEqualTo(CREATED); + assertThat(blockingStorage.exists("jobs/" + jobId + "/config.json")).isTrue(); + } + + @TestTemplate + void testLazyModeLoadsAllFilesCompleted() throws Exception { + int numJobs = 5; + List<JobID> jobIds = new ArrayList<>(); + for (int i = 0; i < numJobs; i++) { + JobID jobId = JobID.generate(); + jobIds.add(jobId); + createJobArchive(remoteArchiveRootPath, jobId, true); + } + + HistoryServerArchiveFetcher<?> fetcher = + createArchiveFetcher(remoteArchiveRootPath, false, archiveStorage); + fetcher.fetchArchives(LAZY); + + assertThat(archiveEvents).hasSize(numJobs); + for (HistoryServerArchiveFetcher.ArchiveEvent event : archiveEvents) { + assertThat(event.getType()).isEqualTo(OVERVIEW_CREATED); + } + + for (JobID jobId : jobIds) { + waitForArchiveLoaded(archiveMetaInfoCache, jobId.toString()); + } + + // when async task completed, all detail archives should be loaded. + for (JobID jobId : jobIds) { + assertThat(archiveMetaInfoCache.get(jobId.toString()).getEventType()) + .isEqualTo(CREATED); + assertThat(archiveStorage.exists("jobs/" + jobId + "/config.json")).isTrue(); + } + } + + @TestTemplate + void testLazyProcessJobArchiveThrowsForEmptyArchive() throws Exception { + JobID jobId = JobID.generate(); + Path archivePath = new Path(remoteArchiveRootPath.toURI().toString(), jobId.toString()); + FsJsonArchivist.writeArchivedJsons(archivePath, Collections.emptyList()); + + HistoryServerArchiveFetcher<?> fetcher = + createArchiveFetcher(remoteArchiveRootPath, false, archiveStorage); + + assertThatThrownBy(() -> fetcher.lazyProcessJobArchive(jobId.toString(), archivePath)) + .isInstanceOf(RuntimeException.class) + .hasMessage("Archive of job " + jobId + " is empty"); + + assertThat(archiveMetaInfoCache.containsKey(jobId.toString())).isFalse(); + } + + // ========================================================================= + // ArchiveMetaInfoCache STATE TRANSITION TESTS + // ========================================================================= + + /** STATE: PENDING -> OVERVIEW_PARSING -> OVERVIEW_CREATED -> DETAIL_PARSING -> CREATED. */ + @TestTemplate + void testArchiveStateTransition() throws Exception { + JobID jobIdWithDetail = JobID.generate(); + Path jobWithDetailArchivePath = + createJobArchive(remoteArchiveRootPath, jobIdWithDetail, true); + JobID jobIdWithoutDetail = JobID.generate(); + createJobArchive(remoteArchiveRootPath, jobIdWithoutDetail, false); + + HistoryServerTestUtils.BlockingArchiveStorage<Object> blockingStorage = + new HistoryServerTestUtils.BlockingArchiveStorage<>( + archiveStorage, "jobs/" + jobIdWithDetail + "/config"); + archiveStorage = blockingStorage; + + HistoryServerArchiveFetcher<?> fetcher = + createArchiveFetcher(remoteArchiveRootPath, false, blockingStorage); + + fetcher.fetchArchives(LAZY); + + assertThat(archiveMetaInfoCache.size()).isEqualTo(2); + assertThat(archiveMetaInfoCache.containsKey(jobIdWithoutDetail.toString())).isTrue(); + assertThat(archiveMetaInfoCache.get(jobIdWithoutDetail.toString()).getEventType()) + .isEqualTo(OVERVIEW_CREATED); + + assertThat(archiveMetaInfoCache.containsKey(jobIdWithDetail.toString())).isTrue(); + // Phase 1: overview archive is loaded synchronously, but detail archive haven't loaded yet. + boolean asyncReached = blockingStorage.asyncStartLatch.await(10, TimeUnit.SECONDS); + assertThat(asyncReached).isTrue(); + assertThat(archiveMetaInfoCache.get(jobIdWithDetail.toString()).getEventType()) + .isEqualTo(DETAIL_PARSING); + + // try to call lazyProcessJobArchive again, should return DETAIL_PARSING. + HistoryServerArchiveFetcher.ArchiveEvent callAgainEvent = + fetcher.lazyProcessJobArchive(jobIdWithDetail.toString(), jobWithDetailArchivePath); + assertThat(callAgainEvent.getType()).isEqualTo(DETAIL_PARSING); + + // Phase 2: execute the asynchronous task + blockingStorage.releaseLatch.countDown(); + waitForArchiveLoaded(archiveMetaInfoCache, jobIdWithDetail.toString()); + assertThat(archiveMetaInfoCache.get(jobIdWithDetail.toString()).getEventType()) + .isEqualTo(CREATED); + + // Phase 3: remove all archives + Map<Path, Set<String>> archivesToRemove = new HashMap<>(); + Set<String> archiveIdsToRemove = new HashSet<>(); + archiveIdsToRemove.add(jobIdWithDetail.toString()); + archiveIdsToRemove.add(jobIdWithoutDetail.toString()); + archivesToRemove.put( + new Path(remoteArchiveRootPath.toURI().toString()), archiveIdsToRemove); + fetcher.cleanupExpiredArchives(archivesToRemove); + assertThat(archiveMetaInfoCache.size()).isEqualTo(0); + } + + // ========================================================================= + // Other Tests + // ========================================================================= + + /** + * This test reads {@code jobs/overview.json} directly from the local file system, so it only + * applies to the {@link FileArchiveStorage} backend. + */ + @TestTemplate + void testUpdateJobOverview() throws Exception { + JobID job1 = JobID.generate(); + JobID job2 = JobID.generate(); + createJobArchive(remoteArchiveRootPath, job1, true); + createJobArchive(remoteArchiveRootPath, job2, true); + + HistoryServerArchiveFetcher<?> fetcher = + createArchiveFetcher(remoteArchiveRootPath, true, archiveStorage); + fetcher.fetchArchives(EAGER); + + Object overviewObject = archiveStorage.getEntry("jobs/overview.json"); + String overviewContent = archiveStorage.readArchiveContent(overviewObject); + MultipleJobsDetails overview = + OBJECT_MAPPER.readValue(overviewContent, MultipleJobsDetails.class); + + assertThat(overview.getJobs()).hasSize(2); + Set<String> jobIds = new HashSet<>(); + for (JobDetails jobDetails : overview.getJobs()) { + jobIds.add(jobDetails.getJobId().toString()); + } + assertThat(jobIds).containsExactlyInAnyOrder(job1.toString(), job2.toString()); + + // remove job1 + new File(remoteArchiveRootPath, job1.toString()).delete(); + fetcher.fetchArchives(EAGER); + + // verify overview now only contains job2 + overviewObject = archiveStorage.getEntry("jobs/overview.json"); + overviewContent = archiveStorage.readArchiveContent(overviewObject); + overview = OBJECT_MAPPER.readValue(overviewContent, MultipleJobsDetails.class); + assertThat(overview.getJobs()).hasSize(1); + assertThat(overview.getJobs().iterator().next().getJobId()).isEqualTo(job2); + } + + @TestTemplate + void testLegacyJobOverviewMigration() throws Exception { + JobID jobId = createLegacyArchive(remoteArchiveRootPath.toPath(), false); + + HistoryServerArchiveFetcher<?> fetcher = + createArchiveFetcher(remoteArchiveRootPath, false, archiveStorage); + fetcher.fetchArchives(LAZY); + + assertThat(archiveEvents).hasSize(1); + assertThat(archiveEvents.get(0).getType()).isEqualTo(OVERVIEW_CREATED); + + assertThat(archiveStorage.exists("overviews/" + jobId + ".json")).isTrue(); + assertThat(archiveStorage.exists("jobs/overview.json")).isTrue(); + } +} diff --git a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerTest.java b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerTest.java index 4c7cd990d23..26fcceedf45 100644 --- a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerTest.java +++ b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerTest.java @@ -52,8 +52,6 @@ import org.apache.flink.test.util.MiniClusterWithClientResource; import org.apache.flink.util.FlinkException; import org.apache.flink.util.jackson.JacksonMapperFactory; -import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonFactory; -import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator; import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.DeserializationFeature; import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper; @@ -68,7 +66,6 @@ import javax.annotation.Nullable; import java.io.File; import java.io.IOException; -import java.io.StringWriter; import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; @@ -87,16 +84,14 @@ import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; import java.util.stream.Stream; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerTestUtils.createLegacyArchive; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerTestUtils.createLegacyArchiveWithModifiedDate; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Tests for the HistoryServer. */ class HistoryServerTest { - private static final JsonFactory JACKSON_FACTORY = - new JsonFactory() - .enable(JsonGenerator.Feature.AUTO_CLOSE_TARGET) - .disable(JsonGenerator.Feature.AUTO_CLOSE_JSON_CONTENT); private static final ObjectMapper OBJECT_MAPPER = JacksonMapperFactory.createObjectMapper() .enable(DeserializationFeature.FAIL_ON_MISSING_CREATOR_PROPERTIES); @@ -198,7 +193,7 @@ class HistoryServerTest { for (int j = 0; j < numArchivesBeforeHsStarted; j++) { JobID jobId = - createLegacyArchive( + createLegacyArchiveWithModifiedDate( jmDirectory.toPath(), j * oneMinuteSinceEpoch, versionLessThan14); if (j >= numArchivesToRemoveUponHsStart) { expectedJobIdsToKeep.add(jobId); @@ -253,7 +248,7 @@ class HistoryServerTest { j++) { expectedJobIdsToKeep.remove(0); expectedJobIdsToKeep.add( - createLegacyArchive( + createLegacyArchiveWithModifiedDate( jmDirectory.toPath(), j * oneMinuteSinceEpoch, versionLessThan14)); } assertThat(numArchivesCreatedTotal.await(10L, TimeUnit.SECONDS)).isTrue(); @@ -534,62 +529,6 @@ class HistoryServerTest { env.execute(); } - private static JobID createLegacyArchive( - Path directory, long fileModifiedDate, boolean versionLessThan14) throws IOException { - JobID jobId = createLegacyArchive(directory, versionLessThan14); - File jobArchive = directory.resolve(jobId.toString()).toFile(); - jobArchive.setLastModified(fileModifiedDate); - return jobId; - } - - private static JobID createLegacyArchive(Path directory, boolean versionLessThan14) - throws IOException { - JobID jobId = JobID.generate(); - - StringWriter sw = new StringWriter(); - try (JsonGenerator gen = JACKSON_FACTORY.createGenerator(sw)) { - try (JsonObject root = new JsonObject(gen)) { - try (JsonArray finished = new JsonArray(gen, "finished")) { - try (JsonObject job = new JsonObject(gen)) { - gen.writeStringField("jid", jobId.toString()); - gen.writeStringField("name", "testjob"); - gen.writeStringField("state", JobStatus.FINISHED.name()); - - gen.writeNumberField("start-time", 0L); - gen.writeNumberField("end-time", 1L); - gen.writeNumberField("duration", 1L); - gen.writeNumberField("last-modification", 1L); - - try (JsonObject tasks = new JsonObject(gen, "tasks")) { - gen.writeNumberField("total", 0); - - if (versionLessThan14) { - gen.writeNumberField("pending", 0); - } else { - gen.writeNumberField("created", 0); - gen.writeNumberField("deploying", 0); - gen.writeNumberField("scheduled", 0); - } - gen.writeNumberField("running", 0); - gen.writeNumberField("finished", 0); - gen.writeNumberField("canceling", 0); - gen.writeNumberField("canceled", 0); - gen.writeNumberField("failed", 0); - } - } - } - } - } - String json = sw.toString(); - ArchivedJson archivedJson = new ArchivedJson("/joboverview", json); - FsJsonArchivist.writeArchivedJsons( - new org.apache.flink.core.fs.Path( - directory.toAbsolutePath().toString(), jobId.toString()), - Collections.singleton(archivedJson)); - - return jobId; - } - @Test void testApplicationAndJobArchives() throws Exception { int numApplications = 2; @@ -991,39 +930,4 @@ class HistoryServerTest { .getParent(); applicationArchiveDir.getFileSystem().delete(applicationArchiveDir, true); } - - private static final class JsonObject implements AutoCloseable { - - private final JsonGenerator gen; - - JsonObject(JsonGenerator gen) throws IOException { - this.gen = gen; - gen.writeStartObject(); - } - - private JsonObject(JsonGenerator gen, String name) throws IOException { - this.gen = gen; - gen.writeObjectFieldStart(name); - } - - @Override - public void close() throws IOException { - gen.writeEndObject(); - } - } - - private static final class JsonArray implements AutoCloseable { - - private final JsonGenerator gen; - - JsonArray(JsonGenerator gen, String name) throws IOException { - this.gen = gen; - gen.writeArrayFieldStart(name); - } - - @Override - public void close() throws IOException { - gen.writeEndArray(); - } - } } diff --git a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerTestUtils.java b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerTestUtils.java new file mode 100644 index 00000000000..de42e501a71 --- /dev/null +++ b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerTestUtils.java @@ -0,0 +1,295 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.runtime.webmonitor.history; + +import org.apache.flink.api.common.JobID; +import org.apache.flink.api.common.JobStatus; +import org.apache.flink.core.fs.FileSystem; +import org.apache.flink.core.fs.Path; +import org.apache.flink.runtime.execution.ExecutionState; +import org.apache.flink.runtime.history.FsJsonArchivist; +import org.apache.flink.runtime.messages.webmonitor.JobDetails; +import org.apache.flink.runtime.messages.webmonitor.MultipleJobsDetails; +import org.apache.flink.runtime.rest.messages.JobsOverviewHeaders; +import org.apache.flink.runtime.webmonitor.history.retaining.ArchiveRetainedStrategy; +import org.apache.flink.util.jackson.JacksonMapperFactory; + +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonFactory; +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator; +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper; + +import javax.annotation.Nullable; + +import java.io.File; +import java.io.IOException; +import java.io.StringWriter; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import static org.apache.flink.runtime.webmonitor.history.HistoryServerArchiveFetcher.ArchiveEventType.CREATED; + +/** Common utilities for history server tests. */ +public class HistoryServerTestUtils { + + private static final JsonFactory JACKSON_FACTORY = + new JsonFactory() + .enable(JsonGenerator.Feature.AUTO_CLOSE_TARGET) + .disable(JsonGenerator.Feature.AUTO_CLOSE_JSON_CONTENT); + + public static final ObjectMapper OBJECT_MAPPER = JacksonMapperFactory.createObjectMapper(); + public static final ArchiveRetainedStrategy RETAIN_ALL = (file, index) -> true; + + static HistoryServer.RefreshLocation createRefreshLocation(File dir) throws Exception { + Path path = new Path(dir.toURI().toString()); + FileSystem fs = path.getFileSystem(); + return new HistoryServer.RefreshLocation(path, fs); + } + + /** + * Create a job archive at {@code baseDir/<jobId>}. + * + * @param baseDir base directory under which the archive will be written + * @param jobId job id used as the archive file name + * @param withDetail whether to include the {@code /jobs/<jobId>/config} sub-archive + * @return the archive {@link Path} + */ + static Path createJobArchive(File baseDir, JobID jobId, boolean withDetail) throws Exception { + MultipleJobsDetails overview = + new MultipleJobsDetails( + Collections.singleton( + new JobDetails( + jobId, + "test-job", + 0L, + 1L, + 1L, + JobStatus.FINISHED, + 1L, + new int[ExecutionState.values().length], + 0))); + String overviewJson = OBJECT_MAPPER.writeValueAsString(overview); + ArchivedJson overviewArchive = new ArchivedJson(JobsOverviewHeaders.URL, overviewJson); + + // mock a simple /jobs/<jobid>.json + String jobJson = "{\"jid\":\"" + jobId + "\"}"; + + List<ArchivedJson> archives = new ArrayList<>(); + archives.add(overviewArchive); + archives.add(new ArchivedJson("/jobs/" + jobId, jobJson)); + + if (withDetail) { + String detailJson = "{\"jid\":\"" + jobId + "\",\"name\":\"test-job\"}"; + archives.add(new ArchivedJson("/jobs/" + jobId + "/config", detailJson)); + } + + Path archivePath = new Path(baseDir.toURI().toString(), jobId.toString()); + FsJsonArchivist.writeArchivedJsons(archivePath, archives); + return archivePath; + } + + /** Create a legacy job archive at {@code directory/<jobId>} with only {@code /joboverview}. */ + public static JobID createLegacyArchive(java.nio.file.Path directory, boolean versionLessThan14) + throws IOException { + JobID jobId = JobID.generate(); + + StringWriter sw = new StringWriter(); + try (JsonGenerator gen = JACKSON_FACTORY.createGenerator(sw)) { + try (JsonObject root = new JsonObject(gen)) { + try (JsonArray finished = new JsonArray(gen, "finished")) { + try (JsonObject job = new JsonObject(gen)) { + gen.writeStringField("jid", jobId.toString()); + gen.writeStringField("name", "testjob"); + gen.writeStringField("state", JobStatus.FINISHED.name()); + + gen.writeNumberField("start-time", 0L); + gen.writeNumberField("end-time", 1L); + gen.writeNumberField("duration", 1L); + gen.writeNumberField("last-modification", 1L); + + try (JsonObject tasks = new JsonObject(gen, "tasks")) { + gen.writeNumberField("total", 0); + + if (versionLessThan14) { + gen.writeNumberField("pending", 0); + } else { + gen.writeNumberField("created", 0); + gen.writeNumberField("deploying", 0); + gen.writeNumberField("scheduled", 0); + } + gen.writeNumberField("running", 0); + gen.writeNumberField("finished", 0); + gen.writeNumberField("canceling", 0); + gen.writeNumberField("canceled", 0); + gen.writeNumberField("failed", 0); + } + } + } + } + } + String json = sw.toString(); + ArchivedJson archivedJson = new ArchivedJson("/joboverview", json); + FsJsonArchivist.writeArchivedJsons( + new Path(directory.toAbsolutePath().toString(), jobId.toString()), + Collections.singleton(archivedJson)); + + return jobId; + } + + public static JobID createLegacyArchiveWithModifiedDate( + java.nio.file.Path directory, long fileModifiedDate, boolean versionLessThan14) + throws IOException { + JobID jobId = createLegacyArchive(directory, versionLessThan14); + File jobArchive = directory.resolve(jobId.toString()).toFile(); + jobArchive.setLastModified(fileModifiedDate); + return jobId; + } + + /** + * Wait until {@code cache.get(archiveId).eventType == CREATED}. Throws {@link AssertionError} + * if the deadline is exceeded. + */ + static void waitForArchiveLoaded(Map<String, ArchiveMetaInfo> cache, String archiveId) + throws InterruptedException { + long deadline = System.currentTimeMillis() + 10_000; + while (System.currentTimeMillis() < deadline) { + ArchiveMetaInfo metaInfo = cache.get(archiveId); + if (metaInfo != null && metaInfo.getEventType() == CREATED) { + return; + } + Thread.sleep(50); + } + throw new AssertionError( + "Timed out waiting for archive " + archiveId + " to reach CREATED state"); + } + + /** + * Blocking archive storage for testing the LAZY load mode. Calls to {@link + * #putArchiveContent(String, String)} whose key contains {@code blockOnKeyPrefix} will block + * until {@link #releaseLatch} is counted down. + * + * <p>This is a decorator that delegates to a wrapped {@link ArchiveStorage}, so it can be used + * with any storage backend (e.g. {@link FileArchiveStorage} or {@link RocksDBArchiveStorage}). + */ + public static class BlockingArchiveStorage<Entry> implements ArchiveStorage<Entry> { + + public final CountDownLatch asyncStartLatch = new CountDownLatch(1); + public final CountDownLatch releaseLatch = new CountDownLatch(1); + + private final ArchiveStorage<Entry> delegate; + private final String blockOnKeyPrefix; + + public BlockingArchiveStorage(ArchiveStorage<Entry> delegate, String blockOnKeyPrefix) { + this.delegate = delegate; + this.blockOnKeyPrefix = blockOnKeyPrefix; + } + + @Override + public boolean exists(String key) throws IOException { + return delegate.exists(key); + } + + @Nullable + @Override + public Entry getEntry(String key) throws IOException { + return delegate.getEntry(key); + } + + @Override + public void putArchiveContent(String key, String archiveContent) throws IOException { + if (key.contains(blockOnKeyPrefix)) { + asyncStartLatch.countDown(); + try { + boolean released = releaseLatch.await(10, TimeUnit.SECONDS); + if (!released) { + throw new IOException( + "BlockingArchiveStorage: timed out waiting for release latch"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException("BlockingArchiveStorage: interrupted", e); + } + } + delegate.putArchiveContent(key, archiveContent); + } + + @Override + public void delete(String key) throws IOException { + delegate.delete(key); + } + + @Override + public void deleteEntriesByPrefix(String keyPrefix) throws IOException { + delegate.deleteEntriesByPrefix(keyPrefix); + } + + @Override + public List<Entry> getEntriesByPrefix(String prefix) throws IOException { + return delegate.getEntriesByPrefix(prefix); + } + + @Override + public String readArchiveContent(Entry entry) throws IOException { + return delegate.readArchiveContent(entry); + } + + @Override + public void close() throws IOException { + delegate.close(); + } + } + + private static final class JsonObject implements AutoCloseable { + + private final JsonGenerator gen; + + JsonObject(JsonGenerator gen) throws IOException { + this.gen = gen; + gen.writeStartObject(); + } + + private JsonObject(JsonGenerator gen, String name) throws IOException { + this.gen = gen; + gen.writeObjectFieldStart(name); + } + + @Override + public void close() throws IOException { + gen.writeEndObject(); + } + } + + private static final class JsonArray implements AutoCloseable { + + private final JsonGenerator gen; + + JsonArray(JsonGenerator gen, String name) throws IOException { + this.gen = gen; + gen.writeArrayFieldStart(name); + } + + @Override + public void close() throws IOException { + gen.writeEndArray(); + } + } +}
