You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ch...@apache.org on 2021/10/18 07:44:18 UTC
[flink] branch release-1.14 updated (32d9309 -> 891cc24)
This is an automated email from the ASF dual-hosted git repository.
chesnay pushed a change to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 32d9309 [FLINK-23704][test] Remove unused CollectorDataOutput class
new 026a7aa [FLINK-17914][hs] Keep archives if listStatus fails
new 018ca16 [hotfix] Refactor JobArchiveFetcherTask#run
new 0649bc9 [FLINK-17914] Restructure HistoryServer
new 891cc24 [FLINK-17914][tests] Increase chances of concurrent deletions
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../runtime/webmonitor/history/HistoryServer.java | 30 +-
.../history/HistoryServerArchiveFetcher.java | 449 ++++++++++-----------
.../webmonitor/history/HistoryServerTest.java | 4 +-
3 files changed, 245 insertions(+), 238 deletions(-)
[flink] 04/04: [FLINK-17914][tests] Increase chances of concurrent
deletions
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chesnay pushed a commit to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 891cc24c9147866260bc4e575d8d4fb51c952d7d
Author: Chesnay Schepler <ch...@apache.org>
AuthorDate: Fri Oct 8 13:03:21 2021 +0200
[FLINK-17914][tests] Increase chances of concurrent deletions
---
.../org/apache/flink/runtime/webmonitor/history/HistoryServer.java | 5 +++++
.../apache/flink/runtime/webmonitor/history/HistoryServerTest.java | 4 +++-
2 files changed, 8 insertions(+), 1 deletion(-)
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 25f4125..7f7a170 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
@@ -257,6 +257,11 @@ public class HistoryServer {
return netty.getServerPort();
}
+ @VisibleForTesting
+ void fetchArchives() {
+ executor.execute(getArchiveFetchingRunnable());
+ }
+
public void run() {
try {
start();
diff --git a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerTest.java b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerTest.java
index 54108e0..4e3a529 100644
--- a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerTest.java
+++ b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerTest.java
@@ -327,7 +327,9 @@ public class HistoryServerTest extends TestLogger {
new IllegalStateException(
"Expected at least one existing job"));
- // delete one archive from jm
+ // trigger another fetch and delete one archive from jm
+ // we fetch again to probabilistically cause a concurrent deletion
+ hs.fetchArchives();
Files.deleteIfExists(jmDirectory.toPath().resolve(jobIdToDelete));
assertTrue(firstArchiveExpiredLatch.await(10L, TimeUnit.SECONDS));
[flink] 02/04: [hotfix] Refactor JobArchiveFetcherTask#run
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chesnay pushed a commit to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 018ca16220ca2d3916913a0001a22f8413edfc5f
Author: Chesnay Schepler <ch...@apache.org>
AuthorDate: Mon Oct 4 13:50:09 2021 +0200
[hotfix] Refactor JobArchiveFetcherTask#run
---
.../history/HistoryServerArchiveFetcher.java | 131 ++++++++++++---------
1 file changed, 74 insertions(+), 57 deletions(-)
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java
index c9cd0a4..a547aff 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java
@@ -217,13 +217,12 @@ class HistoryServerArchiveFetcher {
Map<Path, Set<Path>> archivesBeyondSizeLimit = new HashMap<>();
for (HistoryServer.RefreshLocation refreshLocation : refreshDirs) {
Path refreshDir = refreshLocation.getPath();
- FileSystem refreshFS = refreshLocation.getFs();
LOG.debug("Checking archive directory {}.", refreshDir);
// contents of /:refreshDir
FileStatus[] jobArchives;
try {
- jobArchives = refreshFS.listStatus(refreshDir);
+ jobArchives = listArchives(refreshLocation.getFs(), refreshDir);
} catch (IOException e) {
LOG.error(
"Failed to access job archive location for path {}.",
@@ -234,27 +233,12 @@ class HistoryServerArchiveFetcher {
jobsToRemove.remove(refreshDir);
continue;
}
- if (jobArchives == null) {
- // the entire refreshDirectory was removed
- continue;
- }
-
- Arrays.sort(
- jobArchives,
- Comparator.comparingLong(FileStatus::getModificationTime).reversed());
int historySize = 0;
for (FileStatus jobArchive : jobArchives) {
Path jobArchivePath = jobArchive.getPath();
String jobID = jobArchivePath.getName();
- try {
- JobID.fromHexString(jobID);
- } catch (IllegalArgumentException iae) {
- LOG.debug(
- "Archive directory {} contained file with unexpected name {}. Ignoring file.",
- refreshDir,
- jobID,
- iae);
+ if (!isValidJobID(jobID, refreshDir)) {
continue;
}
@@ -275,45 +259,7 @@ class HistoryServerArchiveFetcher {
} else {
LOG.info("Processing archive {}.", jobArchivePath);
try {
- for (ArchivedJson archive :
- FsJobArchivist.getArchivedJsons(jobArchive.getPath())) {
- String path = archive.getPath();
- String json = archive.getJson();
-
- File target;
- if (path.equals(JobsOverviewHeaders.URL)) {
- target = new File(webOverviewDir, jobID + JSON_FILE_ENDING);
- } else if (path.equals("/joboverview")) { // legacy path
- LOG.debug("Migrating legacy archive {}", jobArchivePath);
- json = convertLegacyJobOverview(json);
- target = new File(webOverviewDir, jobID + JSON_FILE_ENDING);
- } else {
- // this implicitly writes into webJobDir
- target = new File(webDir, path + JSON_FILE_ENDING);
- }
-
- java.nio.file.Path parent = target.getParentFile().toPath();
-
- try {
- Files.createDirectories(parent);
- } catch (FileAlreadyExistsException ignored) {
- // there may be left-over directories from the previous
- // attempt
- }
-
- java.nio.file.Path targetPath = target.toPath();
-
- // We overwrite existing files since this may be another attempt
- // at fetching this archive.
- // Existing files may be incomplete/corrupt.
- Files.deleteIfExists(targetPath);
-
- Files.createFile(target.toPath());
- try (FileWriter fw = new FileWriter(target)) {
- fw.write(json);
- fw.flush();
- }
- }
+ processArchive(jobID, jobArchivePath);
events.add(new ArchiveEvent(jobID, ArchiveEventType.CREATED));
cachedArchivesPerRefreshDirectory.get(refreshDir).add(jobID);
LOG.info("Processing archive {} finished.", jobArchivePath);
@@ -345,6 +291,77 @@ class HistoryServerArchiveFetcher {
}
}
+ private static FileStatus[] listArchives(FileSystem refreshFS, Path refreshDir)
+ throws IOException {
+ // contents of /:refreshDir
+ FileStatus[] jobArchives = refreshFS.listStatus(refreshDir);
+ if (jobArchives == null) {
+ // the entire refreshDirectory was removed
+ return new FileStatus[0];
+ }
+
+ Arrays.sort(
+ jobArchives,
+ Comparator.comparingLong(FileStatus::getModificationTime).reversed());
+
+ return jobArchives;
+ }
+
+ private static boolean isValidJobID(String jobId, Path refreshDir) {
+ try {
+ JobID.fromHexString(jobId);
+ return true;
+ } catch (IllegalArgumentException iae) {
+ LOG.debug(
+ "Archive directory {} contained file with unexpected name {}. Ignoring file.",
+ refreshDir,
+ jobId,
+ iae);
+ return false;
+ }
+ }
+
+ private void processArchive(String jobID, Path jobArchive) throws IOException {
+ for (ArchivedJson archive : FsJobArchivist.getArchivedJsons(jobArchive)) {
+ String path = archive.getPath();
+ String json = archive.getJson();
+
+ File target;
+ if (path.equals(JobsOverviewHeaders.URL)) {
+ target = new File(webOverviewDir, jobID + JSON_FILE_ENDING);
+ } else if (path.equals("/joboverview")) { // legacy path
+ LOG.debug("Migrating legacy archive {}", jobArchive);
+ json = convertLegacyJobOverview(json);
+ target = new File(webOverviewDir, jobID + JSON_FILE_ENDING);
+ } else {
+ // this implicitly writes into webJobDir
+ target = new File(webDir, path + JSON_FILE_ENDING);
+ }
+
+ java.nio.file.Path parent = target.getParentFile().toPath();
+
+ try {
+ Files.createDirectories(parent);
+ } catch (FileAlreadyExistsException ignored) {
+ // there may be left-over directories from the previous
+ // attempt
+ }
+
+ java.nio.file.Path targetPath = target.toPath();
+
+ // We overwrite existing files since this may be another attempt
+ // at fetching this archive.
+ // Existing files may be incomplete/corrupt.
+ Files.deleteIfExists(targetPath);
+
+ Files.createFile(target.toPath());
+ try (FileWriter fw = new FileWriter(target)) {
+ fw.write(json);
+ fw.flush();
+ }
+ }
+ }
+
private List<ArchiveEvent> cleanupJobsBeyondSizeLimit(
Map<Path, Set<Path>> jobArchivesToRemove) {
Map<Path, Set<String>> allJobIdsToRemoveFromOverview = new HashMap<>();
[flink] 03/04: [FLINK-17914] Restructure HistoryServer
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chesnay pushed a commit to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 0649bc9c60f7ee5e3f9198bdb815f333f7aeb412
Author: Chesnay Schepler <ch...@apache.org>
AuthorDate: Fri Oct 8 09:57:49 2021 +0200
[FLINK-17914] Restructure HistoryServer
Moves the fetcher scheduling to history server and merge HistoryServerArchiveFetcher and JobArchiveFetcherTask.
This way the HistoryServer takes care of everything required for a long-running application, while the fetcher is entirely focused on a single fetch.
---
.../runtime/webmonitor/history/HistoryServer.java | 25 +-
.../history/HistoryServerArchiveFetcher.java | 483 +++++++++------------
2 files changed, 230 insertions(+), 278 deletions(-)
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 384dda0..25f4125 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
@@ -35,12 +35,16 @@ import org.apache.flink.runtime.rest.messages.DashboardConfiguration;
import org.apache.flink.runtime.security.SecurityConfiguration;
import org.apache.flink.runtime.security.SecurityUtils;
import org.apache.flink.runtime.util.EnvironmentInformation;
+import org.apache.flink.runtime.util.Runnables;
import org.apache.flink.runtime.webmonitor.utils.WebFrontendBootstrap;
import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.ExecutorUtils;
+import org.apache.flink.util.FatalExitExceptionHandler;
import org.apache.flink.util.FileUtils;
import org.apache.flink.util.FlinkException;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.ShutdownHookUtil;
+import org.apache.flink.util.concurrent.ExecutorThreadFactory;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
@@ -60,6 +64,9 @@ import java.util.List;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
@@ -102,6 +109,11 @@ public class HistoryServer {
@Nullable private final SSLHandlerFactory serverSSLFactory;
private WebFrontendBootstrap netty;
+ private final long refreshIntervalMillis;
+ private final ScheduledExecutorService executor =
+ Executors.newSingleThreadScheduledExecutor(
+ new ExecutorThreadFactory("Flink-HistoryServer-ArchiveFetcher"));
+
private final Object startupShutdownLock = new Object();
private final AtomicBoolean shutdownRequested = new AtomicBoolean(false);
private final Thread shutdownHook;
@@ -219,7 +231,7 @@ public class HistoryServer {
"Failed to validate any of the configured directories to monitor.");
}
- long refreshIntervalMillis =
+ refreshIntervalMillis =
config.getLong(HistoryServerOptions.HISTORY_SERVER_ARCHIVE_REFRESH_INTERVAL);
int maxHistorySize = config.getInteger(HistoryServerOptions.HISTORY_SERVER_RETAINED_JOBS);
if (maxHistorySize == 0 || maxHistorySize < -1) {
@@ -229,7 +241,6 @@ public class HistoryServer {
}
archiveFetcher =
new HistoryServerArchiveFetcher(
- refreshIntervalMillis,
refreshDirs,
webDir,
jobArchiveEventListener,
@@ -273,7 +284,8 @@ public class HistoryServer {
createDashboardConfigFile();
- archiveFetcher.start();
+ executor.scheduleWithFixedDelay(
+ getArchiveFetchingRunnable(), 0, refreshIntervalMillis, TimeUnit.MILLISECONDS);
netty =
new WebFrontendBootstrap(
@@ -281,6 +293,11 @@ public class HistoryServer {
}
}
+ private Runnable getArchiveFetchingRunnable() {
+ return Runnables.withUncaughtExceptionHandler(
+ () -> archiveFetcher.fetchArchives(), FatalExitExceptionHandler.INSTANCE);
+ }
+
void stop() {
if (shutdownRequested.compareAndSet(false, true)) {
synchronized (startupShutdownLock) {
@@ -292,7 +309,7 @@ public class HistoryServer {
LOG.warn("Error while shutting down WebFrontendBootstrap.", t);
}
- archiveFetcher.stop();
+ ExecutorUtils.gracefulShutdown(1, TimeUnit.SECONDS, executor);
try {
LOG.info("Removing web dashboard root cache directory {}", webDir);
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java
index a547aff..eb5a34b 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java
@@ -29,10 +29,7 @@ import org.apache.flink.runtime.history.FsJobArchivist;
import org.apache.flink.runtime.messages.webmonitor.JobDetails;
import org.apache.flink.runtime.messages.webmonitor.MultipleJobsDetails;
import org.apache.flink.runtime.rest.messages.JobsOverviewHeaders;
-import org.apache.flink.runtime.util.Runnables;
-import org.apache.flink.util.FatalExitExceptionHandler;
import org.apache.flink.util.FileUtils;
-import org.apache.flink.util.concurrent.ExecutorThreadFactory;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonFactory;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator;
@@ -58,10 +55,6 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
-import java.util.TimerTask;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -72,6 +65,10 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
* intervals, defined by {@link HistoryServerOptions#HISTORY_SERVER_ARCHIVE_REFRESH_INTERVAL}.
*
* <p>The archives are downloaded and expanded into a file structure analog to the REST API.
+ *
+ * <p>Removes existing archives from these directories and the cache if configured by {@link
+ * HistoryServerOptions#HISTORY_SERVER_CLEANUP_EXPIRED_JOBS} or {@link
+ * HistoryServerOptions#HISTORY_SERVER_RETAINED_JOBS}.
*/
class HistoryServerArchiveFetcher {
@@ -109,28 +106,44 @@ class HistoryServerArchiveFetcher {
private static final JsonFactory jacksonFactory = new JsonFactory();
private static final ObjectMapper mapper = new ObjectMapper();
- private final ScheduledExecutorService executor =
- Executors.newSingleThreadScheduledExecutor(
- new ExecutorThreadFactory("Flink-HistoryServer-ArchiveFetcher"));
- private final JobArchiveFetcherTask fetcherTask;
- private final long refreshIntervalMillis;
+ private static final String JSON_FILE_ENDING = ".json";
+
+ private final List<HistoryServer.RefreshLocation> refreshDirs;
+ private final Consumer<ArchiveEvent> jobArchiveEventListener;
+ private final boolean processExpiredArchiveDeletion;
+ private final boolean processBeyondLimitArchiveDeletion;
+ private final int maxHistorySize;
+
+ /** Cache of all available jobs identified by their id. */
+ private final Map<Path, Set<String>> cachedArchivesPerRefreshDirectory;
+
+ private final File webDir;
+ private final File webJobDir;
+ private final File webOverviewDir;
HistoryServerArchiveFetcher(
- long refreshIntervalMillis,
List<HistoryServer.RefreshLocation> refreshDirs,
File webDir,
Consumer<ArchiveEvent> jobArchiveEventListener,
boolean cleanupExpiredArchives,
int maxHistorySize)
throws IOException {
- this.refreshIntervalMillis = refreshIntervalMillis;
- this.fetcherTask =
- new JobArchiveFetcherTask(
- refreshDirs,
- webDir,
- jobArchiveEventListener,
- cleanupExpiredArchives,
- maxHistorySize);
+ this.refreshDirs = checkNotNull(refreshDirs);
+ this.jobArchiveEventListener = jobArchiveEventListener;
+ this.processExpiredArchiveDeletion = cleanupExpiredArchives;
+ this.maxHistorySize = maxHistorySize;
+ this.processBeyondLimitArchiveDeletion = this.maxHistorySize > 0;
+ this.cachedArchivesPerRefreshDirectory = new HashMap<>();
+ for (HistoryServer.RefreshLocation refreshDir : refreshDirs) {
+ cachedArchivesPerRefreshDirectory.put(refreshDir.getPath(), new HashSet<>());
+ }
+ this.webDir = checkNotNull(webDir);
+ this.webJobDir = new File(webDir, "jobs");
+ Files.createDirectories(webJobDir.toPath());
+ this.webOverviewDir = new File(webDir, "overviews");
+ Files.createDirectories(webOverviewDir.toPath());
+ updateJobOverview(webOverviewDir, webDir);
+
if (LOG.isInfoEnabled()) {
for (HistoryServer.RefreshLocation refreshDir : refreshDirs) {
LOG.info("Monitoring directory {} for archived jobs.", refreshDir.getPath());
@@ -138,296 +151,218 @@ class HistoryServerArchiveFetcher {
}
}
- void start() {
- final Runnable guardedTask =
- Runnables.withUncaughtExceptionHandler(
- fetcherTask, FatalExitExceptionHandler.INSTANCE);
- executor.scheduleWithFixedDelay(
- guardedTask, 0, refreshIntervalMillis, TimeUnit.MILLISECONDS);
- }
-
- void stop() {
- executor.shutdown();
-
+ void fetchArchives() {
try {
- if (!executor.awaitTermination(1, TimeUnit.SECONDS)) {
- executor.shutdownNow();
- }
- } catch (InterruptedException ignored) {
- executor.shutdownNow();
- }
- }
-
- /**
- * {@link TimerTask} that polls the directories configured as {@link
- * HistoryServerOptions#HISTORY_SERVER_ARCHIVE_DIRS} for new job archives. Removes existing
- * archives from these directories and the cache if configured by {@link
- * HistoryServerOptions#HISTORY_SERVER_CLEANUP_EXPIRED_JOBS} or {@link
- * HistoryServerOptions#HISTORY_SERVER_RETAINED_JOBS}
- */
- static class JobArchiveFetcherTask extends TimerTask {
-
- private final List<HistoryServer.RefreshLocation> refreshDirs;
- private final Consumer<ArchiveEvent> jobArchiveEventListener;
- private final boolean processExpiredArchiveDeletion;
- private final boolean processBeyondLimitArchiveDeletion;
- private final int maxHistorySize;
-
- /** Cache of all available jobs identified by their id. */
- private final Map<Path, Set<String>> cachedArchivesPerRefreshDirectory;
-
- private final File webDir;
- private final File webJobDir;
- private final File webOverviewDir;
-
- private static final String JSON_FILE_ENDING = ".json";
-
- JobArchiveFetcherTask(
- List<HistoryServer.RefreshLocation> refreshDirs,
- File webDir,
- Consumer<ArchiveEvent> jobArchiveEventListener,
- boolean processExpiredArchiveDeletion,
- int maxHistorySize)
- throws IOException {
- this.refreshDirs = checkNotNull(refreshDirs);
- this.jobArchiveEventListener = jobArchiveEventListener;
- this.processExpiredArchiveDeletion = processExpiredArchiveDeletion;
- this.maxHistorySize = maxHistorySize;
- this.processBeyondLimitArchiveDeletion = this.maxHistorySize > 0;
- this.cachedArchivesPerRefreshDirectory = new HashMap<>();
- for (HistoryServer.RefreshLocation refreshDir : refreshDirs) {
- cachedArchivesPerRefreshDirectory.put(refreshDir.getPath(), new HashSet<>());
- }
- this.webDir = checkNotNull(webDir);
- this.webJobDir = new File(webDir, "jobs");
- Files.createDirectories(webJobDir.toPath());
- this.webOverviewDir = new File(webDir, "overviews");
- Files.createDirectories(webOverviewDir.toPath());
- updateJobOverview(webOverviewDir, webDir);
- }
+ LOG.debug("Starting archive fetching.");
+ List<ArchiveEvent> events = new ArrayList<>();
+ Map<Path, Set<String>> jobsToRemove = new HashMap<>();
+ cachedArchivesPerRefreshDirectory.forEach(
+ (path, archives) -> jobsToRemove.put(path, new HashSet<>(archives)));
+ Map<Path, Set<Path>> archivesBeyondSizeLimit = new HashMap<>();
+ for (HistoryServer.RefreshLocation refreshLocation : refreshDirs) {
+ Path refreshDir = refreshLocation.getPath();
+ LOG.debug("Checking archive directory {}.", refreshDir);
+
+ // contents of /:refreshDir
+ FileStatus[] jobArchives;
+ try {
+ jobArchives = listArchives(refreshLocation.getFs(), refreshDir);
+ } catch (IOException e) {
+ LOG.error("Failed to access job archive location for path {}.", refreshDir, e);
+ // something went wrong, potentially due to a concurrent deletion
+ // do not remove any jobs now; we will retry later
+ jobsToRemove.remove(refreshDir);
+ continue;
+ }
- @Override
- public void run() {
- try {
- LOG.debug("Starting archive fetching.");
- List<ArchiveEvent> events = new ArrayList<>();
- Map<Path, Set<String>> jobsToRemove = new HashMap<>();
- cachedArchivesPerRefreshDirectory.forEach(
- (path, archives) -> jobsToRemove.put(path, new HashSet<>(archives)));
- Map<Path, Set<Path>> archivesBeyondSizeLimit = new HashMap<>();
- for (HistoryServer.RefreshLocation refreshLocation : refreshDirs) {
- Path refreshDir = refreshLocation.getPath();
- LOG.debug("Checking archive directory {}.", refreshDir);
-
- // contents of /:refreshDir
- FileStatus[] jobArchives;
- try {
- jobArchives = listArchives(refreshLocation.getFs(), refreshDir);
- } catch (IOException e) {
- LOG.error(
- "Failed to access job archive location for path {}.",
- refreshDir,
- e);
- // something went wrong, potentially due to a concurrent deletion
- // do not remove any jobs now; we will retry later
- jobsToRemove.remove(refreshDir);
+ int historySize = 0;
+ for (FileStatus jobArchive : jobArchives) {
+ Path jobArchivePath = jobArchive.getPath();
+ String jobID = jobArchivePath.getName();
+ if (!isValidJobID(jobID, refreshDir)) {
continue;
}
- int historySize = 0;
- for (FileStatus jobArchive : jobArchives) {
- Path jobArchivePath = jobArchive.getPath();
- String jobID = jobArchivePath.getName();
- if (!isValidJobID(jobID, refreshDir)) {
- continue;
- }
+ jobsToRemove.get(refreshDir).remove(jobID);
- jobsToRemove.get(refreshDir).remove(jobID);
-
- historySize++;
- if (historySize > maxHistorySize && processBeyondLimitArchiveDeletion) {
- archivesBeyondSizeLimit
- .computeIfAbsent(refreshDir, ignored -> new HashSet<>())
- .add(jobArchivePath);
- continue;
- }
+ historySize++;
+ if (historySize > maxHistorySize && processBeyondLimitArchiveDeletion) {
+ archivesBeyondSizeLimit
+ .computeIfAbsent(refreshDir, ignored -> new HashSet<>())
+ .add(jobArchivePath);
+ continue;
+ }
- if (cachedArchivesPerRefreshDirectory.get(refreshDir).contains(jobID)) {
- LOG.trace(
- "Ignoring archive {} because it was already fetched.",
- jobArchivePath);
- } else {
- LOG.info("Processing archive {}.", jobArchivePath);
- try {
- processArchive(jobID, jobArchivePath);
- events.add(new ArchiveEvent(jobID, ArchiveEventType.CREATED));
- cachedArchivesPerRefreshDirectory.get(refreshDir).add(jobID);
- LOG.info("Processing archive {} finished.", jobArchivePath);
- } catch (IOException e) {
- LOG.error(
- "Failure while fetching/processing job archive for job {}.",
- jobID,
- e);
- deleteJobFiles(jobID);
- }
+ if (cachedArchivesPerRefreshDirectory.get(refreshDir).contains(jobID)) {
+ LOG.trace(
+ "Ignoring archive {} because it was already fetched.",
+ jobArchivePath);
+ } else {
+ LOG.info("Processing archive {}.", jobArchivePath);
+ try {
+ processArchive(jobID, jobArchivePath);
+ events.add(new ArchiveEvent(jobID, ArchiveEventType.CREATED));
+ cachedArchivesPerRefreshDirectory.get(refreshDir).add(jobID);
+ LOG.info("Processing archive {} finished.", jobArchivePath);
+ } catch (IOException e) {
+ LOG.error(
+ "Failure while fetching/processing job archive for job {}.",
+ jobID,
+ e);
+ deleteJobFiles(jobID);
}
}
}
+ }
- if (jobsToRemove.values().stream().flatMap(Set::stream).findAny().isPresent()
- && processExpiredArchiveDeletion) {
- events.addAll(cleanupExpiredJobs(jobsToRemove));
- }
- if (!archivesBeyondSizeLimit.isEmpty() && processBeyondLimitArchiveDeletion) {
- events.addAll(cleanupJobsBeyondSizeLimit(archivesBeyondSizeLimit));
- }
- if (!events.isEmpty()) {
- updateJobOverview(webOverviewDir, webDir);
- }
- events.forEach(jobArchiveEventListener::accept);
- LOG.debug("Finished archive fetching.");
- } catch (Exception e) {
- LOG.error("Critical failure while fetching/processing job archives.", e);
+ if (jobsToRemove.values().stream().flatMap(Set::stream).findAny().isPresent()
+ && processExpiredArchiveDeletion) {
+ events.addAll(cleanupExpiredJobs(jobsToRemove));
+ }
+ if (!archivesBeyondSizeLimit.isEmpty() && processBeyondLimitArchiveDeletion) {
+ events.addAll(cleanupJobsBeyondSizeLimit(archivesBeyondSizeLimit));
}
+ if (!events.isEmpty()) {
+ updateJobOverview(webOverviewDir, webDir);
+ }
+ events.forEach(jobArchiveEventListener::accept);
+ LOG.debug("Finished archive fetching.");
+ } catch (Exception e) {
+ LOG.error("Critical failure while fetching/processing job archives.", e);
}
+ }
- private static FileStatus[] listArchives(FileSystem refreshFS, Path refreshDir)
- throws IOException {
- // contents of /:refreshDir
- FileStatus[] jobArchives = refreshFS.listStatus(refreshDir);
- if (jobArchives == null) {
- // the entire refreshDirectory was removed
- return new FileStatus[0];
- }
+ private static FileStatus[] listArchives(FileSystem refreshFS, Path refreshDir)
+ throws IOException {
+ // contents of /:refreshDir
+ FileStatus[] jobArchives = refreshFS.listStatus(refreshDir);
+ if (jobArchives == null) {
+ // the entire refreshDirectory was removed
+ return new FileStatus[0];
+ }
- Arrays.sort(
- jobArchives,
- Comparator.comparingLong(FileStatus::getModificationTime).reversed());
+ Arrays.sort(
+ jobArchives, Comparator.comparingLong(FileStatus::getModificationTime).reversed());
- return jobArchives;
+ return jobArchives;
+ }
+
+ private static boolean isValidJobID(String jobId, Path refreshDir) {
+ try {
+ JobID.fromHexString(jobId);
+ return true;
+ } catch (IllegalArgumentException iae) {
+ LOG.debug(
+ "Archive directory {} contained file with unexpected name {}. Ignoring file.",
+ refreshDir,
+ jobId,
+ iae);
+ return false;
}
+ }
+
+ private void processArchive(String jobID, Path jobArchive) throws IOException {
+ for (ArchivedJson archive : FsJobArchivist.getArchivedJsons(jobArchive)) {
+ String path = archive.getPath();
+ String json = archive.getJson();
+
+ File target;
+ if (path.equals(JobsOverviewHeaders.URL)) {
+ target = new File(webOverviewDir, jobID + JSON_FILE_ENDING);
+ } else if (path.equals("/joboverview")) { // legacy path
+ LOG.debug("Migrating legacy archive {}", jobArchive);
+ json = convertLegacyJobOverview(json);
+ target = new File(webOverviewDir, jobID + JSON_FILE_ENDING);
+ } else {
+ // this implicitly writes into webJobDir
+ target = new File(webDir, path + JSON_FILE_ENDING);
+ }
+
+ java.nio.file.Path parent = target.getParentFile().toPath();
- private static boolean isValidJobID(String jobId, Path refreshDir) {
try {
- JobID.fromHexString(jobId);
- return true;
- } catch (IllegalArgumentException iae) {
- LOG.debug(
- "Archive directory {} contained file with unexpected name {}. Ignoring file.",
- refreshDir,
- jobId,
- iae);
- return false;
+ Files.createDirectories(parent);
+ } catch (FileAlreadyExistsException ignored) {
+ // there may be left-over directories from the previous
+ // attempt
}
- }
- private void processArchive(String jobID, Path jobArchive) throws IOException {
- for (ArchivedJson archive : FsJobArchivist.getArchivedJsons(jobArchive)) {
- String path = archive.getPath();
- String json = archive.getJson();
-
- File target;
- if (path.equals(JobsOverviewHeaders.URL)) {
- target = new File(webOverviewDir, jobID + JSON_FILE_ENDING);
- } else if (path.equals("/joboverview")) { // legacy path
- LOG.debug("Migrating legacy archive {}", jobArchive);
- json = convertLegacyJobOverview(json);
- target = new File(webOverviewDir, jobID + JSON_FILE_ENDING);
- } else {
- // this implicitly writes into webJobDir
- target = new File(webDir, path + JSON_FILE_ENDING);
- }
+ java.nio.file.Path targetPath = target.toPath();
- java.nio.file.Path parent = target.getParentFile().toPath();
+ // We overwrite existing files since this may be another attempt
+ // at fetching this archive.
+ // Existing files may be incomplete/corrupt.
+ Files.deleteIfExists(targetPath);
- try {
- Files.createDirectories(parent);
- } catch (FileAlreadyExistsException ignored) {
- // there may be left-over directories from the previous
- // attempt
- }
+ Files.createFile(target.toPath());
+ try (FileWriter fw = new FileWriter(target)) {
+ fw.write(json);
+ fw.flush();
+ }
+ }
+ }
- java.nio.file.Path targetPath = target.toPath();
+ private List<ArchiveEvent> cleanupJobsBeyondSizeLimit(
+ Map<Path, Set<Path>> jobArchivesToRemove) {
+ Map<Path, Set<String>> allJobIdsToRemoveFromOverview = new HashMap<>();
- // We overwrite existing files since this may be another attempt
- // at fetching this archive.
- // Existing files may be incomplete/corrupt.
- Files.deleteIfExists(targetPath);
+ for (Map.Entry<Path, Set<Path>> pathSetEntry : jobArchivesToRemove.entrySet()) {
+ HashSet<String> jobIdsToRemoveFromOverview = new HashSet<>();
- Files.createFile(target.toPath());
- try (FileWriter fw = new FileWriter(target)) {
- fw.write(json);
- fw.flush();
+ for (Path archive : pathSetEntry.getValue()) {
+ jobIdsToRemoveFromOverview.add(archive.getName());
+ try {
+ archive.getFileSystem().delete(archive, false);
+ } catch (IOException ioe) {
+ LOG.warn("Could not delete old archive " + archive, ioe);
}
}
+ allJobIdsToRemoveFromOverview.put(pathSetEntry.getKey(), jobIdsToRemoveFromOverview);
}
- private List<ArchiveEvent> cleanupJobsBeyondSizeLimit(
- Map<Path, Set<Path>> jobArchivesToRemove) {
- Map<Path, Set<String>> allJobIdsToRemoveFromOverview = new HashMap<>();
+ return cleanupExpiredJobs(allJobIdsToRemoveFromOverview);
+ }
- for (Map.Entry<Path, Set<Path>> pathSetEntry : jobArchivesToRemove.entrySet()) {
- HashSet<String> jobIdsToRemoveFromOverview = new HashSet<>();
+ private List<ArchiveEvent> cleanupExpiredJobs(Map<Path, Set<String>> jobsToRemove) {
- for (Path archive : pathSetEntry.getValue()) {
- jobIdsToRemoveFromOverview.add(archive.getName());
- try {
- archive.getFileSystem().delete(archive, false);
- } catch (IOException ioe) {
- LOG.warn("Could not delete old archive " + archive, ioe);
- }
- }
- allJobIdsToRemoveFromOverview.put(
- pathSetEntry.getKey(), jobIdsToRemoveFromOverview);
- }
+ List<ArchiveEvent> deleteLog = new ArrayList<>();
+ LOG.info("Archive directories for jobs {} were deleted.", jobsToRemove);
- return cleanupExpiredJobs(allJobIdsToRemoveFromOverview);
- }
+ jobsToRemove.forEach(
+ (refreshDir, archivesToRemove) -> {
+ cachedArchivesPerRefreshDirectory.get(refreshDir).removeAll(archivesToRemove);
+ });
+ jobsToRemove.values().stream()
+ .flatMap(Set::stream)
+ .forEach(
+ removedJobID -> {
+ deleteJobFiles(removedJobID);
+ deleteLog.add(new ArchiveEvent(removedJobID, ArchiveEventType.DELETED));
+ });
- private List<ArchiveEvent> cleanupExpiredJobs(Map<Path, Set<String>> jobsToRemove) {
-
- List<ArchiveEvent> deleteLog = new ArrayList<>();
- LOG.info("Archive directories for jobs {} were deleted.", jobsToRemove);
-
- jobsToRemove.forEach(
- (refreshDir, archivesToRemove) -> {
- cachedArchivesPerRefreshDirectory
- .get(refreshDir)
- .removeAll(archivesToRemove);
- });
- jobsToRemove.values().stream()
- .flatMap(Set::stream)
- .forEach(
- removedJobID -> {
- deleteJobFiles(removedJobID);
- deleteLog.add(
- new ArchiveEvent(removedJobID, ArchiveEventType.DELETED));
- });
-
- return deleteLog;
- }
+ return deleteLog;
+ }
- private void deleteJobFiles(String jobID) {
- // Make sure we do not include this job in the overview
- try {
- Files.deleteIfExists(new File(webOverviewDir, jobID + JSON_FILE_ENDING).toPath());
- } catch (IOException ioe) {
- LOG.warn("Could not delete file from overview directory.", ioe);
- }
+ private void deleteJobFiles(String jobID) {
+ // Make sure we do not include this job in the overview
+ try {
+ Files.deleteIfExists(new File(webOverviewDir, jobID + JSON_FILE_ENDING).toPath());
+ } catch (IOException ioe) {
+ LOG.warn("Could not delete file from overview directory.", ioe);
+ }
- // Clean up job files we may have created
- File jobDirectory = new File(webJobDir, jobID);
- try {
- FileUtils.deleteDirectory(jobDirectory);
- } catch (IOException ioe) {
- LOG.warn("Could not clean up job directory.", ioe);
- }
+ // Clean up job files we may have created
+ File jobDirectory = new File(webJobDir, jobID);
+ try {
+ FileUtils.deleteDirectory(jobDirectory);
+ } catch (IOException ioe) {
+ LOG.warn("Could not clean up job directory.", ioe);
+ }
- try {
- Files.deleteIfExists(new File(webJobDir, jobID + JSON_FILE_ENDING).toPath());
- } catch (IOException ioe) {
- LOG.warn("Could not delete file from job directory.", ioe);
- }
+ try {
+ Files.deleteIfExists(new File(webJobDir, jobID + JSON_FILE_ENDING).toPath());
+ } catch (IOException ioe) {
+ LOG.warn("Could not delete file from job directory.", ioe);
}
}
[flink] 01/04: [FLINK-17914][hs] Keep archives if listStatus fails
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chesnay pushed a commit to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 026a7aab323cbdf33efd911bec0e353d5caac0e4
Author: Chesnay Schepler <ch...@apache.org>
AuthorDate: Mon Oct 4 13:38:26 2021 +0200
[FLINK-17914][hs] Keep archives if listStatus fails
If FileSystem#listStatus threw an exception then all archives from that directory were removed, because we did not remove the contained jobs from the set of jobs to remove.
It should be fine to keep these archives around, because such failures should only be temporary.
---
.../history/HistoryServerArchiveFetcher.java | 77 +++++++++++++++-------
1 file changed, 54 insertions(+), 23 deletions(-)
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java
index 62c9be4..c9cd0a4 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServerArchiveFetcher.java
@@ -53,8 +53,10 @@ import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import java.util.TimerTask;
import java.util.concurrent.Executors;
@@ -172,7 +174,7 @@ class HistoryServerArchiveFetcher {
private final int maxHistorySize;
/** Cache of all available jobs identified by their id. */
- private final Set<String> cachedArchives;
+ private final Map<Path, Set<String>> cachedArchivesPerRefreshDirectory;
private final File webDir;
private final File webJobDir;
@@ -192,7 +194,10 @@ class HistoryServerArchiveFetcher {
this.processExpiredArchiveDeletion = processExpiredArchiveDeletion;
this.maxHistorySize = maxHistorySize;
this.processBeyondLimitArchiveDeletion = this.maxHistorySize > 0;
- this.cachedArchives = new HashSet<>();
+ this.cachedArchivesPerRefreshDirectory = new HashMap<>();
+ for (HistoryServer.RefreshLocation refreshDir : refreshDirs) {
+ cachedArchivesPerRefreshDirectory.put(refreshDir.getPath(), new HashSet<>());
+ }
this.webDir = checkNotNull(webDir);
this.webJobDir = new File(webDir, "jobs");
Files.createDirectories(webJobDir.toPath());
@@ -206,8 +211,10 @@ class HistoryServerArchiveFetcher {
try {
LOG.debug("Starting archive fetching.");
List<ArchiveEvent> events = new ArrayList<>();
- Set<String> jobsToRemove = new HashSet<>(cachedArchives);
- Set<Path> archivesBeyondSizeLimit = new HashSet<>();
+ Map<Path, Set<String>> jobsToRemove = new HashMap<>();
+ cachedArchivesPerRefreshDirectory.forEach(
+ (path, archives) -> jobsToRemove.put(path, new HashSet<>(archives)));
+ Map<Path, Set<Path>> archivesBeyondSizeLimit = new HashMap<>();
for (HistoryServer.RefreshLocation refreshLocation : refreshDirs) {
Path refreshDir = refreshLocation.getPath();
FileSystem refreshFS = refreshLocation.getFs();
@@ -222,9 +229,13 @@ class HistoryServerArchiveFetcher {
"Failed to access job archive location for path {}.",
refreshDir,
e);
+ // something went wrong, potentially due to a concurrent deletion
+ // do not remove any jobs now; we will retry later
+ jobsToRemove.remove(refreshDir);
continue;
}
if (jobArchives == null) {
+ // the entire refreshDirectory was removed
continue;
}
@@ -247,15 +258,17 @@ class HistoryServerArchiveFetcher {
continue;
}
- jobsToRemove.remove(jobID);
+ jobsToRemove.get(refreshDir).remove(jobID);
historySize++;
if (historySize > maxHistorySize && processBeyondLimitArchiveDeletion) {
- archivesBeyondSizeLimit.add(jobArchivePath);
+ archivesBeyondSizeLimit
+ .computeIfAbsent(refreshDir, ignored -> new HashSet<>())
+ .add(jobArchivePath);
continue;
}
- if (cachedArchives.contains(jobID)) {
+ if (cachedArchivesPerRefreshDirectory.get(refreshDir).contains(jobID)) {
LOG.trace(
"Ignoring archive {} because it was already fetched.",
jobArchivePath);
@@ -302,7 +315,7 @@ class HistoryServerArchiveFetcher {
}
}
events.add(new ArchiveEvent(jobID, ArchiveEventType.CREATED));
- cachedArchives.add(jobID);
+ cachedArchivesPerRefreshDirectory.get(refreshDir).add(jobID);
LOG.info("Processing archive {} finished.", jobArchivePath);
} catch (IOException e) {
LOG.error(
@@ -315,7 +328,8 @@ class HistoryServerArchiveFetcher {
}
}
- if (!jobsToRemove.isEmpty() && processExpiredArchiveDeletion) {
+ if (jobsToRemove.values().stream().flatMap(Set::stream).findAny().isPresent()
+ && processExpiredArchiveDeletion) {
events.addAll(cleanupExpiredJobs(jobsToRemove));
}
if (!archivesBeyondSizeLimit.isEmpty() && processBeyondLimitArchiveDeletion) {
@@ -331,30 +345,47 @@ class HistoryServerArchiveFetcher {
}
}
- private List<ArchiveEvent> cleanupJobsBeyondSizeLimit(Set<Path> jobArchivesToRemove) {
- Set<String> jobIdsToRemoveFromOverview = new HashSet<>();
- for (Path archive : jobArchivesToRemove) {
- jobIdsToRemoveFromOverview.add(archive.getName());
- try {
- archive.getFileSystem().delete(archive, false);
- } catch (IOException ioe) {
- LOG.warn("Could not delete old archive " + archive, ioe);
+ private List<ArchiveEvent> cleanupJobsBeyondSizeLimit(
+ Map<Path, Set<Path>> jobArchivesToRemove) {
+ Map<Path, Set<String>> allJobIdsToRemoveFromOverview = new HashMap<>();
+
+ for (Map.Entry<Path, Set<Path>> pathSetEntry : jobArchivesToRemove.entrySet()) {
+ HashSet<String> jobIdsToRemoveFromOverview = new HashSet<>();
+
+ for (Path archive : pathSetEntry.getValue()) {
+ jobIdsToRemoveFromOverview.add(archive.getName());
+ try {
+ archive.getFileSystem().delete(archive, false);
+ } catch (IOException ioe) {
+ LOG.warn("Could not delete old archive " + archive, ioe);
+ }
}
+ allJobIdsToRemoveFromOverview.put(
+ pathSetEntry.getKey(), jobIdsToRemoveFromOverview);
}
- return cleanupExpiredJobs(jobIdsToRemoveFromOverview);
+
+ return cleanupExpiredJobs(allJobIdsToRemoveFromOverview);
}
- private List<ArchiveEvent> cleanupExpiredJobs(Set<String> jobsToRemove) {
+ private List<ArchiveEvent> cleanupExpiredJobs(Map<Path, Set<String>> jobsToRemove) {
List<ArchiveEvent> deleteLog = new ArrayList<>();
LOG.info("Archive directories for jobs {} were deleted.", jobsToRemove);
- cachedArchives.removeAll(jobsToRemove);
jobsToRemove.forEach(
- removedJobID -> {
- deleteJobFiles(removedJobID);
- deleteLog.add(new ArchiveEvent(removedJobID, ArchiveEventType.DELETED));
+ (refreshDir, archivesToRemove) -> {
+ cachedArchivesPerRefreshDirectory
+ .get(refreshDir)
+ .removeAll(archivesToRemove);
});
+ jobsToRemove.values().stream()
+ .flatMap(Set::stream)
+ .forEach(
+ removedJobID -> {
+ deleteJobFiles(removedJobID);
+ deleteLog.add(
+ new ArchiveEvent(removedJobID, ArchiveEventType.DELETED));
+ });
return deleteLog;
}