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 <[email protected]> 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. + */ [email protected] +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>
