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 2019/02/07 12:02:08 UTC
[flink] branch master updated: [FLINK-11495][runtime] Remove legacy
job archiving paths
This is an automated email from the ASF dual-hosted git repository.
chesnay pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 57a2b75 [FLINK-11495][runtime] Remove legacy job archiving paths
57a2b75 is described below
commit 57a2b754f6a5d8844aa35afb511901ad7ee43068
Author: zentol <ch...@apache.org>
AuthorDate: Thu Jan 31 14:50:15 2019 +0100
[FLINK-11495][runtime] Remove legacy job archiving paths
---
.../history/HistoryServerArchiveFetcher.java | 2 +-
.../webmonitor/history/FsJobArchivistTest.java | 55 ++++------------
.../flink/runtime/history/FsJobArchivist.java | 44 -------------
.../runtime/webmonitor/history/JsonArchivist.java | 3 -
.../flink/runtime/jobmanager/MemoryArchivist.scala | 74 +---------------------
5 files changed, 15 insertions(+), 163 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 e6a189d..86da704 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
@@ -65,7 +65,7 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
* {@link HistoryServerOptions#HISTORY_SERVER_ARCHIVE_DIRS}. The directories are polled in regular 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 defined in the WebRuntimeMonitor.
+ * <p>The archives are downloaded and expanded into a file structure analog to the REST API.
*/
class HistoryServerArchiveFetcher {
diff --git a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/FsJobArchivistTest.java b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/FsJobArchivistTest.java
index 03666a8..312879b 100644
--- a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/FsJobArchivistTest.java
+++ b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/FsJobArchivistTest.java
@@ -18,11 +18,9 @@
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.executiongraph.AccessExecutionGraph;
import org.apache.flink.runtime.history.FsJobArchivist;
-import org.apache.flink.runtime.rest.handler.legacy.utils.ArchivedJobGenerationUtils;
-import org.apache.flink.runtime.webmonitor.WebRuntimeMonitor;
import org.junit.Assert;
import org.junit.Rule;
@@ -31,56 +29,29 @@ import org.junit.rules.TemporaryFolder;
import java.util.ArrayList;
import java.util.Collection;
-import java.util.Iterator;
-import java.util.regex.Pattern;
+
+import static org.hamcrest.Matchers.containsInAnyOrder;
/**
- * This test must reside in flink-runtime-web since the {@link FsJobArchivist} relies on
- * {@link WebRuntimeMonitor#getJsonArchivists()}.
+ * Tests for the {@link FsJobArchivist}.
*/
public class FsJobArchivistTest {
@Rule
- public TemporaryFolder tmpFolder = new TemporaryFolder();
+ public final TemporaryFolder tmpFolder = new TemporaryFolder();
@Test
public void testArchiveJob() throws Exception {
- Path tmpPath = new Path(tmpFolder.getRoot().getAbsolutePath());
-
- AccessExecutionGraph graph = ArchivedJobGenerationUtils.getTestJob();
-
- Collection<ArchivedJson> expected = new ArrayList<>();
- for (JsonArchivist archivist : WebRuntimeMonitor.getJsonArchivists()) {
- for (ArchivedJson archive : archivist.archiveJsonWithPath(graph)) {
- expected.add(archive);
- }
- }
-
- Path archivePath = FsJobArchivist.archiveJob(tmpPath, graph);
- Collection<ArchivedJson> actual = FsJobArchivist.getArchivedJsons(archivePath);
+ final Path tmpPath = new Path(tmpFolder.getRoot().getAbsolutePath());
+ final JobID jobId = new JobID();
- Assert.assertEquals(expected.size(), actual.size());
+ final Collection<ArchivedJson> toArchive = new ArrayList<>(2);
+ toArchive.add(new ArchivedJson("dir1", "hello"));
+ toArchive.add(new ArchivedJson("dir1/dir11", "world"));
- Iterator<ArchivedJson> eI = expected.iterator();
- Iterator<ArchivedJson> aI = actual.iterator();
+ final Path archive = FsJobArchivist.archiveJob(tmpPath, jobId, toArchive);
+ final Collection<ArchivedJson> restored = FsJobArchivist.getArchivedJsons(archive);
- // several jsons contain a dynamic "now" field that depends on the time of creation, so we can't easily compare
- // the json and only check the path
- // /jobs/:jobid
- // /jobs/:jobid/vertices
- // /jobs/:jobid/vertices/:vertexid
- // /jobs/:jobid/vertices/:vertexid/subtasktimes
- // /jobs/:jobid/vertices/:vertexid/taskmanagers
- Pattern jobDetailsPattern = Pattern.compile("/jobs/[a-fA-F0-9]{32}(/vertices(/[a-fA-F0-9]{32}(/(subtasktimes|taskmanagers))?)?)?");
- while (eI.hasNext() && aI.hasNext()) {
- // technically there isn't guarantee that the order is identical, but as it stands this is the case
- ArchivedJson exp = eI.next();
- ArchivedJson act = aI.next();
- if (jobDetailsPattern.matcher(exp.getPath()).matches()) {
- Assert.assertEquals(exp.getPath(), act.getPath());
- } else {
- Assert.assertEquals(exp, act);
- }
- }
+ Assert.assertThat(restored, containsInAnyOrder(toArchive.toArray()));
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/history/FsJobArchivist.java b/flink-runtime/src/main/java/org/apache/flink/runtime/history/FsJobArchivist.java
index 1cfbf96..d0fbc5e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/history/FsJobArchivist.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/history/FsJobArchivist.java
@@ -24,9 +24,7 @@ import org.apache.flink.core.fs.FSDataInputStream;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.Path;
import org.apache.flink.runtime.executiongraph.AccessExecutionGraph;
-import org.apache.flink.runtime.webmonitor.WebMonitorUtils;
import org.apache.flink.runtime.webmonitor.history.ArchivedJson;
-import org.apache.flink.runtime.webmonitor.history.JsonArchivist;
import org.apache.flink.util.IOUtils;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonEncoding;
@@ -65,48 +63,6 @@ public class FsJobArchivist {
* {@link JobManagerOptions#ARCHIVE_DIR}.
*
* @param rootPath directory to which the archive should be written to
- * @param graph graph to archive
- * @return path to where the archive was written, or null if no archive was created
- * @throws IOException
- * @deprecated only kept for legacy reasons
- */
- @Deprecated
- public static Path archiveJob(Path rootPath, AccessExecutionGraph graph) throws IOException {
- try {
- FileSystem fs = rootPath.getFileSystem();
- Path path = new Path(rootPath, graph.getJobID().toString());
- OutputStream out = fs.create(path, FileSystem.WriteMode.NO_OVERWRITE);
-
- try (JsonGenerator gen = jacksonFactory.createGenerator(out, JsonEncoding.UTF8)) {
- gen.writeStartObject();
- gen.writeArrayFieldStart(ARCHIVE);
- for (JsonArchivist archiver : WebMonitorUtils.getJsonArchivists()) {
- for (ArchivedJson archive : archiver.archiveJsonWithPath(graph)) {
- gen.writeStartObject();
- gen.writeStringField(PATH, archive.getPath());
- gen.writeStringField(JSON, archive.getJson());
- gen.writeEndObject();
- }
- }
- gen.writeEndArray();
- gen.writeEndObject();
- } catch (Exception e) {
- fs.delete(path, false);
- throw e;
- }
- LOG.info("Job {} has been archived at {}.", graph.getJobID(), path);
- return path;
- } catch (IOException e) {
- LOG.error("Failed to archive job.", e);
- throw e;
- }
- }
-
- /**
- * Writes the given {@link AccessExecutionGraph} to the {@link FileSystem} pointed to by
- * {@link JobManagerOptions#ARCHIVE_DIR}.
- *
- * @param rootPath directory to which the archive should be written to
* @param jobId job id
* @param jsonToArchive collection of json-path pairs to that should be archived
* @return path to where the archive was written, or null if no archive was created
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/history/JsonArchivist.java b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/history/JsonArchivist.java
index c0a54c6..286f182 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/history/JsonArchivist.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/history/JsonArchivist.java
@@ -25,9 +25,6 @@ import java.util.Collection;
/**
* Interface for all classes that want to participate in the archiving of job-related json responses.
- *
- * <p>Note that all JsonArchivists that are to be used for the history server must be added
- * to {@link WebRuntimeMonitor#getArchivers()}.
*/
public interface JsonArchivist {
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/MemoryArchivist.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/MemoryArchivist.scala
index de42ab2..a2c554b 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/MemoryArchivist.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/MemoryArchivist.scala
@@ -18,27 +18,23 @@
package org.apache.flink.runtime.jobmanager
-import java.io.IOException
-import java.net.URI
import java.util
import akka.actor.ActorRef
import grizzled.slf4j.Logger
import org.apache.flink.api.common.JobID
-import org.apache.flink.core.fs.{FileSystem, Path}
+import org.apache.flink.core.fs.Path
import org.apache.flink.runtime.jobgraph.JobStatus
import org.apache.flink.runtime.messages.accumulators._
import org.apache.flink.runtime.webmonitor.WebMonitorUtils
import org.apache.flink.runtime.{FlinkActor, LogMessages}
import org.apache.flink.runtime.messages.webmonitor._
import org.apache.flink.runtime.executiongraph.{ArchivedExecutionGraph, ExecutionGraph}
-import org.apache.flink.runtime.history.FsJobArchivist
import org.apache.flink.runtime.messages.ArchiveMessages._
import org.apache.flink.runtime.messages.JobManagerMessages._
import org.apache.flink.runtime.messages.webmonitor.JobIdsWithStatusOverview.JobIdWithStatus
import scala.collection.mutable
-import scala.concurrent.future
/**
* Actor which stores terminated Flink jobs. The number of stored Flink jobs is set by max_entries.
@@ -105,8 +101,6 @@ class MemoryArchivist(
case _ =>
}
- archiveJsonFiles(graph)
-
trimHistory()
case msg : InfoMessage => handleWebServerInfoMessage(msg, sender())
@@ -194,27 +188,6 @@ class MemoryArchivist(
}
}
- private def archiveJsonFiles(graph: ArchivedExecutionGraph) {
- // a suspended job is expected to continue on another job manager,
- // so we aren't archiving it yet.
- if (archivePath.isDefined && graph.getState.isGloballyTerminalState) {
- try {
- val p = validateAndNormalizeUri(archivePath.get.toUri)
- future {
- try {
- FsJobArchivist.archiveJob(p, graph)
- } catch {
- case e: Exception =>
- log.error("Failed to archive job.", e)
- }
- }(context.dispatcher)
- } catch {
- case e: Exception =>
- log.warn(s"Failed to create Path for $archivePath. Job will not be archived.", e)
- }
- }
- }
-
// --------------------------------------------------------------------------
// Request Responses
// --------------------------------------------------------------------------
@@ -250,49 +223,4 @@ class MemoryArchivist(
graphs.remove(jobID)
}
}
-
- /**
- * Checks and normalizes the archive path URI. This method first checks the validity of the
- * URI (scheme, path, availability of a matching file system) and then normalizes the URL
- * to a path.
- *
- * If the URI does not include an authority, but the file system configured for the URI has an
- * authority, then the normalized path will include this authority.
- *
- * @param archivePathUri The URI to check and normalize.
- * @return a normalized URI as a Path.
- *
- * @throws IllegalArgumentException Thrown, if the URI misses schema or path.
- * @throws IOException Thrown, if no file system can be found for the URI's scheme.
- */
- @throws[IOException]
- private def validateAndNormalizeUri(archivePathUri: URI): Path = {
- val scheme = archivePathUri.getScheme
- val path = archivePathUri.getPath
-
- // some validity checks
- if (scheme == null) {
- throw new IllegalArgumentException("The scheme (hdfs://, file://, etc) is null. " +
- "Please specify the file system scheme explicitly in the URI: " + archivePathUri)
- }
-
- if (path == null) {
- throw new IllegalArgumentException("The path to store the job archives is null. " +
- "Please specify a directory path for storing job archives. and the URI is: " +
- archivePathUri)
- }
-
- if (path.length == 0 || path == "/") {
- throw new IllegalArgumentException("Cannot use the root directory for storing job archives.")
- }
-
- try {
- FileSystem.get(archivePathUri)
- }
- catch {
- case e: Exception =>
- throw new IllegalArgumentException(s"No file system found for URI '${archivePathUri}'.")
- }
- new Path(archivePathUri)
- }
}