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 ca95b59981bac50525eb8c226fa3e98b79bf024e Author: Zihao Chen <[email protected]> AuthorDate: Thu Jun 11 22:55:29 2026 +0800 [FLINK-39911][historyserver] Introduce ArchiveStorage abstraction and FileArchiveStorage implementation --- .../runtime/webmonitor/history/ArchiveStorage.java | 104 +++++++++++ .../webmonitor/history/FileArchiveStorage.java | 125 +++++++++++++ .../runtime/webmonitor/history/HistoryServer.java | 70 ++++---- .../HistoryServerApplicationArchiveFetcher.java | 100 ++++++----- .../history/HistoryServerArchiveFetcher.java | 109 +++++------- .../webmonitor/history/ArchiveStorageTest.java | 198 +++++++++++++++++++++ 6 files changed, 565 insertions(+), 141 deletions(-) diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/ArchiveStorage.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/ArchiveStorage.java new file mode 100644 index 00000000000..16e7d407dfc --- /dev/null +++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/ArchiveStorage.java @@ -0,0 +1,104 @@ +/* + * 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 javax.annotation.Nullable; + +import java.io.Closeable; +import java.io.IOException; +import java.util.List; + +/** + * Abstraction for the storage backend used by the {@link HistoryServer} to access archived job + * data. + * + * <p>Implementations can be backed by the local file system, RocksDB, or any other storage medium. + * + * <p>The archive JSON will be stored in different entry types, and each implementation can decide + * whether to read the full content or not. + * + * @param <Entry> Type of the storage entries. + */ +public interface ArchiveStorage<Entry> extends Closeable { + + /** + * Returns whether the entry identified by {@code key} exists in this storage. + * + * @param key storage key (typically the request path, e.g. {@code jobs/xxx/config.json}) + * @return {@code true} if the entry exists + */ + boolean exists(String key) throws IOException; + + /** + * Returns the entry identified by {@code key} from this storage. + * + * @param key storage key + * @return the entry, or null if the entry does not exist + * @throws IOException if the entry cannot be read + */ + @Nullable + Entry getEntry(String key) throws IOException; + + /** + * Stores the entry identified by {@code key} in this storage. + * + * @param key storage key + * @param archiveContent the archive content to store, this type is string because the archive + * content is always a JSON String + * @throws IOException if the entry cannot be written + */ + void putArchiveContent(String key, String archiveContent) throws IOException; + + /** + * Deletes the entry identified by {@code key} from this storage. + * + * @param key storage key + * @throws IOException if the entry cannot be deleted + */ + void delete(String key) throws IOException; + + /** + * Deletes all entries with key starting with {@code keyPrefix} from this storage. + * + * <p>Such as deleting all archived files for a given job or application. + * + * @param keyPrefix key prefix + * @throws IOException if any entries cannot be deleted + */ + void deleteEntriesByPrefix(String keyPrefix) throws IOException; + + /** + * Returns the entries identified by {@code prefix} from this storage. + * + * @param prefix storage key prefix + * @return the entries + * @throws IOException if any entries cannot be read + */ + List<Entry> getEntriesByPrefix(String prefix) throws IOException; + + /** + * Read the archive content from the entry. + * + * @param entry the entry to read + * @return the archive content + * @throws IOException if the entry cannot be read + */ + String readArchiveContent(Entry entry) throws IOException; +} diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/FileArchiveStorage.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/FileArchiveStorage.java new file mode 100644 index 00000000000..939887a1361 --- /dev/null +++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/FileArchiveStorage.java @@ -0,0 +1,125 @@ +/* + * 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.util.FileUtils; + +import javax.annotation.Nullable; + +import java.io.File; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.FileAlreadyExistsException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +import static org.apache.flink.util.Preconditions.checkNotNull; + +/** + * A file-system backed implementation of {@link ArchiveStorage}. + * + * <p>Each storage key is treated as a relative path under the configured {@code rootPath}. + */ +public class FileArchiveStorage implements ArchiveStorage<File> { + + /** Root directory under which all archive files are stored. */ + private final Path rootPath; + + /** + * Creates a new {@link FileArchiveStorage}. + * + * @param rootPath root directory for archive files; must not be {@code null} + * @throws IOException if the canonical path of {@code rootPath} cannot be resolved + */ + public FileArchiveStorage(File rootPath) throws IOException { + this.rootPath = checkNotNull(rootPath).getCanonicalFile().toPath(); + } + + @Override + public boolean exists(String key) throws IOException { + return Files.exists(rootPath.resolve(key)); + } + + @Nullable + @Override + public File getEntry(String key) throws IOException { + if (!exists(key)) { + return null; + } + return rootPath.resolve(key).toFile(); + } + + @Override + public void putArchiveContent(String key, String archiveContent) throws IOException { + writeTargetFile(rootPath.resolve(key), archiveContent); + } + + @Override + public void delete(String key) throws IOException { + FileUtils.deleteFileOrDirectory(rootPath.resolve(key).toFile()); + } + + @Override + public void deleteEntriesByPrefix(String keyPrefix) throws IOException { + FileUtils.deleteFileOrDirectory(rootPath.resolve(keyPrefix).toFile()); + } + + @Override + public List<File> getEntriesByPrefix(String keyPrefix) throws IOException { + Path directory = rootPath.resolve(keyPrefix); + if (!Files.isDirectory(directory)) { + return new ArrayList<>(); + } + try (Stream<Path> entries = Files.list(directory)) { + return entries.map(Path::toFile).collect(Collectors.toList()); + } + } + + @Override + public String readArchiveContent(File file) throws IOException { + return FileUtils.readFileUtf8(file); + } + + void writeTargetFile(Path target, String json) throws IOException { + Path parent = target.getParent(); + + try { + Files.createDirectories(parent); + } catch (FileAlreadyExistsException ignored) { + // there may be left-over directories from the previous attempt + } + + // We overwrite existing files since this may be another attempt + // at fetching this archive. + // Existing files may be incomplete/corrupt. + Files.deleteIfExists(target); + + Files.write(target, json.getBytes(StandardCharsets.UTF_8)); + } + + @Override + public void close() throws IOException { + // Nothing to close. + } +} 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 95648156093..8ddcd57a402 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 @@ -59,7 +59,6 @@ import org.slf4j.LoggerFactory; import javax.annotation.Nullable; import java.io.File; -import java.io.FileWriter; import java.io.IOException; import java.lang.reflect.UndeclaredThrowableException; import java.nio.file.Files; @@ -76,6 +75,11 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Consumer; +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; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerArchiveFetcher.JOB_OVERVIEWS_SUBDIR; + /** * The HistoryServer provides a WebInterface and REST API to retrieve information about finished * jobs for which the JobManager may have already shut down. @@ -116,13 +120,13 @@ public class HistoryServer { * The archive fetcher is responsible for fetching job archives that are not part of an * application (legacy jobs created before application archiving was introduced in FLINK-38761). */ - private final HistoryServerArchiveFetcher archiveFetcher; + private final HistoryServerArchiveFetcher<?> archiveFetcher; /** * The archive fetcher is responsible for fetching application archives and their associated job * archives. */ - private final HistoryServerApplicationArchiveFetcher applicationArchiveFetcher; + private final HistoryServerApplicationArchiveFetcher<?> applicationArchiveFetcher; @Nullable private final SSLHandlerFactory serverSSLFactory; private WebFrontendBootstrap netty; @@ -136,6 +140,8 @@ public class HistoryServer { private final AtomicBoolean shutdownRequested = new AtomicBoolean(false); private final Thread shutdownHook; + private final ArchiveStorage<?> archiveStorage; + public static void main(String[] args) throws Exception { EnvironmentInformation.logEnvironmentInfo(LOG, "HistoryServer", args); @@ -246,20 +252,30 @@ public class HistoryServer { refreshIntervalMillis = config.get(HistoryServerOptions.HISTORY_SERVER_ARCHIVE_REFRESH_INTERVAL).toMillis(); + + // create directories for job and application overview updates + Files.createDirectories(webDir.toPath().resolve(JOBS_SUBDIR)); + Files.createDirectories(webDir.toPath().resolve(JOB_OVERVIEWS_SUBDIR)); + Files.createDirectories(webDir.toPath().resolve(APPLICATIONS_SUBDIR)); + Files.createDirectories(webDir.toPath().resolve(APPLICATION_OVERVIEWS_SUBDIR)); + archiveStorage = new FileArchiveStorage(webDir); + archiveFetcher = - new HistoryServerArchiveFetcher( + new HistoryServerArchiveFetcher<>( refreshDirs, webDir, jobArchiveEventListener, cleanupExpiredJobs, - CompositeArchiveRetainedStrategy.createForJobFromConfig(config)); + CompositeArchiveRetainedStrategy.createForJobFromConfig(config), + archiveStorage); applicationArchiveFetcher = - new HistoryServerApplicationArchiveFetcher( + new HistoryServerApplicationArchiveFetcher<>( refreshDirs, webDir, applicationArchiveEventListener, cleanupExpiredApplications, - CompositeArchiveRetainedStrategy.createForApplicationFromConfig(config)); + CompositeArchiveRetainedStrategy.createForApplicationFromConfig(config), + archiveStorage); this.shutdownHook = ShutdownHookUtil.addShutdownHook( @@ -382,6 +398,12 @@ public class HistoryServer { ExecutorUtils.gracefulShutdown(1, TimeUnit.SECONDS, executor); + try { + archiveStorage.close(); + } catch (Throwable t) { + LOG.warn("Error while closing archive storage.", t); + } + try { LOG.info("Removing web dashboard root cache directory {}", webDir); FileUtils.deleteDirectory(webDir); @@ -401,31 +423,17 @@ public class HistoryServer { // File generation // ------------------------------------------------------------------------ - static FileWriter createOrGetFile(File folder, String name) throws IOException { - File file = new File(folder, name + ".json"); - if (!file.exists()) { - Files.createFile(file.toPath()); - } - FileWriter fr = new FileWriter(file); - return fr; - } - private void createDashboardConfigFile() throws IOException { - try (FileWriter fw = createOrGetFile(webDir, "config")) { - fw.write( - createConfigJson( - DashboardConfiguration.from( - webRefreshIntervalMillis, - ZonedDateTime.now(), - false, - false, - false, - true))); - fw.flush(); - } catch (IOException ioe) { - LOG.error("Failed to write config file."); - throw ioe; - } + String configJson = + createConfigJson( + DashboardConfiguration.from( + webRefreshIntervalMillis, + ZonedDateTime.now(), + false, + false, + false, + true)); + archiveStorage.putArchiveContent("config.json", configJson); } private static String createConfigJson(DashboardConfiguration dashboardConfiguration) 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 310e91b7753..10b8161b95b 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 @@ -29,16 +29,12 @@ 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.runtime.webmonitor.history.retaining.ArchiveRetainedStrategy; -import org.apache.flink.util.FileUtils; - -import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.File; import java.io.IOException; -import java.nio.file.Files; import java.util.ArrayList; import java.util.Collection; import java.util.HashMap; @@ -59,37 +55,42 @@ import java.util.function.Consumer; * <p>Removes existing archives from these directories and the cache according to {@link * ArchiveRetainedStrategy} and {@link * HistoryServerOptions#HISTORY_SERVER_CLEANUP_EXPIRED_APPLICATIONS}. + * + * @param <Entry> the type of entries returned by the underlying {@link ArchiveStorage}. */ -public class HistoryServerApplicationArchiveFetcher extends HistoryServerArchiveFetcher { +public class HistoryServerApplicationArchiveFetcher<Entry> + extends HistoryServerArchiveFetcher<Entry> { private static final Logger LOG = LoggerFactory.getLogger(HistoryServerApplicationArchiveFetcher.class); - private static final String APPLICATIONS_SUBDIR = "applications"; - private static final String APPLICATION_OVERVIEWS_SUBDIR = "application-overviews"; + protected static final String APPLICATIONS_SUBDIR = "applications"; + protected static final String APPLICATION_OVERVIEWS_SUBDIR = "application-overviews"; + private static final String APPLICATION_KEY_PREFIX = APPLICATIONS_SUBDIR + "/"; + private static final String APPLICATION_OVERVIEWS_KEY_PREFIX = + APPLICATION_OVERVIEWS_SUBDIR + "/"; private final Map<Path, Map<String, Set<String>>> cachedApplicationIdsToJobIds = new HashMap<>(); - private final File webApplicationDir; - private final File webApplicationsOverviewDir; - HistoryServerApplicationArchiveFetcher( List<HistoryServer.RefreshLocation> refreshDirs, File webDir, Consumer<HistoryServerApplicationArchiveFetcher.ArchiveEvent> archiveEventListener, boolean cleanupExpiredArchives, - ArchiveRetainedStrategy retainedStrategy) - throws IOException { - super(refreshDirs, webDir, archiveEventListener, cleanupExpiredArchives, retainedStrategy); + ArchiveRetainedStrategy retainedStrategy, + ArchiveStorage<Entry> archiveStorage) { + super( + refreshDirs, + webDir, + archiveEventListener, + cleanupExpiredArchives, + retainedStrategy, + archiveStorage); for (HistoryServer.RefreshLocation refreshDir : refreshDirs) { cachedApplicationIdsToJobIds.put(refreshDir.getPath(), new HashMap<>()); } - this.webApplicationDir = new File(webDir, APPLICATIONS_SUBDIR); - Files.createDirectories(webApplicationDir.toPath()); - this.webApplicationsOverviewDir = new File(webDir, APPLICATION_OVERVIEWS_SUBDIR); - Files.createDirectories(webApplicationsOverviewDir.toPath()); updateApplicationOverview(); } @@ -173,15 +174,15 @@ public class HistoryServerApplicationArchiveFetcher extends HistoryServerArchive String path = archive.getPath(); String json = archive.getJson(); - File target; + String key; if (path.equals(ApplicationsOverviewHeaders.URL)) { - target = new File(webApplicationsOverviewDir, applicationId + JSON_FILE_ENDING); + key = APPLICATION_OVERVIEWS_KEY_PREFIX + applicationId + JSON_FILE_ENDING; } else { - // this implicitly writes into webApplicationDir - target = new File(webDir, path + JSON_FILE_ENDING); + // the key should be a relative sub-path under the storage root + key = path.substring(1) + JSON_FILE_ENDING; } - writeTargetFile(target, json); + archiveStorage.putArchiveContent(key, json); } return new ArchiveEvent(applicationId, ArchiveEventType.CREATED); @@ -209,26 +210,26 @@ public class HistoryServerApplicationArchiveFetcher extends HistoryServerArchive } private ArchiveEvent deleteApplicationFiles(String applicationId) { - // Make sure we do not include this application in the overview + // Delete application overview file in application-overviews directory try { - Files.deleteIfExists( - new File(webApplicationsOverviewDir, applicationId + JSON_FILE_ENDING) - .toPath()); + archiveStorage.delete( + APPLICATION_OVERVIEWS_KEY_PREFIX + applicationId + JSON_FILE_ENDING); } catch (IOException ioe) { LOG.warn("Could not delete file from overview directory.", ioe); } - // Clean up application files we may have created - File applicationDirectory = new File(webApplicationDir, applicationId); + // Delete application details directory in applications directory, + // applications/application-id/ try { - FileUtils.deleteDirectory(applicationDirectory); + archiveStorage.deleteEntriesByPrefix(APPLICATION_KEY_PREFIX + applicationId + "/"); } catch (IOException ioe) { LOG.warn("Could not clean up application directory.", ioe); } + // Delete application overview file in applications directory, + // applications/application-id.json try { - Files.deleteIfExists( - new File(webApplicationDir, applicationId + JSON_FILE_ENDING).toPath()); + archiveStorage.delete(APPLICATION_KEY_PREFIX + applicationId + JSON_FILE_ENDING); } catch (IOException ioe) { LOG.warn("Could not delete file from application directory.", ioe); } @@ -253,21 +254,32 @@ public class HistoryServerApplicationArchiveFetcher extends HistoryServerArchive * <p>For the display in the HistoryServer WebFrontend we have to combine these overviews. */ private void updateApplicationOverview() { - try (JsonGenerator gen = - jacksonFactory.createGenerator( - HistoryServer.createOrGetFile(webDir, ApplicationsOverviewHeaders.URL))) { - File[] overviews = new File(webApplicationsOverviewDir.getPath()).listFiles(); - if (overviews != null) { - Collection<ApplicationDetails> allApplications = new ArrayList<>(overviews.length); - for (File overview : overviews) { - MultipleApplicationsDetails subApplications = - mapper.readValue(overview, MultipleApplicationsDetails.class); - allApplications.addAll(subApplications.getApplications()); + try { + Collection<ApplicationDetails> allApplications = new ArrayList<>(); + List<Entry> overviews = + archiveStorage.getEntriesByPrefix(APPLICATION_OVERVIEWS_KEY_PREFIX); + for (Entry overview : overviews) { + MultipleApplicationsDetails subApplications; + // We treated File as a special case, mainly as a performance trade-off to avoid the + // overhead of loading the archive into string. + if (overview instanceof File) { + subApplications = + mapper.readValue((File) overview, MultipleApplicationsDetails.class); + } else { + subApplications = + mapper.readValue( + archiveStorage.readArchiveContent(overview), + MultipleApplicationsDetails.class); } - mapper.writeValue(gen, new MultipleApplicationsDetails(allApplications)); + allApplications.addAll(subApplications.getApplications()); } - } catch (IOException ioe) { - LOG.error("Failed to update application overview.", ioe); + String overviewWithApplications = + mapper.writeValueAsString(new MultipleApplicationsDetails(allApplications)); + archiveStorage.putArchiveContent( + ApplicationsOverviewHeaders.URL.substring(1) + JSON_FILE_ENDING, + overviewWithApplications); + } catch (Exception e) { + LOG.error("Failed to update application overview.", e); } } } 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 59eb258b3ed..e64114ef3da 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,11 +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.FileUtils; 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.JsonNode; import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper; @@ -42,11 +39,8 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.File; -import java.io.FileWriter; import java.io.IOException; import java.io.StringWriter; -import java.nio.file.FileAlreadyExistsException; -import java.nio.file.Files; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; @@ -69,8 +63,10 @@ import static org.apache.flink.util.Preconditions.checkNotNull; * * <p>Removes existing archives from these directories and the cache according to {@link * ArchiveRetainedStrategy} and {@link HistoryServerOptions#HISTORY_SERVER_CLEANUP_EXPIRED_JOBS}. + * + * @param <Entry> the type of entries returned by the underlying {@link ArchiveStorage}. */ -class HistoryServerArchiveFetcher { +public class HistoryServerArchiveFetcher<Entry> { /** Possible archive operations in history-server. */ public enum ArchiveEventType { @@ -104,8 +100,9 @@ class HistoryServerArchiveFetcher { protected static final String JSON_FILE_ENDING = ".json"; protected static final String JOBS_SUBDIR = "jobs"; protected static final String JOB_OVERVIEWS_SUBDIR = "overviews"; + protected static final String JOBS_KEY_PREFIX = JOBS_SUBDIR + "/"; + protected static final String JOB_OVERVIEWS_KEY_PREFIX = JOB_OVERVIEWS_SUBDIR + "/"; - protected final JsonFactory jacksonFactory = new JsonFactory(); protected final ObjectMapper mapper = JacksonMapperFactory.createObjectMapper(); protected final List<HistoryServer.RefreshLocation> refreshDirs; @@ -116,17 +113,15 @@ class HistoryServerArchiveFetcher { /** Cache of all available archives identified by their id. */ protected final Map<Path, Set<String>> cachedArchivesPerRefreshDirectory; - protected final File webDir; - protected final File webJobDir; - protected final File webOverviewDir; + protected final ArchiveStorage<Entry> archiveStorage; HistoryServerArchiveFetcher( List<HistoryServer.RefreshLocation> refreshDirs, File webDir, Consumer<ArchiveEvent> archiveEventListener, boolean cleanupExpiredArchives, - ArchiveRetainedStrategy retainedStrategy) - throws IOException { + ArchiveRetainedStrategy retainedStrategy, + ArchiveStorage<Entry> archiveStorage) { this.refreshDirs = checkNotNull(refreshDirs); this.archiveEventListener = archiveEventListener; this.processExpiredArchiveDeletion = cleanupExpiredArchives; @@ -135,11 +130,8 @@ class HistoryServerArchiveFetcher { for (HistoryServer.RefreshLocation refreshDir : refreshDirs) { cachedArchivesPerRefreshDirectory.put(refreshDir.getPath(), new HashSet<>()); } - this.webDir = checkNotNull(webDir); - this.webJobDir = new File(webDir, JOBS_SUBDIR); - Files.createDirectories(webJobDir.toPath()); - this.webOverviewDir = new File(webDir, JOB_OVERVIEWS_SUBDIR); - Files.createDirectories(webOverviewDir.toPath()); + checkNotNull(webDir); + this.archiveStorage = archiveStorage; updateJobOverview(); if (LOG.isInfoEnabled()) { @@ -272,47 +264,25 @@ class HistoryServerArchiveFetcher { String path = archive.getPath(); String json = archive.getJson(); - File target; + String key; if (path.equals(JobsOverviewHeaders.URL)) { - target = new File(webOverviewDir, jobId + JSON_FILE_ENDING); + key = JOB_OVERVIEWS_KEY_PREFIX + jobId + JSON_FILE_ENDING; } else if (path.equals("/joboverview")) { // legacy path LOG.debug("Migrating legacy archive {}", jobArchive); json = convertLegacyJobOverview(json); - target = new File(webOverviewDir, jobId + JSON_FILE_ENDING); + key = JOB_OVERVIEWS_KEY_PREFIX + jobId + JSON_FILE_ENDING; } else { - // this implicitly writes into webJobDir - target = new File(webDir, path + JSON_FILE_ENDING); + // 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 + key = path.substring(1) + JSON_FILE_ENDING; } - writeTargetFile(target, json); + archiveStorage.putArchiveContent(key, json); } return new ArchiveEvent(jobId, ArchiveEventType.CREATED); } - void writeTargetFile(File target, String json) throws IOException { - java.nio.file.Path parent = target.getParentFile().toPath(); - - try { - Files.createDirectories(parent); - } catch (FileAlreadyExistsException ignored) { - // there may be left-over directories from the previous attempt - } - - java.nio.file.Path targetPath = target.toPath(); - - // We overwrite existing files since this may be another attempt - // at fetching this archive. - // Existing files may be incomplete/corrupt. - Files.deleteIfExists(targetPath); - - Files.createFile(target.toPath()); - try (FileWriter fw = new FileWriter(target)) { - fw.write(json); - fw.flush(); - } - } - List<ArchiveEvent> cleanupArchivesBeyondRetainedLimit(Map<Path, Set<Path>> archivesToRemove) { Map<Path, Set<String>> allArchiveIdsToRemove = new HashMap<>(); @@ -357,23 +327,23 @@ class HistoryServerArchiveFetcher { } ArchiveEvent deleteJobFiles(String jobId) { - // Make sure we do not include this job in the overview + // Delete job overview file in overviews directory try { - Files.deleteIfExists(new File(webOverviewDir, jobId + JSON_FILE_ENDING).toPath()); + archiveStorage.delete(JOB_OVERVIEWS_KEY_PREFIX + jobId + JSON_FILE_ENDING); } catch (IOException ioe) { LOG.warn("Could not delete file from overview directory.", ioe); } - // Clean up job files we may have created - File jobDirectory = new File(webJobDir, jobId); + // Delete job details directory in jobs directory, jobs/job-id/ try { - FileUtils.deleteDirectory(jobDirectory); + archiveStorage.deleteEntriesByPrefix(JOBS_KEY_PREFIX + jobId + "/"); } catch (IOException ioe) { LOG.warn("Could not clean up job directory.", ioe); } + // Delete job overview file in jobs directory, jobs/job-id.json try { - Files.deleteIfExists(new File(webJobDir, jobId + JSON_FILE_ENDING).toPath()); + archiveStorage.delete(JOBS_KEY_PREFIX + jobId + JSON_FILE_ENDING); } catch (IOException ioe) { LOG.warn("Could not delete file from job directory.", ioe); } @@ -464,21 +434,28 @@ class HistoryServerArchiveFetcher { * <p>For the display in the HistoryServer WebFrontend we have to combine these overviews. */ void updateJobOverview() { - try (JsonGenerator gen = - jacksonFactory.createGenerator( - HistoryServer.createOrGetFile(webDir, JobsOverviewHeaders.URL))) { - File[] overviews = new File(webOverviewDir.getPath()).listFiles(); - if (overviews != null) { - Collection<JobDetails> allJobs = new ArrayList<>(overviews.length); - for (File overview : overviews) { - MultipleJobsDetails subJobs = - mapper.readValue(overview, MultipleJobsDetails.class); - allJobs.addAll(subJobs.getJobs()); + try { + Collection<JobDetails> allJobs = new ArrayList<>(); + List<Entry> overviews = archiveStorage.getEntriesByPrefix(JOB_OVERVIEWS_KEY_PREFIX); + for (Entry overview : overviews) { + MultipleJobsDetails subJobs; + // We treated File as a special case, mainly as a performance trade-off to avoid the + // overhead of loading the archive into string. + if (overview instanceof File) { + subJobs = mapper.readValue((File) overview, MultipleJobsDetails.class); + } else { + subJobs = + mapper.readValue( + archiveStorage.readArchiveContent(overview), + MultipleJobsDetails.class); } - mapper.writeValue(gen, new MultipleJobsDetails(allJobs)); + allJobs.addAll(subJobs.getJobs()); } - } catch (IOException ioe) { - LOG.error("Failed to update job overview.", ioe); + String overviewWithJobs = mapper.writeValueAsString(new MultipleJobsDetails(allJobs)); + archiveStorage.putArchiveContent( + JobsOverviewHeaders.URL.substring(1) + JSON_FILE_ENDING, overviewWithJobs); + } catch (Exception e) { + LOG.error("Failed to update job overview.", e); } } } diff --git a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/ArchiveStorageTest.java b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/ArchiveStorageTest.java new file mode 100644 index 00000000000..46587a9d58e --- /dev/null +++ b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/ArchiveStorageTest.java @@ -0,0 +1,198 @@ +/* + * 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.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.HashMap; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Basic tests for {@link ArchiveStorage} implementations. + * + * <p>New {@link ArchiveStorage} implementations only need to register an instance in {@link + * #storageFactories()} to be covered by these tests. + */ +@ExtendWith(ParameterizedTestExtension.class) +class ArchiveStorageTest { + + @TempDir private File tempDir; + + @Parameter public ArchiveStorageFactory<Object> storageFactory; + + private ArchiveStorage<Object> storage; + + @Parameters(name = "storageFactory={0}") + private static Collection<ArchiveStorageFactory<?>> storageFactories() { + ArchiveStorageFactory<File> fileArchiveStorageFactory = FileArchiveStorage::new; + return List.of(fileArchiveStorageFactory); + } + + /** 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 { + storage = storageFactory.create(tempDir); + } + + @AfterEach + void tearDown() throws Exception { + if (storage != null) { + storage.close(); + } + } + + // ----------------------------- exists / put / get / delete ---------------------------- + + /** + * Covers the whole lifecycle of a single key across {@code exists}, {@code put}, {@code get} + * and {@code delete}. + */ + @TestTemplate + void testBasicLifecycle() throws Exception { + String key = "overviews/abc123.json"; + String content = "{\"test\":\"data\"}"; + + // exists: missing key + assertThat(storage.exists(key)).isFalse(); + + // put + exists + get returns the same content + storage.putArchiveContent(key, content); + assertThat(storage.exists(key)).isTrue(); + assertThat(storage.readArchiveContent(storage.getEntry(key))).isEqualTo(content); + + // delete + exists + get returns null + storage.delete(key); + assertThat(storage.exists(key)).isFalse(); + assertThat(storage.getEntry(key)).isNull(); + } + + @TestTemplate + void testOverwrite() throws Exception { + String key = "overviews/abc123.json"; + String content = "{\"test\":\"data\"}"; + + // put + exists + get returns the same content + storage.putArchiveContent(key, content); + assertThat(storage.exists(key)).isTrue(); + assertThat(storage.readArchiveContent(storage.getEntry(key))).isEqualTo(content); + + // put again: the latest content overwrites the previous one + String overwriteContent = "{\"test\":\"overwrite_data\"}"; + storage.putArchiveContent(key, overwriteContent); + assertThat(storage.exists(key)).isTrue(); + assertThat(storage.readArchiveContent(storage.getEntry(key))).isEqualTo(overwriteContent); + } + + // ----------------------------- deleteEntriesByPrefix ---------------------------- + + @TestTemplate + void testDeleteEntriesByPrefix() throws Exception { + String keyUnderPrefix1 = "jobs/abc123/config.json"; + String keyUnderPrefix2 = "jobs/abc123/vertices.json"; + String keyOutsidePrefix = "jobs/def456/config.json"; + storage.putArchiveContent(keyUnderPrefix1, "{\"config\":\"under_prefix_data\"}"); + storage.putArchiveContent(keyUnderPrefix2, "{\"vertices\":\"under_prefix_data\"}"); + storage.putArchiveContent(keyOutsidePrefix, "{\"config\":\"outside_prefix_data\"}"); + + storage.deleteEntriesByPrefix("jobs/abc123/"); + + assertThat(storage.exists(keyUnderPrefix1)).isFalse(); + assertThat(storage.exists(keyUnderPrefix2)).isFalse(); + assertThat(storage.exists(keyOutsidePrefix)).isTrue(); + assertThat(storage.readArchiveContent(storage.getEntry(keyOutsidePrefix))) + .isEqualTo("{\"config\":\"outside_prefix_data\"}"); + } + + @TestTemplate + void testDeleteEntriesByPrefixDoesNotTouchSiblingKey() throws Exception { + String keyUnderPrefix = "jobs/abc123/config.json"; + String siblingKey = "jobs/abc123.json"; + storage.putArchiveContent(keyUnderPrefix, "{\"config\":\"under_prefix_data\"}"); + storage.putArchiveContent(siblingKey, "{\"sibling\":\"data\"}"); + + storage.deleteEntriesByPrefix("jobs/abc123/"); + + assertThat(storage.exists(keyUnderPrefix)).isFalse(); + assertThat(storage.exists(siblingKey)).isTrue(); + assertThat(storage.readArchiveContent(storage.getEntry(siblingKey))) + .isEqualTo("{\"sibling\":\"data\"}"); + } + + // ----------------------------- getEntriesByPrefix ---------------------------- + + @TestTemplate + void testGetEntriesByPrefix() throws Exception { + Map<String, String> entriesUnderPrefix = new HashMap<>(); + entriesUnderPrefix.put("overviews/job1.json", "{\"job\":\"job1\"}"); + entriesUnderPrefix.put("overviews/job2.json", "{\"job\":\"job2\"}"); + entriesUnderPrefix.put("overviews/job3.json", "{\"job\":\"job3\"}"); + String keyOutsidePrefix = "jobs/job1/config.json"; + + for (Map.Entry<String, String> e : entriesUnderPrefix.entrySet()) { + storage.putArchiveContent(e.getKey(), e.getValue()); + } + storage.putArchiveContent(keyOutsidePrefix, "{\"config\":\"data\"}"); + + List<Object> result = storage.getEntriesByPrefix("overviews/"); + List<String> contents = new ArrayList<>(); + for (Object entry : result) { + contents.add(storage.readArchiveContent(entry)); + } + assertThat(contents).containsExactlyInAnyOrderElementsOf(entriesUnderPrefix.values()); + } + + @TestTemplate + void testGetEntriesByPrefixDoesNotIncludeSiblingKey() throws Exception { + Map<String, String> entriesUnderPrefix = new HashMap<>(); + entriesUnderPrefix.put("overviews/job1.json", "{\"job\":\"job1\"}"); + entriesUnderPrefix.put("overviews/job2.json", "{\"job\":\"job2\"}"); + String siblingKey = "overviews-legacy.json"; + + for (Map.Entry<String, String> e : entriesUnderPrefix.entrySet()) { + storage.putArchiveContent(e.getKey(), e.getValue()); + } + storage.putArchiveContent(siblingKey, "{\"sibling\":\"data\"}"); + + List<Object> result = storage.getEntriesByPrefix("overviews/"); + List<String> contents = new ArrayList<>(); + for (Object entry : result) { + contents.add(storage.readArchiveContent(entry)); + } + assertThat(contents).containsExactlyInAnyOrderElementsOf(entriesUnderPrefix.values()); + } +}
