[FLINK-5091] Formalize the Mesos AppMaster environment for docker com…

…patibility

- introduced ContainerSpecification.
- reworked how the TM container environment is constructed; eliminated
- special-case environment variables, file layout.
- added dynamic configuration support to GlobalConfiguration.
- integrated the SecurityContext into AM/TM runners.
- added config setting for Mesos framework user.
- support DCOS side-channel authentication.
- set the FS default scheme.
- made the artifact server more generic (no assumption about existence
- of dispatcher, Path-based).
- moved some test code related to overriding the JVM’s env.
- moved the Mesos containerizer config code to the MesosTaskManagerParameters.

This closes apache#2915.

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

@@ -439,6 +439,11 @@ public final class ConfigConstants {
 
 	// ------------------------ Mesos Configuration ------------------------
 
+	/**
+	 * The initial number of Mesos tasks to allocate.
+	 */
+	public static final String MESOS_INITIAL_TASKS = "mesos.initial-tasks";
+
 	/**
 	 * The maximum number of failed Mesos tasks before entirely stopping
 	 * the Mesos session / job on Mesos.
@@ -484,6 +489,8 @@ public final class ConfigConstants {
 
 	public static final String MESOS_RESOURCEMANAGER_FRAMEWORK_SECRET = "mesos.resourcemanager.framework.secret";
 
+	public static final String MESOS_RESOURCEMANAGER_FRAMEWORK_USER = "mesos.resourcemanager.framework.user";
+
 	/**
 	 * The cpus to acquire from Mesos.
 	 *
@@ -1186,6 +1193,8 @@ public final class ConfigConstants {
 
 	public static final String DEFAULT_MESOS_RESOURCEMANAGER_FRAMEWORK_ROLE = "*";
 
+	public static final String DEFAULT_MESOS_RESOURCEMANAGER_FRAMEWORK_USER = "";
+
 	/** Default value to override SSL support for the Artifact Server */
 	public static final boolean DEFAULT_MESOS_ARTIFACT_SERVER_SSL_ENABLED = true;
 
@@ -1405,6 +1414,12 @@ public final class ConfigConstants {
 	/** The environment variable name which contains the location of the lib folder */
 	public static final String ENV_FLINK_LIB_DIR = "FLINK_LIB_DIR";
 
+	/** The environment variable name which contains the location of the bin directory */
+	public static final String ENV_FLINK_BIN_DIR = "FLINK_BIN_DIR";
+
+	/** The environment variable name which contains the Flink installation root directory */
+	public static final String ENV_FLINK_HOME_DIR = "FLINK_HOME";
+
 	// -------------------------------- Security -------------------------------
 
 	/**

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

@@ -39,12 +39,31 @@ public final class GlobalConfiguration {
 
 	public static final String FLINK_CONF_FILENAME = "flink-conf.yaml";
 
+
 	// --------------------------------------------------------------------------------------------
 
 	private GlobalConfiguration() {}
 
 	// --------------------------------------------------------------------------------------------
 
+	private static Configuration dynamicProperties = null;
+
+	/**
+	 * Set the process-wide dynamic properties to be merged with the loaded configuration.
+     */
+	public static void setDynamicProperties(Configuration dynamicProperties) {
+		GlobalConfiguration.dynamicProperties = new Configuration(dynamicProperties);
+	}
+
+	/**
+	 * Get the dynamic properties.
+     */
+	public static Configuration getDynamicProperties() {
+		return GlobalConfiguration.dynamicProperties;
+	}
+
+	// --------------------------------------------------------------------------------------------
+
 	/**
 	 * Loads the global configuration from the environment. Fails if an error occurs during loading. Returns an
 	 * empty configuration object if the environment variable is not set. In production this variable is set but
@@ -90,7 +109,13 @@ public static Configuration loadConfiguration(final String configDir) {
 					"' (" + confDirFile.getAbsolutePath() + ") does not exist.");
 		}
 
-		return loadYAMLResource(yamlConfigFile);
+		Configuration conf = loadYAMLResource(yamlConfigFile);
+
+		if(dynamicProperties != null) {
+			conf.addAll(dynamicProperties);
+		}
+
+		return conf;
 	}
 
 	/**

flink-dist/src/main/assemblies/bin.xml

@@ -82,6 +82,12 @@ under the License.
 			<outputDirectory>bin</outputDirectory>
 			<fileMode>0755</fileMode>
 		</fileSet>
+		<!-- copy Mesos start scripts -->
+		<fileSet>
+			<directory>src/main/flink-bin/mesos-bin</directory>
+			<outputDirectory>bin</outputDirectory>
+			<fileMode>0755</fileMode>
+		</fileSet>
 
 		<!-- copy default configuration -->
 		<fileSet>

flink-dist/src/main/flink-bin/mesos-bin/mesos-appmaster.sh

@@ -0,0 +1,51 @@
+#!/usr/bin/env bash
+################################################################################
+#  Licensed to the Apache Software Foundation (ASF) under one
+#  or more contributor license agreements.  See the NOTICE file
+#  distributed with this work for additional information
+#  regarding copyright ownership.  The ASF licenses this file
+#  to you under the Apache License, Version 2.0 (the
+#  "License"); you may not use this file except in compliance
+#  with the License.  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+bin=`dirname "$0"`
+bin=`cd "$bin"; pwd`
+
+# get Flink config
+. "$bin"/config.sh
+
+# auxilliary function to construct a lightweight classpath for the
+# Flink AppMaster
+constructAppMasterClassPath() {
+
+    while read -d '' -r jarfile ; do
+        if [[ $CC_CLASSPATH = "" ]]; then
+            CC_CLASSPATH="$jarfile";
+        else
+            CC_CLASSPATH="$CC_CLASSPATH":"$jarfile"
+        fi
+    done < <(find "$FLINK_LIB_DIR" ! -type d -name '*.jar' -print0)
+
+    echo $CC_CLASSPATH:$INTERNAL_HADOOP_CLASSPATHS
+}
+
+CC_CLASSPATH=`manglePathList $(constructAppMasterClassPath)`
+
+log=flink-appmaster.log
+log_setting="-Dlog.file="$log" -Dlog4j.configuration=file:"$FLINK_CONF_DIR"/log4j.properties -Dlogback.configurationFile=file:"$FLINK_CONF_DIR"/logback.xml"
+
+export FLINK_CONF_DIR
+export FLINK_BIN_DIR
+export FLINK_LIB_DIR
+
+$JAVA_RUN $JVM_ARGS -classpath "$CC_CLASSPATH" $log_setting org.apache.flink.mesos.runtime.clusterframework.MesosApplicationMasterRunner "$@"
+

flink-dist/src/main/flink-bin/mesos-bin/mesos-taskmanager.sh

@@ -0,0 +1,60 @@
+#!/usr/bin/env bash
+################################################################################
+#  Licensed to the Apache Software Foundation (ASF) under one
+#  or more contributor license agreements.  See the NOTICE file
+#  distributed with this work for additional information
+#  regarding copyright ownership.  The ASF licenses this file
+#  to you under the Apache License, Version 2.0 (the
+#  "License"); you may not use this file except in compliance
+#  with the License.  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+bin=`dirname "$0"`
+bin=`cd "$bin"; pwd`
+
+# get Flink config
+. "$bin"/config.sh
+
+# auxilliary function to construct a lightweight classpath for the
+# Flink TaskManager
+constructTaskManagerClassPath() {
+
+    while read -d '' -r jarfile ; do
+        if [[ $CC_CLASSPATH = "" ]]; then
+            CC_CLASSPATH="$jarfile";
+        else
+            CC_CLASSPATH="$CC_CLASSPATH":"$jarfile"
+        fi
+    done < <(find "$FLINK_LIB_DIR" ! -type d -name '*.jar' -print0)
+
+    echo $CC_CLASSPATH:$INTERNAL_HADOOP_CLASSPATHS
+}
+
+CC_CLASSPATH=`manglePathList $(constructTaskManagerClassPath)`
+
+log=flink-taskmanager.log
+log_setting="-Dlog.file="$log" -Dlog4j.configuration=file:"$FLINK_CONF_DIR"/log4j.properties -Dlogback.configurationFile=file:"$FLINK_CONF_DIR"/logback.xml"
+
+# Add precomputed memory JVM options
+if [ -z "${FLINK_ENV_JAVA_OPTS_MEM}" ]; then
+    FLINK_ENV_JAVA_OPTS_MEM=""
+fi
+export FLINK_ENV_JAVA_OPTS="${FLINK_ENV_JAVA_OPTS} ${FLINK_ENV_JAVA_OPTS_MEM}"
+
+# Add TaskManager-specific JVM options
+export FLINK_ENV_JAVA_OPTS="${FLINK_ENV_JAVA_OPTS} ${FLINK_ENV_JAVA_OPTS_TM}"
+
+export FLINK_CONF_DIR
+export FLINK_BIN_DIR
+export FLINK_LIB_DIR
+
+$JAVA_RUN $JVM_ARGS ${FLINK_ENV_JAVA_OPTS} -classpath "$CC_CLASSPATH" $log_setting org.apache.flink.mesos.runtime.clusterframework.MesosTaskManager "$@"
+

flink-mesos/src/main/java/org/apache/flink/mesos/Utils.java

@@ -18,7 +18,10 @@
 
 package org.apache.flink.mesos;
 
+import org.apache.flink.mesos.util.MesosArtifactResolver;
+import org.apache.flink.runtime.clusterframework.ContainerSpecification;
 import org.apache.mesos.Protos;
+import scala.Option;
 
 import java.net.URL;
 import java.util.Arrays;
@@ -45,6 +48,24 @@ public static Protos.CommandInfo.URI uri(URL url, boolean cacheable) {
 			.build();
 	}
 
+	/**
+	 * Construct a Mesos URI.
+	 */
+	public static Protos.CommandInfo.URI uri(MesosArtifactResolver resolver, ContainerSpecification.Artifact artifact) {
+		Option<URL> url = resolver.resolve(artifact.dest);
+		if(url.isEmpty()) {
+			throw new IllegalArgumentException("Unresolvable artifact: " + artifact.dest);
+		}
+
+		return Protos.CommandInfo.URI.newBuilder()
+			.setValue(url.get().toExternalForm())
+			.setOutputFile(artifact.dest.toString())
+			.setExtract(artifact.extract)
+			.setCache(artifact.cachable)
+			.setExecutable(artifact.executable)
+			.build();
+	}
+
 	/**
 	 * Construct a scalar resource value.
 	 */