[FLINK-3644] [web monitor] Add new config option to set web monitor t…

…mp dir

This closes apache#1824

flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java

@@ -434,6 +434,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
 	 */

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 @@ else if (currentDecoder != null && msg instanceof HttpContent) {
 							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);

flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/WebRuntimeMonitor.java

@@ -146,14 +146,14 @@ public WebRuntimeMonitor(
 
 		// 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 @@ protected void initChannel(SocketChannel ch) {
 
 				ch.pipeline()
 						.addLast(new HttpServerCodec())
-						.addLast(new HttpRequestHandler())
+						.addLast(new HttpRequestHandler(uploadDir))
 						.addLast(handler.name(), handler)
 						.addLast(new PipelineErrorHandler(LOG));
 			}
@@ -425,7 +425,7 @@ private RuntimeMonitorHandler handler(RequestHandler handler) {
 		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")));
 	}
 }

flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/WebRuntimeMonitorITCase.java

@@ -381,7 +381,7 @@ public void testNoEscape() throws Exception {
 
 				// 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 void testNoCopyFromJar() throws Exception {
 						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());