You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gu...@apache.org on 2026/06/26 06:07:58 UTC
(flink) branch master updated (c5d3273ffdd -> 0fbd749cacc)
This is an automated email from the ASF dual-hosted git repository.
reswqa pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from c5d3273ffdd [hotfix][release] Update japicmp configuration for 2.3.0 (#28536)
new ca95b59981b [FLINK-39911][historyserver] Introduce ArchiveStorage abstraction and FileArchiveStorage implementation
new c1f4c5ea64c [FLINK-39911][historyserver] Extract AbstractHistoryServerHandler from HistoryServerStaticFileServerHandler
new d531fb808ea [FLINK-39911][historyserver] Support RocksDB as archive storage backend
new 0fbd749cacc [FLINK-39911][docs] Document HistoryServer archive storage backend
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../docs/deployment/advanced/historyserver.md | 13 +
.../docs/deployment/advanced/historyserver.md | 13 +
.../generated/history_server_configuration.html | 6 +
.../flink/configuration/HistoryServerOptions.java | 19 ++
flink-runtime-web/pom.xml | 10 +
...dler.java => AbstractHistoryServerHandler.java} | 227 +++++++++-------
.../runtime/webmonitor/history/ArchiveStorage.java | 104 +++++++
.../webmonitor/history/FileArchiveStorage.java | 125 +++++++++
.../runtime/webmonitor/history/HistoryServer.java | 94 ++++---
.../HistoryServerApplicationArchiveFetcher.java | 100 ++++---
.../history/HistoryServerArchiveFetcher.java | 109 +++-----
.../history/HistoryServerRocksDBHandler.java | 55 ++++
.../HistoryServerStaticFileServerHandler.java | 228 +---------------
.../webmonitor/history/RocksDBArchiveStorage.java | 299 +++++++++++++++++++++
.../history/AbstractHistoryServerHandlerTest.java | 187 +++++++++++++
.../webmonitor/history/ArchiveStorageTest.java | 199 ++++++++++++++
.../HistoryServerStaticFileServerHandlerTest.java | 95 -------
.../web-dashboard/src/app/app.interceptor.ts | 2 +-
.../flink-statebackend-rocksdb/pom.xml | 2 +-
pom.xml | 1 +
20 files changed, 1337 insertions(+), 551 deletions(-)
copy flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/{HistoryServerStaticFileServerHandler.java => AbstractHistoryServerHandler.java} (62%)
create mode 100644 flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/ArchiveStorage.java
create mode 100644 flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/FileArchiveStorage.java
create mode 100644 flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerRocksDBHandler.java
create mode 100644 flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/RocksDBArchiveStorage.java
create mode 100644 flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/AbstractHistoryServerHandlerTest.java
create mode 100644 flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/ArchiveStorageTest.java
delete mode 100644 flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerStaticFileServerHandlerTest.java
(flink) 03/04: [FLINK-39911][historyserver] Support RocksDB as archive storage backend
Posted by gu...@apache.org.
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 d531fb808ea7407d2d7b5510191eab172ba52317
Author: Zihao Chen <is...@gmail.com>
AuthorDate: Thu Jun 11 23:43:20 2026 +0800
[FLINK-39911][historyserver] Support RocksDB as archive storage backend
---
.../generated/history_server_configuration.html | 6 +
.../flink/configuration/HistoryServerOptions.java | 19 ++
flink-runtime-web/pom.xml | 10 +
.../runtime/webmonitor/history/HistoryServer.java | 39 ++-
.../history/HistoryServerRocksDBHandler.java | 55 ++++
.../webmonitor/history/RocksDBArchiveStorage.java | 299 +++++++++++++++++++++
.../history/AbstractHistoryServerHandlerTest.java | 12 +-
.../webmonitor/history/ArchiveStorageTest.java | 3 +-
.../flink-statebackend-rocksdb/pom.xml | 2 +-
pom.xml | 1 +
10 files changed, 431 insertions(+), 15 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/history_server_configuration.html b/docs/layouts/shortcodes/generated/history_server_configuration.html
index db696545a18..896c6813563 100644
--- a/docs/layouts/shortcodes/generated/history_server_configuration.html
+++ b/docs/layouts/shortcodes/generated/history_server_configuration.html
@@ -50,6 +50,12 @@
<td>Duration</td>
<td>The time-to-live duration to retain the archived entities (jobs and applications) in each archive directory defined by <code class="highlighter-rouge">historyserver.archive.fs.dir</code>. This option works together with the retention count limits (see <code class="highlighter-rouge">historyserver.archive.retained-applications</code> and <code class="highlighter-rouge">historyserver.archive.retained-jobs</code>). Archived entities will be removed if their TTL has expired o [...]
</tr>
+ <tr>
+ <td><h5>historyserver.archive.storage.type</h5></td>
+ <td style="word-wrap: break-word;">FILE</td>
+ <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.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 1677dc2e1a0..c90f92ccc3d 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
@@ -247,5 +247,24 @@ public class HistoryServerOptions {
text(CONFIGURE_CONSISTENT))
.build());
+ public static final ConfigOption<HistoryServerArchiveStorageType>
+ HISTORY_SERVER_ARCHIVE_STORAGE_TYPE =
+ key("historyserver.archive.storage.type")
+ .enumType(HistoryServerArchiveStorageType.class)
+ .defaultValue(HistoryServerArchiveStorageType.FILE)
+ .withDescription(
+ Description.builder()
+ .text("The type of archive storage.")
+ .build());
+
+ /** The type of archive storage. */
+ public enum HistoryServerArchiveStorageType {
+ /** Local file system. */
+ FILE,
+
+ /** RocksDB. */
+ ROCKSDB
+ }
+
private HistoryServerOptions() {}
}
diff --git a/flink-runtime-web/pom.xml b/flink-runtime-web/pom.xml
index 697d1a33137..279aeb5c76c 100644
--- a/flink-runtime-web/pom.xml
+++ b/flink-runtime-web/pom.xml
@@ -80,6 +80,16 @@ under the License.
<artifactId>flink-shaded-jackson</artifactId>
</dependency>
+ <!-- ===================================================
+ RocksDB Storage Backend
+ =================================================== -->
+
+ <dependency>
+ <groupId>com.ververica</groupId>
+ <artifactId>frocksdbjni</artifactId>
+ <version>${rocksdb.version}</version>
+ </dependency>
+
<!-- ===================================================
Testing
=================================================== -->
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 ba836be15ef..ed59bf2db4c 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
@@ -141,6 +141,7 @@ public class HistoryServer {
private final Thread shutdownHook;
private final ArchiveStorage<?> archiveStorage;
+ private final AbstractHistoryServerHandler<?> historyServerHandler;
public static void main(String[] args) throws Exception {
EnvironmentInformation.logEnvironmentInfo(LOG, "HistoryServer", args);
@@ -253,12 +254,31 @@ 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);
+ HistoryServerOptions.HistoryServerArchiveStorageType archiveStorageType =
+ config.get(HistoryServerOptions.HISTORY_SERVER_ARCHIVE_STORAGE_TYPE);
+ switch (archiveStorageType) {
+ case FILE:
+ // 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);
+ historyServerHandler =
+ new HistoryServerStaticFileServerHandler(
+ (FileArchiveStorage) archiveStorage, webDir);
+ break;
+ case ROCKSDB:
+ File dbPath = new File(webDir, "rocksdb-" + UUID.randomUUID());
+ Files.createDirectories(dbPath.toPath());
+ archiveStorage = new RocksDBArchiveStorage(dbPath, config);
+ historyServerHandler =
+ new HistoryServerRocksDBHandler(
+ (RocksDBArchiveStorage) archiveStorage, webDir);
+ break;
+ default:
+ throw new FlinkException("Unsupported archive storage type: " + archiveStorageType);
+ }
archiveFetcher =
new HistoryServerArchiveFetcher<>(
@@ -363,13 +383,10 @@ public class HistoryServer {
new GeneratedLogUrlHandler(
CompletableFuture.completedFuture(pattern))));
- router.addGet(
- "/:*",
- new HistoryServerStaticFileServerHandler(
- (FileArchiveStorage) archiveStorage, webDir));
-
createDashboardConfigFile();
+ router.addGet("/:*", historyServerHandler);
+
executor.scheduleWithFixedDelay(
getArchiveFetchingRunnable(), 0, refreshIntervalMillis, TimeUnit.MILLISECONDS);
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerRocksDBHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerRocksDBHandler.java
new file mode 100644
index 00000000000..147bc93eaf9
--- /dev/null
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerRocksDBHandler.java
@@ -0,0 +1,55 @@
+/*
+ * 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.runtime.rest.handler.util.HandlerUtils;
+
+import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandler;
+import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandlerContext;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpRequest;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Collections;
+
+/**
+ * HistoryServer handler based on the RocksDB storage backend.
+ *
+ * <p>JSON data will be responded with the JSON string.
+ *
+ * <p>Static files will be responded with the file.
+ */
+@ChannelHandler.Sharable
+public class HistoryServerRocksDBHandler extends AbstractHistoryServerHandler<String> {
+
+ public HistoryServerRocksDBHandler(RocksDBArchiveStorage rocksDBArchiveStorage, File rootPath)
+ throws IOException {
+ super(rocksDBArchiveStorage, rootPath);
+ }
+
+ @Override
+ protected void respondWithResource(
+ ChannelHandlerContext ctx, HttpRequest request, String requestPath, String resource)
+ throws Exception {
+ HandlerUtils.sendResponse(
+ ctx, request, resource, HttpResponseStatus.OK, Collections.emptyMap());
+ }
+}
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/RocksDBArchiveStorage.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/RocksDBArchiveStorage.java
new file mode 100644
index 00000000000..a7908830c9d
--- /dev/null
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/RocksDBArchiveStorage.java
@@ -0,0 +1,299 @@
+/*
+ * 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.configuration.Configuration;
+import org.apache.flink.configuration.HistoryServerOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.util.FileUtils;
+import org.apache.flink.util.IOUtils;
+import org.apache.flink.util.StringUtils;
+
+import org.rocksdb.BlockBasedTableConfig;
+import org.rocksdb.BloomFilter;
+import org.rocksdb.CompressionType;
+import org.rocksdb.NativeLibraryLoader;
+import org.rocksdb.Options;
+import org.rocksdb.ReadOptions;
+import org.rocksdb.RocksDB;
+import org.rocksdb.RocksDBException;
+import org.rocksdb.RocksIterator;
+import org.rocksdb.Slice;
+import org.rocksdb.WriteOptions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * A {@link ArchiveStorage} implementation backed by RocksDB.
+ *
+ * <p>All archived data is stored as key-value pairs in a single RocksDB instance, avoiding the
+ * problem of numerous small files. The key is the request path (e.g. {@code
+ * /jobs/xxx/config.json}), and the value is a JSON string.
+ */
+public class RocksDBArchiveStorage implements ArchiveStorage<String> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RocksDBArchiveStorage.class);
+
+ private final RocksDB db;
+
+ private Options dbOptions;
+
+ private WriteOptions writeOptions;
+
+ private final List<AutoCloseable> handlesToClose;
+
+ /**
+ * The temporary directory that holds the extracted RocksDB native library; cleaned up on {@link
+ * #close()}.
+ */
+ private final File nativeLibDir;
+
+ /**
+ * Creates a new {@link RocksDBArchiveStorage} instance with default RocksDB options.
+ *
+ * @param dbPath the RocksDB database directory path
+ * @throws IOException if the RocksDB database cannot be opened
+ */
+ public RocksDBArchiveStorage(File dbPath) throws IOException {
+ this(dbPath, new Configuration());
+ }
+
+ /**
+ * Creates a new {@link RocksDBArchiveStorage} instance.
+ *
+ * @param dbPath the RocksDB database directory path
+ * @param config the configuration used to read RocksDB related options (see {@link
+ * HistoryServerOptions})
+ * @throws IOException if the RocksDB native library cannot be loaded or the database cannot be
+ * opened
+ */
+ public RocksDBArchiveStorage(File dbPath, ReadableConfig config) throws IOException {
+ checkNotNull(dbPath, "dbPath cannot be null");
+ checkNotNull(config, "config cannot be null");
+ this.handlesToClose = new ArrayList<>();
+ this.nativeLibDir = new File(dbPath, "rocksdb-lib");
+
+ try {
+ loadRocksDBLibrary(this.nativeLibDir);
+ loadRocksDBConfiguration();
+ this.db = RocksDB.open(dbOptions, dbPath.getAbsolutePath());
+ handlesToClose.add(db);
+ } catch (Throwable t) {
+ close();
+ throw new IOException("Failed to initialize RocksDBArchiveStorage", t);
+ }
+ }
+
+ @Override
+ public boolean exists(String key) throws IOException {
+ return db.keyExists(key.getBytes(UTF_8));
+ }
+
+ @Nullable
+ @Override
+ public String getEntry(String key) throws IOException {
+ try {
+ byte[] value = db.get(key.getBytes(UTF_8));
+ if (value == null) {
+ return null;
+ }
+ return new String(value, UTF_8);
+ } catch (RocksDBException e) {
+ throw new IOException("Failed to get key: " + key, e);
+ }
+ }
+
+ @Override
+ public void putArchiveContent(String key, String value) throws IOException {
+ try {
+ db.put(writeOptions, key.getBytes(UTF_8), value.getBytes(UTF_8));
+ } catch (RocksDBException e) {
+ throw new IOException("Failed to put key: " + key, e);
+ }
+ }
+
+ @Override
+ public void delete(String key) throws IOException {
+ try {
+ db.delete(writeOptions, key.getBytes(UTF_8));
+ } catch (RocksDBException e) {
+ throw new IOException("Failed to delete key: " + key, e);
+ }
+ }
+
+ @Override
+ public void deleteEntriesByPrefix(String keyPrefix) throws IOException {
+ if (StringUtils.isNullOrWhitespaceOnly(keyPrefix)) {
+ return;
+ }
+
+ // Delete all keys that start with the given prefix
+ byte[] startKey = keyPrefix.getBytes(UTF_8);
+ byte[] endKey = computeNextKey(startKey);
+ if (endKey == null) {
+ throw new IOException("Failed to compute next key for prefix: " + keyPrefix);
+ }
+
+ try {
+ db.deleteRange(writeOptions, startKey, endKey);
+ } catch (RocksDBException e) {
+ throw new IOException("Failed to delete prefix: " + keyPrefix, e);
+ }
+ }
+
+ @Override
+ public List<String> getEntriesByPrefix(String prefix) throws IOException {
+ List<String> result = new ArrayList<>();
+ if (StringUtils.isNullOrWhitespaceOnly(prefix)) {
+ return result;
+ }
+
+ byte[] prefixBytes = prefix.getBytes(UTF_8);
+ byte[] upperBound = computeNextKey(prefixBytes);
+ if (upperBound == null) {
+ throw new IOException("Failed to compute next key for prefix: " + prefix);
+ }
+
+ try (Slice upperBoundSlice = new Slice(upperBound);
+ ReadOptions readOptions = new ReadOptions().setIterateUpperBound(upperBoundSlice);
+ RocksIterator iterator = db.newIterator(readOptions)) {
+ for (iterator.seek(prefixBytes); iterator.isValid(); iterator.next()) {
+ result.add(new String(iterator.value(), UTF_8));
+ }
+ try {
+ iterator.status();
+ } catch (RocksDBException e) {
+ throw new IOException(
+ String.format("Error iterating over RocksDB, prefix: %s", prefix), e);
+ }
+ }
+
+ return result;
+ }
+
+ /**
+ * Compute the next key in lexicographic order.
+ *
+ * @param key the current key
+ * @return the next key, or null if no such key exists
+ */
+ @Nullable
+ private static byte[] computeNextKey(byte[] key) {
+ // Find the last byte that is not 0xFF; everything after it can be dropped.
+ int lastIncrementableIndex = -1;
+ for (int i = key.length - 1; i >= 0; i--) {
+ if ((key[i] & 0xFF) != 0xFF) {
+ lastIncrementableIndex = i;
+ break;
+ }
+ }
+ if (lastIncrementableIndex < 0) {
+ // All bytes are 0xFF.
+ return null;
+ }
+ byte[] nextKey = new byte[lastIncrementableIndex + 1];
+ System.arraycopy(key, 0, nextKey, 0, lastIncrementableIndex + 1);
+ nextKey[lastIncrementableIndex]++;
+ return nextKey;
+ }
+
+ @Override
+ public String readArchiveContent(String entry) throws IOException {
+ return entry;
+ }
+
+ @Override
+ public void close() {
+ Collections.reverse(handlesToClose);
+ handlesToClose.forEach(IOUtils::closeQuietly);
+ handlesToClose.clear();
+
+ if (nativeLibDir != null) {
+ deleteDirectoryQuietly(nativeLibDir);
+ }
+ }
+
+ /**
+ * Loads the RocksDB native library. The default configuration is suitable for most use cases.
+ * Custom options can be supported in the future if needed.
+ */
+ private void loadRocksDBConfiguration() {
+ // Use the full (non-block-based) BloomFilter format with 10 bits/key, which is the
+ // RocksDB-recommended default and yields ~1% false positive rate.
+ // https://github.com/facebook/rocksdb/wiki/RocksDB-Bloom-Filter#full-filters-new-format
+ BloomFilter bloomFilter = new BloomFilter(10.0D, false);
+ handlesToClose.add(bloomFilter);
+
+ BlockBasedTableConfig tableFormatConfig =
+ new BlockBasedTableConfig()
+ .setFilterPolicy(bloomFilter)
+ .setEnableIndexCompression(false)
+ .setIndexBlockRestartInterval(8)
+ .setFormatVersion(5);
+
+ this.dbOptions =
+ new Options()
+ .setCreateIfMissing(true)
+ .setBottommostCompressionType(CompressionType.ZSTD_COMPRESSION)
+ .setCompressionType(CompressionType.LZ4_COMPRESSION)
+ .setTableFormatConfig(tableFormatConfig);
+ handlesToClose.add(dbOptions);
+
+ this.writeOptions = new WriteOptions().setDisableWAL(true);
+ handlesToClose.add(writeOptions);
+ }
+
+ private static void loadRocksDBLibrary(File libDir) throws IOException {
+ try {
+ Files.createDirectories(libDir.toPath());
+ LOG.info("Try to load RocksDB native library to '{}'.", libDir.getAbsolutePath());
+ NativeLibraryLoader.getInstance().loadLibrary(libDir.getAbsolutePath());
+ // Sanity check that the library is fully usable.
+ RocksDB.loadLibrary();
+ LOG.info(
+ "Successfully loaded RocksDB native library to '{}'.",
+ libDir.getAbsolutePath());
+ } catch (Throwable t) {
+ LOG.warn("Failed to load RocksDB native library to '{}'.", libDir.getAbsolutePath(), t);
+ deleteDirectoryQuietly(libDir);
+ throw t;
+ }
+ }
+
+ private static void deleteDirectoryQuietly(File dir) {
+ try {
+ FileUtils.deleteDirectory(dir);
+ } catch (Throwable t) {
+ LOG.warn("Failed to delete directory: {}", dir.getAbsolutePath(), t);
+ }
+ }
+}
diff --git a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/AbstractHistoryServerHandlerTest.java b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/AbstractHistoryServerHandlerTest.java
index 7a7107da5d4..479f3dcb07b 100644
--- a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/AbstractHistoryServerHandlerTest.java
+++ b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/AbstractHistoryServerHandlerTest.java
@@ -37,8 +37,9 @@ import org.slf4j.LoggerFactory;
import java.io.File;
import java.nio.file.Files;
import java.nio.file.Path;
+import java.util.Arrays;
import java.util.Collection;
-import java.util.Collections;
+import java.util.UUID;
import static org.assertj.core.api.Assertions.assertThat;
@@ -62,7 +63,14 @@ public class AbstractHistoryServerHandlerTest {
@Parameters(name = "handlerFactory={0}")
private static Collection<HandlerFactory> handlerFactories() {
HandlerFactory staticFileServerHandlerFactory = HistoryServerStaticFileServerHandler::new;
- return Collections.singletonList(staticFileServerHandlerFactory);
+ HandlerFactory rocksDBHandlerFactory =
+ webDir ->
+ new HistoryServerRocksDBHandler(
+ new RocksDBArchiveStorage(
+ new File(webDir, "rocksdb-" + UUID.randomUUID()),
+ new Configuration()),
+ webDir);
+ return Arrays.asList(staticFileServerHandlerFactory, rocksDBHandlerFactory);
}
@Parameter public HandlerFactory handlerFactory;
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
index 46587a9d58e..a35ccc7e6ce 100644
--- 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
@@ -55,7 +55,8 @@ class ArchiveStorageTest {
@Parameters(name = "storageFactory={0}")
private static Collection<ArchiveStorageFactory<?>> storageFactories() {
ArchiveStorageFactory<File> fileArchiveStorageFactory = FileArchiveStorage::new;
- return List.of(fileArchiveStorageFactory);
+ ArchiveStorageFactory<String> rocksDBStorageFactory = RocksDBArchiveStorage::new;
+ return List.of(fileArchiveStorageFactory, rocksDBStorageFactory);
}
/** Creates an {@link ArchiveStorage} instance under the given temporary directory. */
diff --git a/flink-state-backends/flink-statebackend-rocksdb/pom.xml b/flink-state-backends/flink-statebackend-rocksdb/pom.xml
index de8beed302e..34bce76316f 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/pom.xml
+++ b/flink-state-backends/flink-statebackend-rocksdb/pom.xml
@@ -63,7 +63,7 @@ under the License.
<dependency>
<groupId>com.ververica</groupId>
<artifactId>frocksdbjni</artifactId>
- <version>8.10.0-ververica-1.0</version>
+ <version>${rocksdb.version}</version>
</dependency>
<dependency>
diff --git a/pom.xml b/pom.xml
index 7daba854ea1..7f8ebc59432 100644
--- a/pom.xml
+++ b/pom.xml
@@ -224,6 +224,7 @@ under the License.
<test.unit.pattern>**/*Test.*</test.unit.pattern>
<snappy.java.version>1.1.10.7</snappy.java.version>
<commons-lang3.version>3.18.0</commons-lang3.version>
+ <rocksdb.version>8.10.0-ververica-1.0</rocksdb.version>
</properties>
<dependencies>
(flink) 02/04: [FLINK-39911][historyserver] Extract AbstractHistoryServerHandler from HistoryServerStaticFileServerHandler
Posted by gu...@apache.org.
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 c1f4c5ea64c1862c9c915f5d071e1a455da5a306
Author: Zihao Chen <is...@gmail.com>
AuthorDate: Thu Jun 11 23:42:56 2026 +0800
[FLINK-39911][historyserver] Extract AbstractHistoryServerHandler from HistoryServerStaticFileServerHandler
---
...dler.java => AbstractHistoryServerHandler.java} | 227 +++++++++++---------
.../runtime/webmonitor/history/HistoryServer.java | 5 +-
.../HistoryServerStaticFileServerHandler.java | 228 +--------------------
.../history/AbstractHistoryServerHandlerTest.java | 179 ++++++++++++++++
.../HistoryServerStaticFileServerHandlerTest.java | 95 ---------
.../web-dashboard/src/app/app.interceptor.ts | 2 +-
6 files changed, 328 insertions(+), 408 deletions(-)
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerStaticFileServerHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/AbstractHistoryServerHandler.java
similarity index 62%
copy from flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerStaticFileServerHandler.java
copy to flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/AbstractHistoryServerHandler.java
index 40e5cb379d8..8b93ef01349 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerStaticFileServerHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/AbstractHistoryServerHandler.java
@@ -9,23 +9,16 @@
*
* 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.
+ * 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;
-/**
- * *************************************************************************** This code is based on
- * the "HttpStaticFileServerHandler" from the Netty project's HTTP server example.
- *
- * <p>See http://netty.io and
- * https://github.com/netty/netty/blob/4.0/example/src/main/java/io/netty/example/http/file/HttpStaticFileServerHandler.java
- * ***************************************************************************
- */
import org.apache.flink.runtime.rest.NotFoundException;
import org.apache.flink.runtime.rest.handler.RestHandlerException;
import org.apache.flink.runtime.rest.handler.legacy.files.StaticFileServerHandler;
@@ -60,7 +53,6 @@ import java.io.RandomAccessFile;
import java.net.URI;
import java.net.URL;
import java.nio.file.Files;
-import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Collections;
import java.util.Date;
@@ -74,46 +66,33 @@ import static org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpRes
import static org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
import static org.apache.flink.util.Preconditions.checkNotNull;
-/**
- * Simple file server handler used by the {@link HistoryServer} that serves requests to web
- * frontend's static files, such as HTML, CSS, JS or JSON files.
- *
- * <p>This code is based on the "HttpStaticFileServerHandler" from the Netty project's HTTP server
- * example.
- *
- * <p>This class is a copy of the {@link StaticFileServerHandler}. The differences are that the
- * request path is modified to end on ".json" if it does not have a filename extension; when
- * "index.html" is requested we load "index_hs.html" instead to inject the modified HistoryServer
- * WebInterface and that the caching of the "/joboverview" page is prevented.
- */
+/** Abstract base class for HistoryServer handlers. */
@ChannelHandler.Sharable
-public class HistoryServerStaticFileServerHandler
+public abstract class AbstractHistoryServerHandler<Entry>
extends SimpleChannelInboundHandler<RoutedRequest> {
- /** Default logger, if none is specified. */
- private static final Logger LOG =
- LoggerFactory.getLogger(HistoryServerStaticFileServerHandler.class);
-
- // ------------------------------------------------------------------------
+ private static final Logger LOG = LoggerFactory.getLogger(AbstractHistoryServerHandler.class);
/** The path in which the static documents are. */
- private final File rootPath;
+ protected final File rootPath;
+
+ protected final ArchiveStorage<Entry> archiveStorage;
- public HistoryServerStaticFileServerHandler(File rootPath) throws IOException {
+ protected AbstractHistoryServerHandler(ArchiveStorage<Entry> archiveStorage, File rootPath)
+ throws IOException {
+ this.archiveStorage = archiveStorage;
this.rootPath = checkNotNull(rootPath).getCanonicalFile();
}
- // ------------------------------------------------------------------------
- // Responses to requests
- // ------------------------------------------------------------------------
+ // -------------------------------------------------------------------------
+ // Netty lifecycle
+ // -------------------------------------------------------------------------
@Override
- public void channelRead0(ChannelHandlerContext ctx, RoutedRequest routedRequest)
+ public final void channelRead0(ChannelHandlerContext ctx, RoutedRequest routedRequest)
throws Exception {
- String requestPath = routedRequest.getPath();
-
try {
- respondWithFile(ctx, routedRequest.getRequest(), requestPath);
+ respondToRequest(ctx, routedRequest);
} catch (RestHandlerException rhe) {
HandlerUtils.sendErrorResponse(
ctx,
@@ -124,9 +103,33 @@ public class HistoryServerStaticFileServerHandler
}
}
- /** Response when running with leading JobManager. */
- private void respondWithFile(ChannelHandlerContext ctx, HttpRequest request, String requestPath)
- throws IOException, ParseException, RestHandlerException {
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+ if (ctx.channel().isActive()) {
+ LOG.error("Caught exception", cause);
+ HandlerUtils.sendErrorResponse(
+ ctx,
+ false,
+ new ErrorResponseBody("Internal server error."),
+ INTERNAL_SERVER_ERROR,
+ Collections.emptyMap());
+ }
+ }
+
+ // -------------------------------------------------------------------------
+ // Request handling
+ // -------------------------------------------------------------------------
+
+ /**
+ * Handles HTTP requests by routing to JSON resource or static file response.
+ *
+ * @param ctx Netty Channel context
+ * @param routedRequest The routed HTTP request
+ * @throws Exception Any exception that occurs during processing
+ */
+ protected void respondToRequest(ChannelHandlerContext ctx, RoutedRequest routedRequest)
+ throws Exception {
+ String requestPath = routedRequest.getPath();
// make sure we request the "index.html" in case there is a directory request
if (requestPath.endsWith("/")) {
@@ -135,49 +138,38 @@ public class HistoryServerStaticFileServerHandler
if (!requestPath.contains(".")) { // we assume that the path ends in either .html or .js
requestPath = requestPath + ".json";
- }
- // convert to absolute path
- final File file = new File(rootPath, requestPath);
-
- if (!file.exists()) {
- // file does not exist. Try to load it with the classloader
- ClassLoader cl = HistoryServerStaticFileServerHandler.class.getClassLoader();
-
- try (InputStream resourceStream = cl.getResourceAsStream("web" + requestPath)) {
- boolean success = false;
- try {
- if (resourceStream != null) {
- URL root = cl.getResource("web");
- URL requested = cl.getResource("web" + requestPath);
-
- if (root != null && requested != null) {
- URI rootURI = new URI(root.getPath()).normalize();
- URI requestedURI = new URI(requested.getPath()).normalize();
-
- // Check that we don't load anything from outside of the
- // expected scope.
- if (!rootURI.relativize(requestedURI).equals(requestedURI)) {
- LOG.debug("Loading missing file from classloader: {}", requestPath);
- // ensure that directory to file exists.
- file.getParentFile().mkdirs();
- Files.copy(resourceStream, file.toPath());
-
- success = true;
- }
- }
- }
- } catch (Throwable t) {
- LOG.error("error while responding", t);
- } finally {
- if (!success) {
- LOG.debug("Unable to load requested file {} from classloader", requestPath);
- throw new NotFoundException("File not found.");
- }
- }
+ LOG.debug("Responding to request for path {}", requestPath);
+ Entry resource = loadResource(requestPath);
+
+ if (resource == null) {
+ LOG.debug("Unable to load requested resource {}", requestPath);
+ throw new NotFoundException("Resource not found.");
+ }
+
+ respondWithResource(ctx, routedRequest.getRequest(), requestPath, resource);
+ } else {
+ File destFile = new File(rootPath, requestPath);
+ if (!destFile.exists()) {
+ tryLoadFromClassloader(destFile, requestPath);
}
+ responseWithFile(ctx, routedRequest.getRequest(), requestPath, destFile);
}
+ }
+
+ // -------------------------------------------------------------------------
+ // Response helpers
+ // -------------------------------------------------------------------------
+
+ /** Respond to the request with the resource. */
+ protected abstract void respondWithResource(
+ ChannelHandlerContext ctx, HttpRequest request, String requestPath, Entry resource)
+ throws Exception;
+ /** Respond to the request with the file. */
+ protected void responseWithFile(
+ ChannelHandlerContext ctx, HttpRequest request, String requestPath, File file)
+ throws Exception {
StaticFileServerHandler.checkFileValidity(file, rootPath, LOG);
// cache validation
@@ -267,16 +259,65 @@ public class HistoryServerStaticFileServerHandler
}
}
- @Override
- public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
- if (ctx.channel().isActive()) {
- LOG.error("Caught exception", cause);
- HandlerUtils.sendErrorResponse(
- ctx,
- false,
- new ErrorResponseBody("Internal server error."),
- INTERNAL_SERVER_ERROR,
- Collections.emptyMap());
+ // -------------------------------------------------------------------------
+ // Resource loading
+ // -------------------------------------------------------------------------
+
+ /**
+ * Loads the resource for the given request path from the archive storage.
+ *
+ * @param requestPath The request path
+ * @return The resource for the given request path, or null if not found
+ */
+ private Entry loadResource(String requestPath) throws Exception {
+ String requestKey = requestPath.startsWith("/") ? requestPath.substring(1) : requestPath;
+ if (archiveStorage.exists(requestKey)) {
+ return archiveStorage.getEntry(requestKey);
+ }
+ return null;
+ }
+
+ /**
+ * Attempts to load a missing resource from the classloader's {@code web/} resource directory
+ * and copies it to {@code destFile} on disk.
+ *
+ * @param destFile destination file
+ * @param requestPath relative resource path (e.g. {@code /index.html})
+ */
+ protected void tryLoadFromClassloader(File destFile, String requestPath) throws Exception {
+ ClassLoader cl = AbstractHistoryServerHandler.class.getClassLoader();
+
+ try (InputStream resourceStream = cl.getResourceAsStream("web" + requestPath)) {
+ boolean success = false;
+ try {
+ if (resourceStream != null) {
+ URL root = cl.getResource("web");
+ URL requested = cl.getResource("web" + requestPath);
+
+ if (root != null && requested != null) {
+ URI rootURI = new URI(root.getPath()).normalize();
+ URI requestedURI = new URI(requested.getPath()).normalize();
+
+ // Check that we don't load anything from outside of the
+ // expected scope.
+ if (!rootURI.relativize(requestedURI).equals(requestedURI)) {
+ LOG.debug("Loading missing resource from classloader: {}", requestPath);
+ // ensure that directory to file exists.
+ destFile.getParentFile().mkdirs();
+ Files.copy(resourceStream, destFile.toPath());
+
+ success = true;
+ }
+ }
+ }
+ } catch (Throwable t) {
+ LOG.error("error while responding", t);
+ }
+
+ if (!success) {
+ LOG.debug("Unable to load requested resource {} from classloader", requestPath);
+ throw new NotFoundException("Resource not found.");
+ }
}
}
}
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 8ddcd57a402..ba836be15ef 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
@@ -363,7 +363,10 @@ public class HistoryServer {
new GeneratedLogUrlHandler(
CompletableFuture.completedFuture(pattern))));
- router.addGet("/:*", new HistoryServerStaticFileServerHandler(webDir));
+ router.addGet(
+ "/:*",
+ new HistoryServerStaticFileServerHandler(
+ (FileArchiveStorage) archiveStorage, webDir));
createDashboardConfigFile();
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerStaticFileServerHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerStaticFileServerHandler.java
index 40e5cb379d8..695f8edfffe 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerStaticFileServerHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerStaticFileServerHandler.java
@@ -26,53 +26,14 @@ package org.apache.flink.runtime.webmonitor.history;
* https://github.com/netty/netty/blob/4.0/example/src/main/java/io/netty/example/http/file/HttpStaticFileServerHandler.java
* ***************************************************************************
*/
-import org.apache.flink.runtime.rest.NotFoundException;
-import org.apache.flink.runtime.rest.handler.RestHandlerException;
import org.apache.flink.runtime.rest.handler.legacy.files.StaticFileServerHandler;
-import org.apache.flink.runtime.rest.handler.router.RoutedRequest;
-import org.apache.flink.runtime.rest.handler.util.HandlerUtils;
-import org.apache.flink.runtime.rest.messages.ErrorResponseBody;
-import org.apache.flink.shaded.netty4.io.netty.channel.ChannelFuture;
-import org.apache.flink.shaded.netty4.io.netty.channel.ChannelFutureListener;
import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandler;
import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandlerContext;
-import org.apache.flink.shaded.netty4.io.netty.channel.DefaultFileRegion;
-import org.apache.flink.shaded.netty4.io.netty.channel.SimpleChannelInboundHandler;
-import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.DefaultHttpResponse;
-import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpChunkedInput;
-import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpHeaderValues;
import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpRequest;
-import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponse;
-import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpUtil;
-import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.LastHttpContent;
-import org.apache.flink.shaded.netty4.io.netty.handler.ssl.SslHandler;
-import org.apache.flink.shaded.netty4.io.netty.handler.stream.ChunkedFile;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import java.io.File;
-import java.io.FileNotFoundException;
import java.io.IOException;
-import java.io.InputStream;
-import java.io.RandomAccessFile;
-import java.net.URI;
-import java.net.URL;
-import java.nio.file.Files;
-import java.text.ParseException;
-import java.text.SimpleDateFormat;
-import java.util.Collections;
-import java.util.Date;
-import java.util.Locale;
-
-import static org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpHeaderNames.CONNECTION;
-import static org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpHeaderNames.IF_MODIFIED_SINCE;
-import static org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus.INTERNAL_SERVER_ERROR;
-import static org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus.NOT_FOUND;
-import static org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus.OK;
-import static org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
-import static org.apache.flink.util.Preconditions.checkNotNull;
/**
* Simple file server handler used by the {@link HistoryServer} that serves requests to web
@@ -87,20 +48,17 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
* WebInterface and that the caching of the "/joboverview" page is prevented.
*/
@ChannelHandler.Sharable
-public class HistoryServerStaticFileServerHandler
- extends SimpleChannelInboundHandler<RoutedRequest> {
-
- /** Default logger, if none is specified. */
- private static final Logger LOG =
- LoggerFactory.getLogger(HistoryServerStaticFileServerHandler.class);
+public class HistoryServerStaticFileServerHandler extends AbstractHistoryServerHandler<File> {
// ------------------------------------------------------------------------
- /** The path in which the static documents are. */
- private final File rootPath;
-
public HistoryServerStaticFileServerHandler(File rootPath) throws IOException {
- this.rootPath = checkNotNull(rootPath).getCanonicalFile();
+ this(new FileArchiveStorage(rootPath), rootPath);
+ }
+
+ public HistoryServerStaticFileServerHandler(
+ FileArchiveStorage fileArchiveStorage, File rootPath) throws IOException {
+ super(fileArchiveStorage, rootPath);
}
// ------------------------------------------------------------------------
@@ -108,175 +66,9 @@ public class HistoryServerStaticFileServerHandler
// ------------------------------------------------------------------------
@Override
- public void channelRead0(ChannelHandlerContext ctx, RoutedRequest routedRequest)
+ protected void respondWithResource(
+ ChannelHandlerContext ctx, HttpRequest request, String requestPath, File file)
throws Exception {
- String requestPath = routedRequest.getPath();
-
- try {
- respondWithFile(ctx, routedRequest.getRequest(), requestPath);
- } catch (RestHandlerException rhe) {
- HandlerUtils.sendErrorResponse(
- ctx,
- routedRequest.getRequest(),
- new ErrorResponseBody(rhe.getMessage()),
- rhe.getHttpResponseStatus(),
- Collections.emptyMap());
- }
- }
-
- /** Response when running with leading JobManager. */
- private void respondWithFile(ChannelHandlerContext ctx, HttpRequest request, String requestPath)
- throws IOException, ParseException, RestHandlerException {
-
- // make sure we request the "index.html" in case there is a directory request
- if (requestPath.endsWith("/")) {
- requestPath = requestPath + "index.html";
- }
-
- if (!requestPath.contains(".")) { // we assume that the path ends in either .html or .js
- requestPath = requestPath + ".json";
- }
-
- // convert to absolute path
- final File file = new File(rootPath, requestPath);
-
- if (!file.exists()) {
- // file does not exist. Try to load it with the classloader
- ClassLoader cl = HistoryServerStaticFileServerHandler.class.getClassLoader();
-
- try (InputStream resourceStream = cl.getResourceAsStream("web" + requestPath)) {
- boolean success = false;
- try {
- if (resourceStream != null) {
- URL root = cl.getResource("web");
- URL requested = cl.getResource("web" + requestPath);
-
- if (root != null && requested != null) {
- URI rootURI = new URI(root.getPath()).normalize();
- URI requestedURI = new URI(requested.getPath()).normalize();
-
- // Check that we don't load anything from outside of the
- // expected scope.
- if (!rootURI.relativize(requestedURI).equals(requestedURI)) {
- LOG.debug("Loading missing file from classloader: {}", requestPath);
- // ensure that directory to file exists.
- file.getParentFile().mkdirs();
- Files.copy(resourceStream, file.toPath());
-
- success = true;
- }
- }
- }
- } catch (Throwable t) {
- LOG.error("error while responding", t);
- } finally {
- if (!success) {
- LOG.debug("Unable to load requested file {} from classloader", requestPath);
- throw new NotFoundException("File not found.");
- }
- }
- }
- }
-
- StaticFileServerHandler.checkFileValidity(file, rootPath, LOG);
-
- // cache validation
- final String ifModifiedSince = request.headers().get(IF_MODIFIED_SINCE);
- if (ifModifiedSince != null && !ifModifiedSince.isEmpty()) {
- SimpleDateFormat dateFormatter =
- new SimpleDateFormat(StaticFileServerHandler.HTTP_DATE_FORMAT, Locale.US);
- Date ifModifiedSinceDate = dateFormatter.parse(ifModifiedSince);
-
- // Only compare up to the second because the datetime format we send to the client
- // does not have milliseconds
- long ifModifiedSinceDateSeconds = ifModifiedSinceDate.getTime() / 1000;
- long fileLastModifiedSeconds = file.lastModified() / 1000;
- if (ifModifiedSinceDateSeconds == fileLastModifiedSeconds) {
- if (LOG.isDebugEnabled()) {
- LOG.debug(
- "Responding 'NOT MODIFIED' for file '" + file.getAbsolutePath() + '\'');
- }
-
- StaticFileServerHandler.sendNotModified(ctx);
- return;
- }
- }
-
- if (LOG.isDebugEnabled()) {
- LOG.debug("Responding with file '" + file.getAbsolutePath() + '\'');
- }
-
- // Don't need to close this manually. Netty's DefaultFileRegion will take care of it.
- final RandomAccessFile raf;
- try {
- raf = new RandomAccessFile(file, "r");
- } catch (FileNotFoundException e) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Could not find file {}.", file.getAbsolutePath());
- }
- HandlerUtils.sendErrorResponse(
- ctx,
- request,
- new ErrorResponseBody("File not found."),
- NOT_FOUND,
- Collections.emptyMap());
- return;
- }
-
- try {
- long fileLength = raf.length();
-
- HttpResponse response = new DefaultHttpResponse(HTTP_1_1, OK);
- StaticFileServerHandler.setContentTypeHeader(response, file);
-
- // the job overview should be updated as soon as possible
- if (!requestPath.equals("/joboverview.json")) {
- StaticFileServerHandler.setDateAndCacheHeaders(response, file);
- }
- if (HttpUtil.isKeepAlive(request)) {
- response.headers().set(CONNECTION, HttpHeaderValues.KEEP_ALIVE);
- }
- HttpUtil.setContentLength(response, fileLength);
-
- // write the initial line and the header.
- ctx.write(response);
-
- // write the content.
- ChannelFuture lastContentFuture;
- if (ctx.pipeline().get(SslHandler.class) == null) {
- ctx.write(
- new DefaultFileRegion(raf.getChannel(), 0, fileLength),
- ctx.newProgressivePromise());
- lastContentFuture = ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
- } else {
- lastContentFuture =
- ctx.writeAndFlush(
- new HttpChunkedInput(new ChunkedFile(raf, 0, fileLength, 8192)),
- ctx.newProgressivePromise());
- // HttpChunkedInput will write the end marker (LastHttpContent) for us.
- }
-
- // close the connection, if no keep-alive is needed
- if (!HttpUtil.isKeepAlive(request)) {
- lastContentFuture.addListener(ChannelFutureListener.CLOSE);
- }
- } catch (Exception e) {
- raf.close();
- LOG.error("Failed to serve file.", e);
- throw new RestHandlerException("Internal server error.", INTERNAL_SERVER_ERROR);
- }
- }
-
- @Override
- public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
- if (ctx.channel().isActive()) {
- LOG.error("Caught exception", cause);
- HandlerUtils.sendErrorResponse(
- ctx,
- false,
- new ErrorResponseBody("Internal server error."),
- INTERNAL_SERVER_ERROR,
- Collections.emptyMap());
- }
+ responseWithFile(ctx, request, requestPath, file);
}
}
diff --git a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/AbstractHistoryServerHandlerTest.java b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/AbstractHistoryServerHandlerTest.java
new file mode 100644
index 00000000000..7a7107da5d4
--- /dev/null
+++ b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/AbstractHistoryServerHandlerTest.java
@@ -0,0 +1,179 @@
+/*
+ * 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.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.rest.handler.router.Router;
+import org.apache.flink.runtime.webmonitor.testutils.HttpUtils;
+import org.apache.flink.runtime.webmonitor.utils.WebFrontendBootstrap;
+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 org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Collection;
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Common HTTP-level tests for {@link AbstractHistoryServerHandler} subclasses. New subclasses can
+ * be exercised by adding an entry to {@link #handlerFactories()}.
+ */
+@ExtendWith(ParameterizedTestExtension.class)
+public class AbstractHistoryServerHandlerTest {
+
+ /** Factory that creates a concrete handler bound to the given web directory. */
+ @FunctionalInterface
+ public interface HandlerFactory {
+ AbstractHistoryServerHandler<?> create(File webDir) throws Exception;
+ }
+
+ /**
+ * Provides the concrete {@link AbstractHistoryServerHandler} implementations under test. To
+ * cover an additional handler simply add another factory here.
+ */
+ @Parameters(name = "handlerFactory={0}")
+ private static Collection<HandlerFactory> handlerFactories() {
+ HandlerFactory staticFileServerHandlerFactory = HistoryServerStaticFileServerHandler::new;
+ return Collections.singletonList(staticFileServerHandlerFactory);
+ }
+
+ @Parameter public HandlerFactory handlerFactory;
+
+ @TempDir private Path tmpDir;
+
+ private Path webDir;
+ private AbstractHistoryServerHandler<?> handler;
+ private WebFrontendBootstrap webUI;
+ private String baseUrl;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ webDir = Files.createDirectory(tmpDir.resolve("webDir"));
+ final Path uploadDir = Files.createDirectory(tmpDir.resolve("uploadDir"));
+
+ handler = handlerFactory.create(webDir.toFile());
+ Router<?> router = new Router().addGet("/:*", handler);
+ webUI =
+ new WebFrontendBootstrap(
+ router,
+ LoggerFactory.getLogger(getClass()),
+ uploadDir.toFile(),
+ null,
+ "localhost",
+ 0,
+ new Configuration());
+ baseUrl = "http://localhost:" + webUI.getServerPort();
+ }
+
+ @AfterEach
+ void tearDown() {
+ if (webUI != null) {
+ webUI.shutdown();
+ }
+ }
+
+ /**
+ * Tests requests against static files served from the {@code web/} resources packaged with the
+ * handler module:
+ *
+ * <ul>
+ * <li>a request for {@code /} is rewritten to {@code /index.html};
+ * <li>{@code /index.html} is loaded via the classloader fallback when not present on disk;
+ * <li>a missing static file (e.g. {@code /hello.html}) results in 404.
+ * </ul>
+ */
+ @TestTemplate
+ void testRespondWithStaticFile() throws Exception {
+ // /index.html is loaded from the classloader (web/index.html) and served
+ Tuple2<Integer, String> index = HttpUtils.getFromHTTP(baseUrl + "/index.html");
+ assertThat(index.f0).isEqualTo(200);
+ assertThat(index.f1).contains("Apache Flink Web Dashboard");
+
+ // a trailing slash is rewritten to "/index.html"
+ Tuple2<Integer, String> index2 = HttpUtils.getFromHTTP(baseUrl + "/");
+ assertThat(index2).isEqualTo(index);
+
+ // a static file that is neither on disk nor in the classloader yields 404
+ Tuple2<Integer, String> missing = HttpUtils.getFromHTTP(baseUrl + "/hello.html");
+ assertThat(missing.f0).isEqualTo(404);
+ assertThat(missing.f1).contains("not found");
+ }
+
+ /**
+ * Tests file-system safety checks performed by {@code responseWithFile}:
+ *
+ * <ul>
+ * <li>a directory request returns 405;
+ * <li>a path that escapes the {@code webDir} returns 403.
+ * </ul>
+ */
+ @TestTemplate
+ void testRespondWithFile() throws Exception {
+ // requesting a directory rather than a file yields 405
+ Files.createDirectory(webDir.resolve("dir.json"));
+ Tuple2<Integer, String> dirNotFound = HttpUtils.getFromHTTP(baseUrl + "/dir.json");
+ assertThat(dirNotFound.f0).isEqualTo(405);
+ assertThat(dirNotFound.f1).contains("not found");
+
+ // requesting a file outside of the webDir is rejected with 403
+ Files.createFile(tmpDir.resolve("secret"));
+ Tuple2<Integer, String> outsideWebDir = HttpUtils.getFromHTTP(baseUrl + "/../secret");
+ assertThat(outsideWebDir.f0).isEqualTo(403);
+ assertThat(outsideWebDir.f1).contains("Forbidden");
+ }
+
+ /**
+ * Tests requests served via the {@link ArchiveStorage} resource.
+ *
+ * <ul>
+ * <li>an existing entry is returned with its content;
+ * <li>a missing entry results in 404.
+ * </ul>
+ */
+ @TestTemplate
+ void testRespondWithResource() throws Exception {
+ String resourcePath = "overviews/job1.json";
+ String resourceContent = "{\"job\":\"job1\"}";
+ handler.archiveStorage.putArchiveContent(resourcePath, resourceContent);
+
+ // request without an extension: handler will append ".json" and serve the
+ // entry from the ArchiveStorage via respondWithResource
+ Tuple2<Integer, String> resource = HttpUtils.getFromHTTP(baseUrl + "/overviews/job1");
+ assertThat(resource.f0).isEqualTo(200);
+ assertThat(resource.f1).isEqualTo(resourceContent);
+
+ // request a missing resource: archiveStorage returns null and the handler
+ // responds 404
+ Tuple2<Integer, String> missing = HttpUtils.getFromHTTP(baseUrl + "/hello");
+ assertThat(missing.f0).isEqualTo(404);
+ assertThat(missing.f1).contains("not found");
+ }
+}
diff --git a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerStaticFileServerHandlerTest.java b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerStaticFileServerHandlerTest.java
deleted file mode 100644
index d908cf633b8..00000000000
--- a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerStaticFileServerHandlerTest.java
+++ /dev/null
@@ -1,95 +0,0 @@
-/*
- * 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.java.tuple.Tuple2;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.runtime.rest.handler.router.Router;
-import org.apache.flink.runtime.webmonitor.testutils.HttpUtils;
-import org.apache.flink.runtime.webmonitor.utils.WebFrontendBootstrap;
-
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.io.TempDir;
-import org.slf4j.LoggerFactory;
-
-import java.nio.file.Files;
-import java.nio.file.Path;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-/** Tests for the HistoryServerStaticFileServerHandler. */
-class HistoryServerStaticFileServerHandlerTest {
-
- @Test
- void testRespondWithFile(@TempDir Path tmpDir) throws Exception {
- final Path webDir = Files.createDirectory(tmpDir.resolve("webDir"));
- final Path uploadDir = Files.createDirectory(tmpDir.resolve("uploadDir"));
-
- Router router =
- new Router()
- .addGet("/:*", new HistoryServerStaticFileServerHandler(webDir.toFile()));
- WebFrontendBootstrap webUI =
- new WebFrontendBootstrap(
- router,
- LoggerFactory.getLogger(HistoryServerStaticFileServerHandlerTest.class),
- uploadDir.toFile(),
- null,
- "localhost",
- 0,
- new Configuration());
-
- int port = webUI.getServerPort();
- try {
- // verify that 404 message is returned when requesting a non-existent file
- Tuple2<Integer, String> notFound404 =
- HttpUtils.getFromHTTP("http://localhost:" + port + "/hello");
- assertThat(notFound404.f0).isEqualTo(404);
- assertThat(notFound404.f1).contains("not found");
-
- // verify that a) a file can be loaded using the ClassLoader and b) that the
- // HistoryServer
- // index_hs.html is injected
- Tuple2<Integer, String> index =
- HttpUtils.getFromHTTP("http://localhost:" + port + "/index.html");
- assertThat(index.f0).isEqualTo(200);
- assertThat(index.f1).contains("Apache Flink Web Dashboard");
-
- // verify that index.html is appended if the request path ends on '/'
- Tuple2<Integer, String> index2 =
- HttpUtils.getFromHTTP("http://localhost:" + port + "/");
- assertThat(index2).isEqualTo(index);
-
- // verify that a 405 message is returned when requesting a directory
- Files.createDirectory(webDir.resolve("dir.json"));
- Tuple2<Integer, String> dirNotFound =
- HttpUtils.getFromHTTP("http://localhost:" + port + "/dir");
- assertThat(dirNotFound.f0).isEqualTo(405);
- assertThat(dirNotFound.f1).contains("not found");
-
- // verify that a 403 message is returned when requesting a file outside the webDir
- Files.createFile(tmpDir.resolve("secret"));
- Tuple2<Integer, String> dirOutsideDirectory =
- HttpUtils.getFromHTTP("http://localhost:" + port + "/../secret");
- assertThat(dirOutsideDirectory.f0).isEqualTo(403);
- assertThat(dirOutsideDirectory.f1).contains("Forbidden");
- } finally {
- webUI.shutdown();
- }
- }
-}
diff --git a/flink-runtime-web/web-dashboard/src/app/app.interceptor.ts b/flink-runtime-web/web-dashboard/src/app/app.interceptor.ts
index f7bbc12f9dd..cfc85d20091 100644
--- a/flink-runtime-web/web-dashboard/src/app/app.interceptor.ts
+++ b/flink-runtime-web/web-dashboard/src/app/app.interceptor.ts
@@ -41,7 +41,7 @@ export class AppInterceptor implements HttpInterceptor {
intercept(req: HttpRequest<unknown>, next: HttpHandler): Observable<HttpEvent<unknown>> {
// Error response from below url should be ignored
const ignoreErrorUrlEndsList = ['checkpoints/config', 'checkpoints'];
- const ignoreErrorMessage = ['File not found.'];
+ const ignoreErrorMessage = ['File not found.', 'Resource not found.'];
const option: NzNotificationDataOptions = {
nzDuration: 0,
nzStyle: { width: 'auto', 'white-space': 'pre-wrap' }
(flink) 01/04: [FLINK-39911][historyserver] Introduce ArchiveStorage abstraction and FileArchiveStorage implementation
Posted by gu...@apache.org.
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 <is...@gmail.com>
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());
+ }
+}
(flink) 04/04: [FLINK-39911][docs] Document HistoryServer archive storage backend
Posted by gu...@apache.org.
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 0fbd749cacc244a247f331a2a06bb6d347f964ba
Author: Zihao Chen <is...@gmail.com>
AuthorDate: Fri Jun 12 11:26:33 2026 +0800
[FLINK-39911][docs] Document HistoryServer archive storage backend
---
docs/content.zh/docs/deployment/advanced/historyserver.md | 13 +++++++++++++
docs/content/docs/deployment/advanced/historyserver.md | 13 +++++++++++++
2 files changed, 26 insertions(+)
diff --git a/docs/content.zh/docs/deployment/advanced/historyserver.md b/docs/content.zh/docs/deployment/advanced/historyserver.md
index 974cb8c4eed..748dbd867fc 100644
--- a/docs/content.zh/docs/deployment/advanced/historyserver.md
+++ b/docs/content.zh/docs/deployment/advanced/historyserver.md
@@ -85,6 +85,19 @@ historyserver.archive.fs.refresh-interval: 10000
请查看配置页面以获取[配置选项的完整列表]({{< ref "docs/deployment/config" >}}#history-server)。
+**存储后端**
+
+HistoryServer 支持通过 `historyserver.archive.storage.type` 选择本地存档存储后端：
+
+* `FILE`（默认）：每个存档解压为多个 JSON 文件，存放在 `historyserver.web.tmpdir` 下。
+* `ROCKSDB`：所有存档以 KV 形式存储在 `historyserver.web.tmpdir/rocksdb` 的单个 RocksDB 实例中，可避免"大量小文件"问题，适合存档数量较多的场景。
+
+启用 RocksDB 后端示例：
+
+```yaml
+historyserver.archive.storage.type: ROCKSDB
+```
+
## 日志集成
Flink 本身并不提供已完成作业的日志收集功能。
diff --git a/docs/content/docs/deployment/advanced/historyserver.md b/docs/content/docs/deployment/advanced/historyserver.md
index 76eaa7c386d..607f08b826b 100644
--- a/docs/content/docs/deployment/advanced/historyserver.md
+++ b/docs/content/docs/deployment/advanced/historyserver.md
@@ -79,6 +79,19 @@ The contained archives are downloaded and cached in the local filesystem. The lo
Check out the configuration page for a [complete list of configuration options]({{< ref "docs/deployment/config" >}}#history-server).
+**Archive Storage Backend**
+
+The HistoryServer supports pluggable local archive storage backends, selected via `historyserver.archive.storage.type`:
+
+* `FILE` (default): Each archive is unpacked into multiple JSON files under `historyserver.web.tmpdir`.
+* `ROCKSDB`: All archives are stored as key-value pairs in a single embedded RocksDB instance under `historyserver.web.tmpdir/rocksdb`. This avoids the "many small files" problem and is recommended when there is a large number of archives.
+
+Example for enabling the RocksDB backend:
+
+```yaml
+historyserver.archive.storage.type: ROCKSDB
+```
+
## Log Integration
Flink does not provide built-in methods for archiving logs of completed jobs.