[FLINK-12869] Add yarn acls capability to flink containers

apache · Jun 17, 2019 · 8acc0e8 · 8acc0e8
1 parent 4320d83
commit 8acc0e8
Show file tree

Hide file tree

Showing 4 changed files with 42 additions and 0 deletions.
diff --git a/docs/_includes/generated/yarn_config_configuration.html b/docs/_includes/generated/yarn_config_configuration.html
@@ -72,5 +72,15 @@
             <td style="word-wrap: break-word;">(none)</td>
             <td>A comma-separated list of tags to apply to the Flink YARN application.</td>
         </tr>
+        <tr>
+            <td><h5>yarn.view.acls</h5></td>
+            <td style="word-wrap: break-word;">(none)</td>
+            <td>Users and groups to give VIEW acess. The ACLs are of for comma-separated-usersspacecomma-separated-groups</td>
+        </tr>
+        <tr>
+            <td><h5>yarn.admin.acls</h5></td>
+            <td style="word-wrap: break-word;">(none)</td>
+            <td>Users and groups to give MODIFY acess. The ACLs are of for comma-separated-usersspacecomma-separated-groups</td>
+        </tr>
     </tbody>
 </table>
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java b/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java
@@ -932,6 +932,8 @@ public ApplicationReport startAppMaster(
 		amContainer.setLocalResources(localResources);
 		fs.close();
 
+		Utils.setAclsFor(amContainer, flinkConfiguration);
+
 		// Setup CLASSPATH and environment variables for ApplicationMaster
 		final Map<String, String> appMasterEnv = new HashMap<>();
 		// set user specified app master environment variables

diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/Utils.java b/flink-yarn/src/main/java/org/apache/flink/yarn/Utils.java
@@ -25,6 +25,7 @@
 import org.apache.flink.runtime.util.HadoopUtils;
 import org.apache.flink.util.FileUtils;
 import org.apache.flink.util.StringUtils;
+import org.apache.flink.yarn.configuration.YarnConfigOptions;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileStatus;
@@ -40,6 +41,7 @@
 import org.apache.hadoop.util.StringInterner;
 import org.apache.hadoop.yarn.api.ApplicationConstants;
 import org.apache.hadoop.yarn.api.ApplicationConstants.Environment;
+import org.apache.hadoop.yarn.api.records.ApplicationAccessType;
 import org.apache.hadoop.yarn.api.records.ContainerLaunchContext;
 import org.apache.hadoop.yarn.api.records.LocalResource;
 import org.apache.hadoop.yarn.api.records.LocalResourceType;
@@ -230,6 +232,13 @@ private static LocalResource registerLocalResource(FileSystem fs, Path remoteRsr
 		return localResource;
 	}
 
+	public static void setAclsFor(ContainerLaunchContext amContainer, org.apache.flink.configuration.Configuration flinkConfig) {
+		amContainer.setApplicationACLs(new HashMap<ApplicationAccessType, String>(){{
+			put(ApplicationAccessType.VIEW_APP, flinkConfig.getString(YarnConfigOptions.APPLICATION_VIEW_ACLS));
+			put(ApplicationAccessType.MODIFY_APP, flinkConfig.getString(YarnConfigOptions.APPLICATION_ADMIN_ACLS));
+		}});
+	}
+
 	public static void setTokensFor(ContainerLaunchContext amContainer, List<Path> paths, Configuration conf) throws IOException {
 		Credentials credentials = new Credentials();
 		// for HDFS
@@ -550,6 +559,8 @@ static ContainerLaunchContext createTaskExecutorContext(
 
 		ctx.setEnvironment(containerEnv);
 
+		setAclsFor(ctx, flinkConfig);
+
 		// For TaskManager YARN container context, read the tokens from the jobmanager yarn container local file.
 		// NOTE: must read the tokens from the local file, not from the UGI context, because if UGI is login
 		// using Kerberos keytabs, there is no HDFS delegation token in the UGI context.

diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/configuration/YarnConfigOptions.java b/flink-yarn/src/main/java/org/apache/flink/yarn/configuration/YarnConfigOptions.java
@@ -188,6 +188,25 @@ public class YarnConfigOptions {
 		.defaultValue("")
 		.withDescription("A comma-separated list of tags to apply to the Flink YARN application.");
 
+	/**
+	 * Users and groups to give VIEW access.
+	 * https://www.cloudera.com/documentation/enterprise/latest/topics/cm_mc_yarn_acl.html
+	 */
+	public static final ConfigOption<String> APPLICATION_VIEW_ACLS =
+		key("yarn.view.acls")
+		.defaultValue("")
+		.withDescription("Users and groups to give VIEW acess. The ACLs are of for" +
+			" comma-separated-usersspacecomma-separated-groups");
+
+	/**
+	 * Users and groups to give MODIFY access.
+	 */
+	public static final ConfigOption<String> APPLICATION_ADMIN_ACLS =
+		key("yarn.admin.acls")
+		.defaultValue("")
+		.withDescription("Users and groups to give MODIFY acess. The ACLs are of for" +
+			" comma-separated-usersspacecomma-separated-groups");
+
 	// ------------------------------------------------------------------------
 
 	/** This class is not meant to be instantiated. */