You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2016/04/13 10:31:11 UTC
[13/14] flink git commit: [FLINK-3644] [web monitor] Add new config
option to set web monitor tmp dir
[FLINK-3644] [web monitor] Add new config option to set web monitor tmp dir
This closes #1824
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/b188637b
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/b188637b
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/b188637b
Branch: refs/heads/master
Commit: b188637b9016116728f0d16f94e213584b2abfd6
Parents: 6bb085e
Author: xueyan.li <xu...@qunar.com>
Authored: Tue Apr 5 14:10:48 2016 +0800
Committer: Stephan Ewen <se...@apache.org>
Committed: Wed Apr 13 01:10:55 2016 +0200
----------------------------------------------------------------------
.../apache/flink/configuration/ConfigConstants.java | 5 +++++
.../flink/runtime/webmonitor/HttpRequestHandler.java | 15 ++++++++-------
.../flink/runtime/webmonitor/WebRuntimeMonitor.java | 10 +++++-----
.../runtime/webmonitor/WebRuntimeMonitorITCase.java | 4 ++--
4 files changed, 20 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/b188637b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
index ba2d880..53d9a37 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
@@ -435,6 +435,11 @@ public final class ConfigConstants {
public static final String JOB_MANAGER_WEB_PORT_KEY = "jobmanager.web.port";
/**
+ * The config parameter defining the flink web directory to be used by the webmonitor.
+ */
+ public static final String JOB_MANAGER_WEB_TMPDIR_KEY = "jobmanager.web.tmpdir";
+
+ /**
* The config parameter defining the number of archived jobs for the jobmanager
*/
public static final String JOB_MANAGER_WEB_ARCHIVE_COUNT = "jobmanager.web.history";
http://git-wip-us.apache.org/repos/asf/flink/blob/b188637b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/HttpRequestHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/HttpRequestHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/HttpRequestHandler.java
index c9190c9..bbd29fa 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/HttpRequestHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/HttpRequestHandler.java
@@ -69,17 +69,18 @@ public class HttpRequestHandler extends SimpleChannelInboundHandler<HttpObject>
/** A decoder factory that always stores POST chunks on disk */
private static final HttpDataFactory DATA_FACTORY = new DefaultHttpDataFactory(true);
-
- private static final File TMP_DIR = new File(System.getProperty("java.io.tmpdir"));
-
-
+
+ private final File tmpDir;
+
private HttpRequest currentRequest;
private HttpPostRequestDecoder currentDecoder;
-
private String currentRequestPath;
-
+ public HttpRequestHandler(File tmpDir) {
+ this.tmpDir = tmpDir;
+ }
+
@Override
public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
if (currentDecoder != null) {
@@ -130,7 +131,7 @@ public class HttpRequestHandler extends SimpleChannelInboundHandler<HttpObject>
if (file.isCompleted()) {
String name = file.getFilename();
- File target = new File(TMP_DIR, UUID.randomUUID() + "_" + name);
+ File target = new File(tmpDir, UUID.randomUUID() + "_" + name);
file.renameTo(target);
QueryStringEncoder encoder = new QueryStringEncoder(currentRequestPath);
http://git-wip-us.apache.org/repos/asf/flink/blob/b188637b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/WebRuntimeMonitor.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/WebRuntimeMonitor.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/WebRuntimeMonitor.java
index b6c578b..582004a 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/WebRuntimeMonitor.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/WebRuntimeMonitor.java
@@ -146,14 +146,14 @@ public class WebRuntimeMonitor implements WebMonitor {
// create an empty directory in temp for the web server
String rootDirFileName = "flink-web-" + UUID.randomUUID();
- webRootDir = new File(getBaseDir(), rootDirFileName);
+ webRootDir = new File(getBaseDir(config), rootDirFileName);
LOG.info("Using directory {} for the web interface files", webRootDir);
final boolean webSubmitAllow = cfg.isProgramSubmitEnabled();
if (webSubmitAllow) {
// create storage for uploads
String uploadDirName = "flink-web-upload-" + UUID.randomUUID();
- this.uploadDir = new File(getBaseDir(), uploadDirName);
+ this.uploadDir = new File(getBaseDir(config), uploadDirName);
if (!uploadDir.mkdir() || !uploadDir.canWrite()) {
throw new IOException("Unable to create temporary directory to support jar uploads.");
}
@@ -306,7 +306,7 @@ public class WebRuntimeMonitor implements WebMonitor {
ch.pipeline()
.addLast(new HttpServerCodec())
- .addLast(new HttpRequestHandler())
+ .addLast(new HttpRequestHandler(uploadDir))
.addLast(handler.name(), handler)
.addLast(new PipelineErrorHandler(LOG));
}
@@ -425,7 +425,7 @@ public class WebRuntimeMonitor implements WebMonitor {
return new RuntimeMonitorHandler(handler, retriever, jobManagerAddressPromise.future(), timeout);
}
- File getBaseDir() {
- return new File(System.getProperty("java.io.tmpdir"));
+ File getBaseDir(Configuration configuration) {
+ return new File(configuration.getString(ConfigConstants.JOB_MANAGER_WEB_TMPDIR_KEY, System.getProperty("java.io.tmpdir")));
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/b188637b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/WebRuntimeMonitorITCase.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/WebRuntimeMonitorITCase.java b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/WebRuntimeMonitorITCase.java
index 8b46cdb..677ff54 100644
--- a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/WebRuntimeMonitorITCase.java
+++ b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/WebRuntimeMonitorITCase.java
@@ -381,7 +381,7 @@ public class WebRuntimeMonitorITCase extends TestLogger {
// 2) Request file outside of web root
// Create a test file in the web base dir (parent of web root)
- File illegalFile = new File(webMonitor.getBaseDir(), "test-file-" + UUID.randomUUID());
+ File illegalFile = new File(webMonitor.getBaseDir(new Configuration()), "test-file-" + UUID.randomUUID());
illegalFile.deleteOnExit();
assertTrue("Failed to create test file", illegalFile.createNewFile());
@@ -467,7 +467,7 @@ public class WebRuntimeMonitorITCase extends TestLogger {
response.getStatus());
assertFalse("Did not respond with the file, but still copied it from the JAR.",
- new File(webMonitor.getBaseDir(), "log4j-test.properties").exists());
+ new File(webMonitor.getBaseDir(new Configuration()), "log4j-test.properties").exists());
// 3) Request non-existing file
client.sendGetRequest("not-existing-resource", deadline.timeLeft());