reswqa commented on code in PR #28685: URL: https://github.com/apache/flink/pull/28685#discussion_r3549334331
########## docs/content/docs/deployment/advanced/historyserver.md: ########## @@ -92,6 +92,24 @@ Example for enabling the RocksDB backend: historyserver.archive.storage.type: ROCKSDB ``` +**Archive Load Mode** + +The HistoryServer supports two modes for loading archives, selected via `historyserver.archive.load.mode`: + +* `EAGER` (default): Archives are automatically and synchronously downloaded to local storage via periodic background refreshes. +* `LAZY`: Archives are displayed on the web interface immediately, while the underlying data is fetched asynchronously in the background. If a specific archive is accessed before its background download completes, the system will prioritize and fetch it on demand. + +Example for enabling the lazy loading: + +```yaml +historyserver.archive.load.mode: LAZY +``` + +In `LAZY` mode, on-demand fetches use two thread pools: + +* `historyserver.lazy.fetch.executor.common.pool-size` (default: `4`) — The size of the shared thread pool used for routine, background archive fetching. Review Comment: I suggest remove the default value here, because we might forget to sync the documentation when changing default values. ########## flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerApplicationArchiveFetcherTest.java: ########## @@ -0,0 +1,408 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.runtime.webmonitor.history; + +import org.apache.flink.api.common.ApplicationID; +import org.apache.flink.api.common.JobID; +import org.apache.flink.core.fs.Path; +import org.apache.flink.runtime.history.ArchivePathUtils; +import org.apache.flink.runtime.history.FsJsonArchivist; +import org.apache.flink.runtime.messages.webmonitor.ApplicationDetails; +import org.apache.flink.runtime.messages.webmonitor.MultipleApplicationsDetails; +import org.apache.flink.runtime.rest.messages.ApplicationsOverviewHeaders; +import org.apache.flink.testutils.junit.extensions.parameterized.Parameter; +import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension; +import org.apache.flink.testutils.junit.extensions.parameterized.Parameters; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.TestTemplate; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.api.io.TempDir; + +import java.io.File; +import java.nio.file.Files; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; + +import static org.apache.flink.configuration.ClusterOptions.CLUSTER_ID; +import static org.apache.flink.configuration.HistoryServerOptions.HistoryServerArchiveLoadMode.EAGER; +import static org.apache.flink.configuration.HistoryServerOptions.HistoryServerArchiveLoadMode.LAZY; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerArchiveFetcher.ArchiveEventType.CREATED; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerArchiveFetcher.ArchiveEventType.DETAIL_PARSING; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerArchiveFetcher.ArchiveEventType.OVERVIEW_CREATED; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerTestUtils.OBJECT_MAPPER; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerTestUtils.RETAIN_ALL; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerTestUtils.createJobArchive; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerTestUtils.createRefreshLocation; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerTestUtils.waitForArchiveLoaded; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Tests for {@link HistoryServerApplicationArchiveFetcher}. Only application-specific behaviours + * that are NOT covered by {@link HistoryServerArchiveFetcherTest} are tested here. + */ +@ExtendWith(ParameterizedTestExtension.class) +class HistoryServerApplicationArchiveFetcherTest { + + private static final String DEFAULT_CLUSTER_ID = CLUSTER_ID.defaultValue(); + + @TempDir File remoteArchiveRootPath; + @TempDir File localArchiveRootPath; + + @Parameter public ArchiveStorageTest.ArchiveStorageFactory<Object> storageFactory; + + private ArchiveStorage<Object> archiveStorage; + + private Map<String, ArchiveMetaInfo> jobMetaInfoCache; + private Map<String, ArchiveMetaInfo> applicationMetaInfoCache; + private List<HistoryServerArchiveFetcher.ArchiveEvent> archiveEvents; + + @Parameters(name = "storageFactory={0}") + private static Collection<ArchiveStorageTest.ArchiveStorageFactory<?>> storageFactories() { + ArchiveStorageTest.ArchiveStorageFactory<File> fileArchiveStorageFactory = + FileArchiveStorage::new; + ArchiveStorageTest.ArchiveStorageFactory<String> rocksDBStorageFactory = + RocksDBArchiveStorage::new; + return List.of(fileArchiveStorageFactory, rocksDBStorageFactory); + } + + /** Creates an {@link ArchiveStorage} instance under the given temporary directory. */ + @FunctionalInterface + interface ArchiveStorageFactory<T> { Review Comment: This is unused. ########## flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java: ########## @@ -458,4 +524,183 @@ void updateJobOverview() { LOG.error("Failed to update job overview.", e); } } + + // -------------------------------- Lazy Load ---------------------------------------- + List<ArchiveEvent> lazyProcessArchive(String archiveId, Path archivePath, Path refreshDir) + throws Exception { + return Collections.singletonList(lazyProcessJobArchive(archiveId, archivePath, false)); + } + + ArchiveEvent lazyProcessJobArchive(String jobId, Path jobArchive, boolean individual) + throws Exception { + if (!archiveMetaInfoCache.containsKey(jobId)) { + archiveMetaInfoCache.put(jobId, new ArchiveMetaInfo(jobId, PENDING, jobArchive)); + } else { + return new ArchiveEvent(jobId, archiveMetaInfoCache.get(jobId).getEventType()); + } + + ExecutorService fetchExecutor; + Map<String, Future<?>> fetchTasks; + if (individual) { + fetchExecutor = individualFetchExecutor; + fetchTasks = individualFetchTasks; + } else { + fetchExecutor = commonFetchExecutor; + fetchTasks = commonFetchTasks; + } + + archiveMetaInfoCache.get(jobId).setEventType(ArchiveEventType.OVERVIEW_PARSING); + + Collection<ArchivedJson> archivedJsons = FsJsonArchivist.readArchivedJsons(jobArchive); + List<ArchivedJson> detailArchives = new ArrayList<>(); + boolean overviewCreated = false; + + for (ArchivedJson archive : archivedJsons) { + String path = archive.getPath(); + String json = archive.getJson(); + + if (path.equals(JobsOverviewHeaders.URL)) { + String key = JOB_OVERVIEWS_KEY_PREFIX + jobId + JSON_FILE_ENDING; + archiveStorage.putArchiveContent(key, json); + overviewCreated = true; + } else if (path.equals("/joboverview")) { // legacy path + LOG.debug("Migrating legacy archive {}", jobArchive); + json = convertLegacyJobOverview(json); + String key = JOB_OVERVIEWS_KEY_PREFIX + jobId + JSON_FILE_ENDING; + archiveStorage.putArchiveContent(key, json); + overviewCreated = true; + } else if (path.equals("/jobs/" + jobId)) { + String key = JOBS_KEY_PREFIX + jobId + JSON_FILE_ENDING; + archiveStorage.putArchiveContent(key, json); + } else { + detailArchives.add(archive); + } + } + + if (!overviewCreated && detailArchives.isEmpty()) { + archiveMetaInfoCache.remove(jobId); + throw new RuntimeException("Archive of job " + jobId + " is empty"); Review Comment: This `RuntimeException` will propagate to the `catch (Exception e)` block in the `scanArchives` method. Is this as expected? ########## flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/ArchiveMetaInfo.java: ########## @@ -0,0 +1,54 @@ +/* + * 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.core.fs.Path; + +/** Meta info for archived job. */ +public class ArchiveMetaInfo { + + private final String archiveId; + private HistoryServerArchiveFetcher.ArchiveEventType eventType; + private final Path archivePath; + + public ArchiveMetaInfo( + String archiveId, + HistoryServerArchiveFetcher.ArchiveEventType eventType, + Path archivePath) { + this.archiveId = archiveId; + this.eventType = eventType; + this.archivePath = archivePath; + } + + public String getArchiveId() { Review Comment: Never used method. ########## flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServer.java: ########## @@ -264,38 +271,53 @@ public HistoryServer( Files.createDirectories(webDir.toPath().resolve(APPLICATIONS_SUBDIR)); Files.createDirectories(webDir.toPath().resolve(APPLICATION_OVERVIEWS_SUBDIR)); archiveStorage = new FileArchiveStorage(webDir); - historyServerHandler = - new HistoryServerStaticFileServerHandler( - (FileArchiveStorage) archiveStorage, webDir); + historyServerHandlerFactory = + createFileHandlerFactory((FileArchiveStorage) archiveStorage, webDir); break; case ROCKSDB: File dbPath = new File(webDir, "rocksdb-" + UUID.randomUUID()); Files.createDirectories(dbPath.toPath()); archiveStorage = new RocksDBArchiveStorage(dbPath, config); - historyServerHandler = - new HistoryServerRocksDBHandler( - (RocksDBArchiveStorage) archiveStorage, webDir); + historyServerHandlerFactory = + createRocksDBHandlerFactory((RocksDBArchiveStorage) archiveStorage, webDir); break; default: throw new FlinkException("Unsupported archive storage type: " + archiveStorageType); } + Map<String, ArchiveMetaInfo> archiveMetaInfoCache = new HashMap<>(); Review Comment: Is this map thread safe? it's better to check the race condition. ########## flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java: ########## @@ -132,6 +162,17 @@ public ArchiveEventType getType() { } checkNotNull(webDir); this.archiveStorage = archiveStorage; + this.archiveMetaInfoCache = archiveMetaInfoCache; + this.commonFetchExecutor = + Executors.newFixedThreadPool( + lazyFetchExecutorCommonPoolSize, + new ExecutorThreadFactory("HistoryServer-commonFetchExecutor")); + this.individualFetchExecutor = Review Comment: Where are these two thread pools being shut-down? ########## flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcherTest.java: ########## @@ -0,0 +1,430 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.runtime.webmonitor.history; + +import org.apache.flink.api.common.JobID; +import org.apache.flink.core.fs.Path; +import org.apache.flink.runtime.history.FsJsonArchivist; +import org.apache.flink.runtime.messages.webmonitor.JobDetails; +import org.apache.flink.runtime.messages.webmonitor.MultipleJobsDetails; +import org.apache.flink.testutils.junit.extensions.parameterized.Parameter; +import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension; +import org.apache.flink.testutils.junit.extensions.parameterized.Parameters; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.TestTemplate; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.api.io.TempDir; + +import java.io.File; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.TimeUnit; + +import static org.apache.flink.configuration.HistoryServerOptions.HistoryServerArchiveLoadMode.EAGER; +import static org.apache.flink.configuration.HistoryServerOptions.HistoryServerArchiveLoadMode.LAZY; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerArchiveFetcher.ArchiveEventType.CREATED; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerArchiveFetcher.ArchiveEventType.DETAIL_PARSING; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerArchiveFetcher.ArchiveEventType.OVERVIEW_CREATED; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerTestUtils.OBJECT_MAPPER; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerTestUtils.RETAIN_ALL; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerTestUtils.createJobArchive; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerTestUtils.createLegacyArchive; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerTestUtils.createRefreshLocation; +import static org.apache.flink.runtime.webmonitor.history.HistoryServerTestUtils.waitForArchiveLoaded; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for {@link HistoryServerArchiveFetcher}. */ +@ExtendWith(ParameterizedTestExtension.class) +class HistoryServerArchiveFetcherTest { + + @TempDir File remoteArchiveRootPath; + @TempDir File localArchiveRootPath; + + @Parameter public ArchiveStorageTest.ArchiveStorageFactory<Object> storageFactory; + + private ArchiveStorage<Object> archiveStorage; + private Map<String, ArchiveMetaInfo> archiveMetaInfoCache; + private List<HistoryServerArchiveFetcher.ArchiveEvent> archiveEvents; + + @Parameters(name = "storageFactory={0}") + private static Collection<ArchiveStorageTest.ArchiveStorageFactory<?>> storageFactories() { + ArchiveStorageTest.ArchiveStorageFactory<File> fileArchiveStorageFactory = + FileArchiveStorage::new; + ArchiveStorageTest.ArchiveStorageFactory<String> rocksDBStorageFactory = + RocksDBArchiveStorage::new; + return List.of(fileArchiveStorageFactory, rocksDBStorageFactory); + } + + /** Creates an {@link ArchiveStorage} instance under the given temporary directory. */ + @FunctionalInterface + interface ArchiveStorageFactory<T> { Review Comment: Can be removed. ########## flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServer.java: ########## @@ -462,11 +529,11 @@ private static String createConfigJson(DashboardConfiguration dashboardConfigura } /** Container for the {@link Path} and {@link FileSystem} of a refresh directory. */ - static class RefreshLocation { + public static class RefreshLocation { Review Comment: What is the purpose of making these change? ########## flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java: ########## @@ -458,4 +524,183 @@ void updateJobOverview() { LOG.error("Failed to update job overview.", e); } } + + // -------------------------------- Lazy Load ---------------------------------------- + List<ArchiveEvent> lazyProcessArchive(String archiveId, Path archivePath, Path refreshDir) + throws Exception { + return Collections.singletonList(lazyProcessJobArchive(archiveId, archivePath, false)); + } + + ArchiveEvent lazyProcessJobArchive(String jobId, Path jobArchive, boolean individual) + throws Exception { + if (!archiveMetaInfoCache.containsKey(jobId)) { + archiveMetaInfoCache.put(jobId, new ArchiveMetaInfo(jobId, PENDING, jobArchive)); Review Comment: Is this thread-safe? `containsKey + put` is not an atomic operation. I think we should add some comments to declare the thread-safe semantic for `archiveMetaInfoCache`. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
