argoyal2212 commented on code in PR #29008:
URL: https://github.com/apache/flink/pull/29008#discussion_r4264148846
##########
flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java:
##########
@@ -240,9 +282,20 @@ void scanArchives(
&& processExpiredArchiveDeletion) {
events.addAll(cleanupExpiredArchives(archivesToRemove));
}
- // clean remote and local
+ if (!archivesExpiredByTtl.isEmpty()) {
+ // clean remote and local; TTL expiry is unaffected by
retainRemoteBeyondLocalLimit
+
events.addAll(cleanupArchivesBeyondRetainedLimit(archivesExpiredByTtl));
+ }
if (!archivesBeyondRetainedLimit.isEmpty()) {
-
events.addAll(cleanupArchivesBeyondRetainedLimit(archivesBeyondRetainedLimit));
+ if (retainRemoteBeyondLocalLimit) {
+ // clean local only; the remote archive is left in place
and remains
+ // fetchable on demand (e.g. via LAZY archive load mode).
+ events.addAll(
+
cleanupLocalArchivesBeyondRetainedLimit(archivesBeyondRetainedLimit));
+ } else {
+ // clean remote and local
+
events.addAll(cleanupArchivesBeyondRetainedLimit(archivesBeyondRetainedLimit));
+ }
}
Review Comment:
Took this. Eviction goes through archivesToEvictLocally and the existing
cleanupExpiredArchives, the flag-off path matches master, and the extra helper
is removed.
##########
flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcherTest.java:
##########
@@ -369,6 +397,104 @@ void testScanArchivesWithoutFetch() throws Exception {
assertThat(archiveStorage.exists("overviews/" + jobId +
".json")).isFalse();
}
+ @TestTemplate
+ void testArchivesBeyondRetainedLimitAreDeletedFromRemoteByDefault() throws
Exception {
+ JobID retainedJobId = JobID.generate();
+ JobID beyondLimitJobId = JobID.generate();
+ Path beyondLimitArchivePath =
+ createJobArchive(remoteArchiveRootPath, beyondLimitJobId,
true);
+ createJobArchive(remoteArchiveRootPath, retainedJobId, true);
+
+ // retain only the archive belonging to retainedJobId, regardless of
file ordering
+
org.apache.flink.runtime.webmonitor.history.retaining.ArchiveRetainedStrategy
+ retainOnlyRetainedJob =
+ (file, index) ->
file.getPath().getName().equals(retainedJobId.toString());
+
+ HistoryServerArchiveFetcher<?> fetcher =
+ createArchiveFetcher(
+ remoteArchiveRootPath, true, archiveStorage,
retainOnlyRetainedJob, false);
+
+ fetcher.fetchArchives(EAGER);
+
+
assertThat(beyondLimitArchivePath.getFileSystem().exists(beyondLimitArchivePath))
+ .as("archive beyond the retained limit should be deleted from
remote by default")
+ .isFalse();
+ }
+
+ @TestTemplate
+ void testArchivesBeyondRetainedLimitAreKeptRemotelyWhenConfigured() throws
Exception {
+ JobID retainedJobId = JobID.generate();
+ JobID beyondLimitJobId = JobID.generate();
+ Path beyondLimitArchivePath =
+ createJobArchive(remoteArchiveRootPath, beyondLimitJobId,
true);
+ createJobArchive(remoteArchiveRootPath, retainedJobId, true);
+
+ // retain only the archive belonging to retainedJobId, regardless of
file ordering
+
org.apache.flink.runtime.webmonitor.history.retaining.ArchiveRetainedStrategy
+ retainOnlyRetainedJob =
+ (file, index) ->
file.getPath().getName().equals(retainedJobId.toString());
+
+ HistoryServerArchiveFetcher<?> fetcher =
+ createArchiveFetcher(
+ remoteArchiveRootPath, true, archiveStorage,
retainOnlyRetainedJob, true);
+
+ fetcher.fetchArchives(EAGER);
+
+ // remote archive beyond the limit must still exist ...
+
assertThat(beyondLimitArchivePath.getFileSystem().exists(beyondLimitArchivePath))
+ .as(
+ "archive beyond the retained limit must not be deleted
from remote when "
+ + "retainRemoteBeyondLocalLimit is enabled")
+ .isTrue();
+ // ... but must not have been processed/cached locally
+ assertThat(archiveStorage.exists("overviews/" + beyondLimitJobId +
".json")).isFalse();
+
+ // and it must still be fetchable on demand
+ fetcher.lazyFetchArchiveProactively(beyondLimitJobId.toString(),
beyondLimitArchivePath);
+ waitForArchiveLoaded(archiveMetaInfoCache,
beyondLimitJobId.toString());
+ assertThat(archiveStorage.exists("overviews/" + beyondLimitJobId +
".json")).isTrue();
+ }
Review Comment:
Added testViewedArchiveBeyondLimitSurvivesNextScan, which rescans and checks
no events are emitted and the on-demand copy survives. I did not add a separate
TTL or application layout case, since the TTL path no longer has special logic.
##########
flink-core/src/main/java/org/apache/flink/configuration/HistoryServerOptions.java:
##########
@@ -178,6 +178,34 @@ public class HistoryServerOptions {
.text(LEGACY_NOTE_MESSAGE)
.build());
+ /**
+ * If this option is enabled, job archives that fall outside {@link
+ * #HISTORY_SERVER_RETAINED_JOBS} are no longer processed/refreshed
locally, but are kept in the
+ * remote archive directory instead of being deleted. They remain
reachable on demand (e.g. by
+ * directly requesting {@code /jobs/<jobId>} in {@link
HistoryServerArchiveLoadMode#LAZY}
+ * mode).
+ */
+ public static final ConfigOption<Boolean>
HISTORY_SERVER_RETAIN_REMOTE_BEYOND_LOCAL_LIMIT =
+ key("historyserver.archive.retain-remote-beyond-local-limit")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ Description.builder()
+ .text(
+ "Whether job archives beyond the
limit configured by %s should still be "
+ + "retained in the remote
archive directory defined by %s, instead of being "
+ + "deleted. ",
+
code(HISTORY_SERVER_RETAINED_JOBS_KEY),
+
code(HISTORY_SERVER_ARCHIVE_DIRS.key()))
+ .text(
+ "When enabled, such archives are
no longer polled/processed locally, but remain "
+ + "fetchable on demand
when %s is set to %s. ",
+
code("historyserver.archive.load.mode"), code("LAZY"))
+ .text(
+ "This option has no effect unless
%s is set to a value other than %s. ",
+
code(HISTORY_SERVER_RETAINED_JOBS_KEY), code("-1"))
Review Comment:
Added to the option description (unbounded remote growth, pair with
retained-ttl) and added a section to historyserver.md in en and zh.
##########
flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcherTest.java:
##########
@@ -369,6 +397,104 @@ void testScanArchivesWithoutFetch() throws Exception {
assertThat(archiveStorage.exists("overviews/" + jobId +
".json")).isFalse();
}
+ @TestTemplate
+ void testArchivesBeyondRetainedLimitAreDeletedFromRemoteByDefault() throws
Exception {
+ JobID retainedJobId = JobID.generate();
+ JobID beyondLimitJobId = JobID.generate();
+ Path beyondLimitArchivePath =
+ createJobArchive(remoteArchiveRootPath, beyondLimitJobId,
true);
+ createJobArchive(remoteArchiveRootPath, retainedJobId, true);
+
+ // retain only the archive belonging to retainedJobId, regardless of
file ordering
+
org.apache.flink.runtime.webmonitor.history.retaining.ArchiveRetainedStrategy
+ retainOnlyRetainedJob =
+ (file, index) ->
file.getPath().getName().equals(retainedJobId.toString());
+
+ HistoryServerArchiveFetcher<?> fetcher =
+ createArchiveFetcher(
+ remoteArchiveRootPath, true, archiveStorage,
retainOnlyRetainedJob, false);
+
+ fetcher.fetchArchives(EAGER);
+
+
assertThat(beyondLimitArchivePath.getFileSystem().exists(beyondLimitArchivePath))
+ .as("archive beyond the retained limit should be deleted from
remote by default")
+ .isFalse();
+ }
+
+ @TestTemplate
+ void testArchivesBeyondRetainedLimitAreKeptRemotelyWhenConfigured() throws
Exception {
+ JobID retainedJobId = JobID.generate();
+ JobID beyondLimitJobId = JobID.generate();
+ Path beyondLimitArchivePath =
+ createJobArchive(remoteArchiveRootPath, beyondLimitJobId,
true);
+ createJobArchive(remoteArchiveRootPath, retainedJobId, true);
+
+ // retain only the archive belonging to retainedJobId, regardless of
file ordering
+
org.apache.flink.runtime.webmonitor.history.retaining.ArchiveRetainedStrategy
+ retainOnlyRetainedJob =
+ (file, index) ->
file.getPath().getName().equals(retainedJobId.toString());
+
+ HistoryServerArchiveFetcher<?> fetcher =
Review Comment:
Fixed. The new tests assign the fetcher field, so tearDown closes them.
##########
flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java:
##########
@@ -368,6 +421,27 @@ List<ArchiveEvent>
cleanupArchivesBeyondRetainedLimit(Map<Path, Set<Path>> archi
return cleanupExpiredArchives(allArchiveIdsToRemove);
}
+ /**
+ * Cleans up archives beyond {@link
HistoryServerOptions#HISTORY_SERVER_RETAINED_JOBS} from the
+ * local cache only. Unlike {@link #cleanupArchivesBeyondRetainedLimit},
the remote archive is
+ * left untouched so that it remains fetchable on demand, e.g. via {@link
+ * HistoryServerOptions.HistoryServerArchiveLoadMode#LAZY} mode.
+ */
+ List<ArchiveEvent> cleanupLocalArchivesBeyondRetainedLimit(
+ Map<Path, Set<Path>> archivesToRemove) {
+ Map<Path, Set<String>> allArchiveIdsToRemove = new HashMap<>();
+
+ for (Map.Entry<Path, Set<Path>> pathSetEntry :
archivesToRemove.entrySet()) {
+ HashSet<String> archiveIdsToRemove = new HashSet<>();
+ for (Path archive : pathSetEntry.getValue()) {
+ archiveIdsToRemove.add(archive.getName());
+ }
+ allArchiveIdsToRemove.put(pathSetEntry.getKey(),
archiveIdsToRemove);
+ }
+
+ return cleanupExpiredArchives(allArchiveIdsToRemove);
Review Comment:
Scan eviction and most recently viewed updates now run under one lock, and a
viewed job is never evicted by a scan. A viewed job can only be cancelled when
it is the least recently viewed one beyond the configured limit.
--
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]
