Run Spark job with specify the username for driver and executor (apache#283)

avovchenko · ekrivokonmapr · commit 1654a42237b7 · 2019-09-19T17:41:51.000+03:00
diff --git a/core/src/main/scala/org/apache/spark/util/Utils.scala b/core/src/main/scala/org/apache/spark/util/Utils.scala
@@ -2421,8 +2421,33 @@ private[spark] object Utils extends Logging {
       .getOrElse(UserGroupInformation.getCurrentUser().getShortUserName())
   }
 
+  def getCurrentUserId(): String = {
+    val username = getCurrentUserName()
+    val cmdSeq = Seq("bash", "-c", "id -u " + username)
+
+    val userId = try {
+      executeAndGetOutput(cmdSeq).stripLineEnd
+    } catch {
+      case e: Exception => logError(s"Error getting user id for user=$username", e)
+      ""
+    }
+    userId
+  }
+
   val EMPTY_USER_GROUPS = Set.empty[String]
 
+  def getCurrentUserGroupsIds(sparkConf: SparkConf, username: String): Set[String] = {
+    val cmdSeq = Seq("bash", "-c", "id -G " + username)
+
+    try {
+      val userGroupsIds = executeAndGetOutput(cmdSeq).stripLineEnd.split(" ").toSet
+      return userGroupsIds
+    } catch {
+      case e: Exception => logError(s"Error getting groups ids for user=$username", e)
+    }
+    EMPTY_USER_GROUPS
+  }
+
   // Returns the groups to which the current user belongs.
   def getCurrentUserGroups(sparkConf: SparkConf, username: String): Set[String] = {
     val groupProviderClassName = sparkConf.get("spark.user.groups.mapping",
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala
@@ -305,6 +305,12 @@ private[spark] object Config extends Logging {
       .stringConf
       .createWithDefault("mapr-cluster-configmap")
 
+  val MAPR_CLUSTER_USER_SECRETS =
+    ConfigBuilder("spark.mapr.user.secrets")
+      .doc("Name of the mapr user secrets")
+      .stringConf
+      .createWithDefault("mapr-user-secrets")
+
   val KUBERNETES_AUTH_SUBMISSION_CONF_PREFIX =
     "spark.kubernetes.authenticate.submission"
 
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Constants.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Constants.scala
@@ -79,8 +79,9 @@ private[spark] object Constants {
   val DRIVER_CONTAINER_NAME = "spark-kubernetes-driver"
   val MEMORY_OVERHEAD_MIN_MIB = 384L
 
-  // Mapr specific Constants
-  val CLDB_HOSTS = "MAPR_CLDB_HOSTS"
-  val ZK_HOSTS = "MAPR_ZK_HOSTS"
-  val CLUSTER_NAME = "MAPR_CLUSTER"
+  // User specific Constants
+  val CURRENT_USER = "CURRENT_USER"
+  val USER_GROUPS = "USER_GROUPS"
+  val CURRENT_USER_ID = "USER_ID"
+  val USER_GROUPS_IDS = "USER_GROUPS_IDS"
 }
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/steps/BasicDriverConfigurationStep.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/steps/BasicDriverConfigurationStep.scala
@@ -17,15 +17,14 @@
 package org.apache.spark.deploy.k8s.submit.steps
 
 import scala.collection.JavaConverters._
-
 import io.fabric8.kubernetes.api.model._
-
 import org.apache.spark.{SparkConf, SparkException}
 import org.apache.spark.deploy.k8s.Config._
 import org.apache.spark.deploy.k8s.Constants._
 import org.apache.spark.deploy.k8s.KubernetesUtils
 import org.apache.spark.deploy.k8s.submit.KubernetesDriverSpec
 import org.apache.spark.internal.config.{DRIVER_CLASS_PATH, DRIVER_MEMORY, DRIVER_MEMORY_OVERHEAD}
+import org.apache.spark.util.Utils
 
 /**
  * Performs basic configuration for the driver pod.
@@ -112,6 +111,16 @@ private[spark] class BasicDriverConfigurationStep(
     }
 
     val clusterConfMap = sparkConf.get(MAPR_CLUSTER_CONFIGMAP).toString
+    val clusterUserSecrets = sparkConf.get(MAPR_CLUSTER_USER_SECRETS).toString
+
+    val username = Utils.getCurrentUserName()
+    val userGroups = Utils.getCurrentUserGroups(sparkConf, username)
+    val userId = Utils.getCurrentUserId()
+    val userGroupsIds = Utils.getCurrentUserGroupsIds(sparkConf, username)
+
+    if (userId.length() == 0 || userGroupsIds.size == 0) {
+      throw new RuntimeException(s"Error getting uid/gid for user=$username")
+    }
 
     val driverContainer = new ContainerBuilder(driverSpec.driverContainer)
       .withName(DRIVER_CONTAINER_NAME)
@@ -120,6 +129,22 @@ private[spark] class BasicDriverConfigurationStep(
       .addAllToEnv(driverCustomEnvs.asJava)
       .addAllToEnv(clusterEnvs.asJava)
       .addToEnv(driverExtraClasspathEnv.toSeq: _*)
+      .addNewEnv()
+        .withName(CURRENT_USER)
+        .withValue(username)
+        .endEnv()
+      .addNewEnv()
+        .withName(USER_GROUPS)
+        .withValue(userGroups.mkString(" "))
+        .endEnv()
+      .addNewEnv()
+        .withName(CURRENT_USER_ID)
+        .withValue(userId)
+        .endEnv()
+      .addNewEnv()
+        .withName(USER_GROUPS_IDS)
+        .withValue(userGroupsIds.mkString(" "))
+        .endEnv()
       .addNewEnv()
         .withName(ENV_DRIVER_MEMORY)
         .withValue(driverMemoryString)
@@ -138,27 +163,16 @@ private[spark] class BasicDriverConfigurationStep(
           .withNewFieldRef("v1", "status.podIP")
           .build())
         .endEnv()
-      .addNewEnv()
-        .withName(CLDB_HOSTS)
-        .withNewValueFrom()
-        .withConfigMapKeyRef(
-          new ConfigMapKeySelector(CLDB_HOSTS, clusterConfMap, true))
-        .endValueFrom()
-        .endEnv()
-      .addNewEnv()
-        .withName(ZK_HOSTS)
-        .withNewValueFrom()
-        .withConfigMapKeyRef(
-          new ConfigMapKeySelector(ZK_HOSTS, clusterConfMap, true))
-        .endValueFrom()
-        .endEnv()
-      .addNewEnv()
-        .withName(CLUSTER_NAME)
-        .withNewValueFrom()
-        .withConfigMapKeyRef(
-          new ConfigMapKeySelector(CLUSTER_NAME, clusterConfMap, true))
-        .endValueFrom()
-        .endEnv()
+        .addNewEnvFrom()
+          .withNewConfigMapRef()
+            .withName(clusterConfMap)
+            .endConfigMapRef()
+          .endEnvFrom()
+      .addNewEnvFrom()
+        .withNewSecretRef()
+          .withName(clusterUserSecrets)
+          .endSecretRef()
+        .endEnvFrom()
       .withNewResources()
         .addToRequests("cpu", driverCpuQuantity)
         .addToRequests("memory", driverMemoryQuantity)
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodFactory.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodFactory.scala
@@ -17,10 +17,8 @@
 package org.apache.spark.scheduler.cluster.k8s
 
 import scala.collection.JavaConverters._
-
 import io.fabric8.kubernetes.api.model._
-
-import org.apache.spark.{SparkConf, SparkException}
+import org.apache.spark.{SparkConf, SparkContext, SparkException}
 import org.apache.spark.deploy.k8s.{InitContainerBootstrap, KubernetesUtils, MountSecretsBootstrap, PodWithDetachedInitContainer}
 import org.apache.spark.deploy.k8s.Config._
 import org.apache.spark.deploy.k8s.Constants._
@@ -179,33 +177,48 @@ private[spark] class ExecutorPodFactory(
       }
 
     val clusterConfMap = sparkConf.get(MAPR_CLUSTER_CONFIGMAP).toString
+    val clusterUserSecrets = sparkConf.get(MAPR_CLUSTER_USER_SECRETS).toString
+
+    val username = Utils.getCurrentUserName()
+    val userGroups = Utils.getCurrentUserGroups(sparkConf, username)
+    val userId = Utils.getCurrentUserId()
+    val userGroupsIds = Utils.getCurrentUserGroupsIds(sparkConf, username)
+
+    if (userId.length() == 0 || userGroupsIds.size == 0) {
+      throw new RuntimeException(s"Error getting uid/gid for user=$username")
+    }
 
     val executorContainer = new ContainerBuilder()
       .withName("executor")
       .withImage(executorContainerImage)
       .withImagePullPolicy(imagePullPolicy)
       .addAllToEnv(clusterEnvs.asJava)
       .addNewEnv()
-        .withName(CLDB_HOSTS)
-        .withNewValueFrom()
-        .withConfigMapKeyRef(
-           new ConfigMapKeySelector(CLDB_HOSTS, clusterConfMap, false))
-        .endValueFrom()
+        .withName(CURRENT_USER)
+        .withValue(username)
+        .endEnv()
+      .addNewEnv()
+        .withName(USER_GROUPS)
+        .withValue(userGroups.mkString(" "))
         .endEnv()
       .addNewEnv()
-        .withName(ZK_HOSTS)
-        .withNewValueFrom()
-        .withConfigMapKeyRef(
-          new ConfigMapKeySelector(ZK_HOSTS, clusterConfMap, false))
-        .endValueFrom()
+        .withName(CURRENT_USER_ID)
+        .withValue(userId)
         .endEnv()
       .addNewEnv()
-        .withName(CLUSTER_NAME)
-        .withNewValueFrom()
-        .withConfigMapKeyRef(
-          new ConfigMapKeySelector(CLUSTER_NAME, clusterConfMap, false))
-        .endValueFrom()
+        .withName(USER_GROUPS_IDS)
+        .withValue(userGroupsIds.mkString(" "))
         .endEnv()
+      .addNewEnvFrom()
+        .withNewConfigMapRef()
+          .withName(clusterConfMap)
+          .endConfigMapRef()
+        .endEnvFrom()
+      .addNewEnvFrom()
+        .withNewSecretRef()
+          .withName(clusterUserSecrets)
+          .endSecretRef()
+        .endEnvFrom()
       .withNewResources()
         .addToRequests("memory", executorMemoryQuantity)
         .addToLimits("memory", executorMemoryLimitQuantity)
diff --git a/resource-managers/kubernetes/docker/src/main/dockerfiles/spark/entrypoint.sh b/resource-managers/kubernetes/docker/src/main/dockerfiles/spark/entrypoint.sh
@@ -139,10 +139,34 @@ case "$SPARK_K8S_CMD" in
     exit 1
 esac
 
+function createUserGroups() {
+  groups=($USER_GROUPS)
+  groupIds=($USER_GROUPS_IDS)
+
+  for i in "${!groups[@]}"
+  do
+    groupadd -f -g ${groupIds[i]} ${groups[i]}
+    usermod -a -G  ${groups[i]} $CURRENT_USER
+  done
+}
+
+function createUser() {
+  if ! id $CURRENT_USER >/dev/null 2>&1; then
+    adduser -u $USER_ID $CURRENT_USER
+  fi
+}
+
 #Run configure.sh
 if [ ! $SPARK_K8S_CMD == "init" ]; then
   /opt/mapr/server/configure.sh -c -C $MAPR_CLDB_HOSTS -Z $MAPR_ZK_HOSTS -N $MAPR_CLUSTER
-fi
+  createUser
+  createUserGroups
+  if [ $SPARK_K8S_CMD == "executor" ]; then
+    chown $CURRENT_USER ./
+  fi
 
-# Execute the container CMD under tini for better hygiene
+  exec sudo -u $CURRENT_USER "${CMD[@]}"
+else
+# Execute the container CMD
 exec "${CMD[@]}"
+fi
