spuru9 commented on code in PR #29008:
URL: https://github.com/apache/flink/pull/29008#discussion_r4262582408
##########
flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServer.java:
##########
@@ -301,7 +306,8 @@ public HistoryServer(
archiveStorage,
archiveMetaInfoCache,
lazyFetchExecutorCommonPoolSize,
- lazyFetchExecutorIndividualPoolSize);
+ lazyFetchExecutorIndividualPoolSize,
+ retainRemoteBeyondLocalLimit);
applicationArchiveFetcher =
new HistoryServerApplicationArchiveFetcher<>(
Review Comment:
major: Since 2.3 (FLINK-38761), every job is archived under
`applications/<app>/jobs/`, and the app fetcher never gets this flag. So it
only covers archives from 2.2 and older JMs, and app jobs can't be fetched on
demand anyway, because `lazyFetchArchiveProactively` only looks in
`<refresh-dir>/<job-id>`.
##########
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:
minor: Goes with the comment on L259-272. Change L234 to `Map<Path,
Set<String>> archivesToEvictLocally`; then the flag-off path matches master and
`cleanupLocalArchivesBeyondRetainedLimit` can go.
```suggestion
// clean remote and local
if (!archivesBeyondRetainedLimit.isEmpty()) {
events.addAll(cleanupArchivesBeyondRetainedLimit(archivesBeyondRetainedLimit));
}
// clean local only, the remote archive is kept
if (!archivesToEvictLocally.isEmpty()) {
events.addAll(cleanupExpiredArchives(archivesToEvictLocally));
}
```
##########
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:
minor: A second scan here would have caught the re-cleaning in
`scanArchives`. The application layout and the real `isExpiredByTtl` aren't
tested either.
```suggestion
// and it must still be fetchable on demand
fetcher.lazyFetchArchiveProactively(beyondLimitJobId.toString(),
beyondLimitArchivePath);
waitForArchiveLoaded(archiveMetaInfoCache,
beyondLimitJobId.toString());
assertThat(archiveStorage.exists("overviews/" + beyondLimitJobId +
".json")).isTrue();
// the next scan must neither re-emit events for it nor wipe the
on-demand copy
archiveEvents.clear();
fetcher.fetchArchives(EAGER);
assertThat(archiveEvents)
.extracting(HistoryServerArchiveFetcher.ArchiveEvent::getId)
.doesNotContain(beyondLimitJobId.toString());
assertThat(archiveStorage.exists("overviews/" + beyondLimitJobId +
".json")).isTrue();
}
```
##########
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:
minor: After a rebase, these locals hide the `fetcher` field that
FLINK-40567 added, so `tearDown` never closes them and the `@TempDir` cleanup
race comes back. Assign the field instead (same at L413 and L484).
##########
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")
Review Comment:
major: FLIP-505 was voted with `cached-retained-jobs` and
`num-cached-most-recently-viewed-jobs`, not this key. Can you get this agreed
on dev@ (the "Reopening FLIP-505" thread got no replies), or switch to the
voted keys? #26878 is also still open on the same JIRA.
##########
flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java:
##########
@@ -223,9 +257,17 @@ void scanArchives(
fileOrderedIndexOnModifiedTime++;
if (!retainedStrategy.shouldRetain(archive,
fileOrderedIndexOnModifiedTime)) {
- archivesBeyondRetainedLimit
- .computeIfAbsent(refreshDir, ignored -> new
HashSet<>())
- .add(archivePath);
+ if (retainedStrategy.isExpiredByTtl(archive)) {
+ // TTL expiry always applies remote deletion,
regardless of
+ // retainRemoteBeyondLocalLimit, which only
concerns the count limit.
+ archivesExpiredByTtl
+ .computeIfAbsent(refreshDir, ignored ->
new HashSet<>())
+ .add(archivePath);
+ } else {
+ archivesBeyondRetainedLimit
+ .computeIfAbsent(refreshDir, ignored ->
new HashSet<>())
+ .add(archivePath);
+ }
continue;
}
Review Comment:
major: Archives past the limit stay in the remote dir, so every scan cleans
all of them again (an INFO log, 3 deletes and a `DELETED` event each). That
same pass also wipes or cancels whatever was fetched on demand. Only evict what
an earlier scan cached:
```suggestion
if (!retainedStrategy.shouldRetain(archive,
fileOrderedIndexOnModifiedTime)) {
if (retainRemoteBeyondLocalLimit
&&
!retainedStrategy.isExpiredByTtl(archive)) {
// the remote copy stays; only evict what an
earlier scan cached
if (cachedArchivesPerRefreshDirectory
.get(refreshDir)
.contains(archiveId)) {
archivesToEvictLocally
.computeIfAbsent(refreshDir, ignored
-> new HashSet<>())
.add(archiveId);
}
} else {
archivesBeyondRetainedLimit
.computeIfAbsent(refreshDir, ignored ->
new HashSet<>())
.add(archivePath);
}
continue;
}
```
##########
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:
minor: Worth stating the cost in the description: the remote dir then grows
without limit and gets listed on every refresh, so pair it with `retained-ttl`.
A line in `historyserver.md` (en and zh) would help too.
--
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]
