spuru9 commented on code in PR #29008:
URL: https://github.com/apache/flink/pull/29008#discussion_r4464666219
##########
flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java:
##########
@@ -669,6 +731,42 @@ void lazyFetchArchiveProactively(String jobId, @Nullable
Path archivePath) throw
}
}
+ /** Marks the job as recently viewed and evicts the least recently viewed
ones locally. */
+ void markRecentlyViewed(String jobId) {
+ if (!isOwnedByRefreshDirectory(jobId)) {
+ return;
+ }
+ List<ArchiveEvent> events = new ArrayList<>();
+ synchronized (mostRecentlyViewedJobs) {
+ mostRecentlyViewedJobs.put(jobId, Boolean.TRUE);
Review Comment:
major: Jobs inside the cached window also take a slot here. With the default
of 1, opening any recent job deletes the old job I'd just opened. Shouldn't
this only track jobs the scan doesn't already cache?
##########
flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java:
##########
@@ -229,6 +271,17 @@ void scanArchives(
continue;
}
+ if (cachedRetainedJobs > 0
+ && fileOrderedIndexOnModifiedTime >
cachedRetainedJobs) {
Review Comment:
major: Jobs past `cached-retained-jobs` never get an overview, so they're
missing from the job list and you can only reach them by URL. Is that intended?
If not, keeping the overviews and evicting only `jobs/<id>/` would fix it. If
it is, the docs should say so.
##########
flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java:
##########
@@ -669,6 +731,42 @@ void lazyFetchArchiveProactively(String jobId, @Nullable
Path archivePath) throw
}
}
+ /** Marks the job as recently viewed and evicts the least recently viewed
ones locally. */
+ void markRecentlyViewed(String jobId) {
+ if (!isOwnedByRefreshDirectory(jobId)) {
+ return;
+ }
+ List<ArchiveEvent> events = new ArrayList<>();
+ synchronized (mostRecentlyViewedJobs) {
+ mostRecentlyViewedJobs.put(jobId, Boolean.TRUE);
+ Iterator<String> eldest =
mostRecentlyViewedJobs.keySet().iterator();
+ while (mostRecentlyViewedJobs.size() >
numCachedMostRecentlyViewedJobs) {
+ String evictedId = eldest.next();
+ eldest.remove();
+ // archives still inside the cached window are managed by the
scan
+ if (cachedArchivesPerRefreshDirectory.values().stream()
+ .noneMatch(ids -> ids.contains(evictedId))) {
+ cleanUpLazyFetchTask(evictedId);
Review Comment:
major: If one request is still waiting on job A's fetch and someone opens
another old job, this evicts A and cancels the fetch, so the first request
fails with a 500. I hit this in minikube with two users on two old jobs. Could
we skip jobs whose fetch is still running and leave the delete to the scan
thread?
##########
flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java:
##########
@@ -669,6 +731,42 @@ void lazyFetchArchiveProactively(String jobId, @Nullable
Path archivePath) throw
}
}
+ /** Marks the job as recently viewed and evicts the least recently viewed
ones locally. */
+ void markRecentlyViewed(String jobId) {
+ if (!isOwnedByRefreshDirectory(jobId)) {
+ return;
+ }
+ List<ArchiveEvent> events = new ArrayList<>();
+ synchronized (mostRecentlyViewedJobs) {
+ mostRecentlyViewedJobs.put(jobId, Boolean.TRUE);
+ Iterator<String> eldest =
mostRecentlyViewedJobs.keySet().iterator();
+ while (mostRecentlyViewedJobs.size() >
numCachedMostRecentlyViewedJobs) {
+ String evictedId = eldest.next();
+ eldest.remove();
+ // archives still inside the cached window are managed by the
scan
+ if (cachedArchivesPerRefreshDirectory.values().stream()
+ .noneMatch(ids -> ids.contains(evictedId))) {
+ cleanUpLazyFetchTask(evictedId);
+ events.add(deleteJobFiles(evictedId));
Review Comment:
major: `cancel(true)` doesn't stop the detail-write loop in `fetchArchive`,
so it keeps writing after this delete and the delete fails with
`DirectoryNotEmptyException`. In my run that left orphan files for a job
nothing tracks anymore, and they were still served. Could that loop stop once
the thread is interrupted?
##########
flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java:
##########
@@ -669,6 +731,42 @@ void lazyFetchArchiveProactively(String jobId, @Nullable
Path archivePath) throw
}
}
+ /** Marks the job as recently viewed and evicts the least recently viewed
ones locally. */
+ void markRecentlyViewed(String jobId) {
+ if (!isOwnedByRefreshDirectory(jobId)) {
+ return;
+ }
+ List<ArchiveEvent> events = new ArrayList<>();
+ synchronized (mostRecentlyViewedJobs) {
+ mostRecentlyViewedJobs.put(jobId, Boolean.TRUE);
+ Iterator<String> eldest =
mostRecentlyViewedJobs.keySet().iterator();
+ while (mostRecentlyViewedJobs.size() >
numCachedMostRecentlyViewedJobs) {
+ String evictedId = eldest.next();
+ eldest.remove();
+ // archives still inside the cached window are managed by the
scan
+ if (cachedArchivesPerRefreshDirectory.values().stream()
+ .noneMatch(ids -> ids.contains(evictedId))) {
+ cleanUpLazyFetchTask(evictedId);
+ events.add(deleteJobFiles(evictedId));
+ }
+ }
+ }
+ if (!events.isEmpty()) {
+ updateOverview();
+ events.forEach(archiveEventListener);
+ }
+ }
+
+ /** Application-owned jobs are nested deeper and managed by the
application fetcher. */
+ private boolean isOwnedByRefreshDirectory(String jobId) {
+ ArchiveMetaInfo metaInfo = archiveMetaInfoCache.get(jobId);
+ if (metaInfo == null || metaInfo.getArchivePath() == null) {
+ return false;
+ }
+ Path parent = metaInfo.getArchivePath().getParent();
+ return refreshDirs.stream().anyMatch(dir ->
dir.getPath().equals(parent));
Review Comment:
minor: The refresh dir is kept exactly as configured, but scanned archive
paths are fully qualified. With `/archives` or `hdfs:///...` this is always
false, so a viewed job still gets evicted by the next scan. `file:///archives`
worked for me and `/archives` didn't, so comparing qualified paths should fix
it.
##########
flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcherTest.java:
##########
@@ -420,6 +456,134 @@ void testScanArchivesWithoutFetch() throws Exception {
assertThat(archiveStorage.exists("overviews/" + jobId +
".json")).isFalse();
}
+ @TestTemplate
+ void testArchivesBeyondCachedLimitAreOnlyCachedLocallyWithinLimit() throws
Exception {
Review Comment:
minor: All the new tests scan in EAGER and call the lazy methods by hand, so
the real LAZY path through the handler isn't tested. Could you add a LAZY case,
maybe in `AbstractHistoryServerHandlerTest` since it already has the setup?
##########
flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java:
##########
@@ -669,6 +731,42 @@ void lazyFetchArchiveProactively(String jobId, @Nullable
Path archivePath) throw
}
}
+ /** Marks the job as recently viewed and evicts the least recently viewed
ones locally. */
+ void markRecentlyViewed(String jobId) {
+ if (!isOwnedByRefreshDirectory(jobId)) {
Review Comment:
minor: This also runs when `cached-retained-jobs` isn't set, so a job opened
by URL before the next scan gets evicted as soon as you open another one.
```suggestion
if (cachedRetainedJobs <= 0 || !isOwnedByRefreshDirectory(jobId)) {
```
--
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]
