Merge pull request #1027 from saLeox/dev-1.0.3

Support proxy user with kerberos
apache · Oct 13, 2021 · b8e2017 · b8e2017
2 parents 509009f + c6e177b
commit b8e2017
Show file tree

Hide file tree

Showing 13 changed files with 42 additions and 21 deletions.
diff --git a/.gitignore b/.gitignore
@@ -2,7 +2,6 @@
 .idea
 .DS_Store
 
-
 out/
 linkis.ipr
 linkis.iws

diff --git a/...-common/src/main/scala/com/webank/wedatasphere/linkis/hadoop/common/conf/HadoopConf.scala b/...-common/src/main/scala/com/webank/wedatasphere/linkis/hadoop/common/conf/HadoopConf.scala
@@ -28,6 +28,10 @@ object HadoopConf {
 
   val KEYTAB_HOST_ENABLED = CommonVars("wds.linkis.keytab.host.enabled", false)
 
+  val KEYTAB_PROXYUSER_ENABLED = CommonVars("wds.linkis.keytab.proxyuser.enable", false)
+
+  val KEYTAB_PROXYUSER_SUPERUSER = CommonVars("wds.linkis.keytab.proxyuser.superuser", "hadoop")
+
   val hadoopConfDir = CommonVars("hadoop.config.dir", CommonVars("HADOOP_CONF_DIR", "").getValue).getValue
 
   val HADOOP_EXTERNAL_CONF_DIR_PREFIX = CommonVars("wds.linkis.hadoop.external.conf.dir.prefix", "/appcom/config/external-conf/hadoop")

diff --git a/...-common/src/main/scala/com/webank/wedatasphere/linkis/hadoop/common/utils/HDFSUtils.scala b/...-common/src/main/scala/com/webank/wedatasphere/linkis/hadoop/common/utils/HDFSUtils.scala
@@ -130,15 +130,23 @@ object HDFSUtils extends Logging {
           }
         }
 
-   def getUserGroupInformation(userName: String): UserGroupInformation = {
+  def getUserGroupInformation(userName: String): UserGroupInformation = {
       if (KERBEROS_ENABLE.getValue) {
-        val path = new File(KEYTAB_FILE.getValue, userName + ".keytab").getPath
-      val user = getKerberosUser(userName)
-      UserGroupInformation.setConfiguration(getConfiguration(userName))
-      UserGroupInformation.loginUserFromKeytabAndReturnUGI(user, path)
-    } else {
-      UserGroupInformation.createRemoteUser(userName)
-    }
+        if (!KEYTAB_PROXYUSER_ENABLED.getValue) {
+          val path = new File(KEYTAB_FILE.getValue, userName + ".keytab").getPath
+          val user = getKerberosUser(userName)
+          UserGroupInformation.setConfiguration(getConfiguration(userName))
+          UserGroupInformation.loginUserFromKeytabAndReturnUGI(user, path)
+        } else {
+          val superUser = KEYTAB_PROXYUSER_SUPERUSER.getValue
+          val path = new File(KEYTAB_FILE.getValue, superUser + ".keytab").getPath
+          val user = getKerberosUser(superUser)
+          UserGroupInformation.setConfiguration(getConfiguration(superUser))
+          UserGroupInformation.createProxyUser(userName, UserGroupInformation.loginUserFromKeytabAndReturnUGI(user, path))
+        }
+      } else {
+        UserGroupInformation.createRemoteUser(userName)
+      }
   }
 
   def getKerberosUser(userName: String): String = {

diff --git a/...gineconn-plugins/engineconn-plugins/flink/src/main/resources/linkis-engineconn.properties b/...gineconn-plugins/engineconn-plugins/flink/src/main/resources/linkis-engineconn.properties
@@ -27,5 +27,3 @@ wds.linkis.engineconn.plugin.default.class=com.webank.wedatasphere.linkis.engine
 wds.linkis.engine.connector.hooks=com.webank.wedatasphere.linkis.engineconn.computation.executor.hook.ComputationEngineConnHook,com.webank.wedatasphere.linkis.engineconn.computation.executor.hook.JarUdfEngineHook
 
 wds.linkis.engineconn.executor.manager.class=com.webank.wedatasphere.linkis.engineconnplugin.flink.executormanager.FlinkExecutorManager
-
-
diff --git a/...ngineconn-plugins/engineconn-plugins/hive/src/main/resources/linkis-engineconn.properties b/...ngineconn-plugins/engineconn-plugins/hive/src/main/resources/linkis-engineconn.properties
@@ -26,4 +26,4 @@ wds.linkis.engineconn.plugin.default.class=com.webank.wedatasphere.linkis.engine
 
 wds.linkis.bdp.hive.init.sql.enable=true
 
-wds.linkis.engine.connector.hooks=com.webank.wedatasphere.linkis.engineconn.computation.executor.hook.ComputationEngineConnHook,com.webank.wedatasphere.linkis.engineconn.computation.executor.hook.JarUdfEngineHook
+wds.linkis.engine.connector.hooks=com.webank.wedatasphere.linkis.engineconn.computation.executor.hook.ComputationEngineConnHook,com.webank.wedatasphere.linkis.engineconn.computation.executor.hook.JarUdfEngineHook
diff --git a/...la/com/webank/wedatasphere/linkis/engineplugin/hive/executor/HiveEngineConnExecutor.scala b/...la/com/webank/wedatasphere/linkis/engineplugin/hive/executor/HiveEngineConnExecutor.scala
@@ -49,6 +49,7 @@ import java.util.concurrent.atomic.AtomicBoolean
 import scala.collection.JavaConversions._
 import scala.collection.mutable
 import scala.collection.mutable.ArrayBuffer
+import com.webank.wedatasphere.linkis.hadoop.common.conf.HadoopConf
 
 class HiveEngineConnExecutor(id: Int,
                              sessionState: SessionState,
@@ -90,6 +91,9 @@ class HiveEngineConnExecutor(id: Int,
 
   override def init(): Unit = {
     LOG.info(s"Ready to change engine state!")
+    if (HadoopConf.KEYTAB_PROXYUSER_ENABLED.getValue) {
+      System.setProperty("javax.security.auth.useSubjectCredsOnly", "false");
+    }
     setCodeParser(new SQLCodeParser)
     super.init()
   }

diff --git a/...neconn-plugins/engineconn-plugins/io_file/src/main/resources/linkis-engineconn.properties b/...neconn-plugins/engineconn-plugins/io_file/src/main/resources/linkis-engineconn.properties
@@ -30,4 +30,4 @@ wds.linkis.engineconn.io.version=1
 wds.linkis.engineconn.support.parallelism=true
 
 wds.linkis.engineconn.max.free.time=0
-wds.linkis.engine.push.log.enable=false
+wds.linkis.engine.push.log.enable=false
diff --git a/...ngineconn-plugins/engineconn-plugins/jdbc/src/main/resources/linkis-engineconn.properties b/...ngineconn-plugins/engineconn-plugins/jdbc/src/main/resources/linkis-engineconn.properties
@@ -26,4 +26,3 @@ wds.linkis.engineconn.plugin.default.class=com.webank.wedatasphere.linkis.manage
 #wds.linkis.engine.io.opts=" -Dfile.encoding=UTF-8  -agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=49100 "
 
 wds.linkis.engineconn.support.parallelism=true
-
diff --git a/...econn-plugins/engineconn-plugins/pipeline/src/main/resources/linkis-engineconn.properties b/...econn-plugins/engineconn-plugins/pipeline/src/main/resources/linkis-engineconn.properties
@@ -26,4 +26,3 @@ wds.linkis.engineconn.debug.enable=true
 wds.linkis.engineconn.plugin.default.class=com.webank.wedatasphere.linkis.manager.engineplugin.pipeline.PipelineEngineConnPlugin
 
 wds.linkis.engineconn.max.free.time=5m
-
diff --git a/...ineconn-plugins/engineconn-plugins/python/src/main/resources/linkis-engineconn.properties b/...ineconn-plugins/engineconn-plugins/python/src/main/resources/linkis-engineconn.properties
@@ -25,4 +25,4 @@ wds.linkis.engineconn.debug.enable=true
 
 wds.linkis.engineconn.plugin.default.class=com.webank.wedatasphere.linkis.manager.engineplugin.python.PythonEngineConnPlugin
 
-wds.linkis.engine.connector.hooks=com.webank.wedatasphere.linkis.engineconn.computation.executor.hook.ComputationEngineConnHook,com.webank.wedatasphere.linkis.engineconn.computation.executor.hook.PyFunctionEngineHook
+wds.linkis.engine.connector.hooks=com.webank.wedatasphere.linkis.engineconn.computation.executor.hook.ComputationEngineConnHook,com.webank.wedatasphere.linkis.engineconn.computation.executor.hook.PyFunctionEngineHook
diff --git a/...gineconn-plugins/engineconn-plugins/spark/src/main/resources/linkis-engineconn.properties b/...gineconn-plugins/engineconn-plugins/spark/src/main/resources/linkis-engineconn.properties
@@ -27,4 +27,3 @@ wds.linkis.engineconn.debug.enable=true
 wds.linkis.engineconn.plugin.default.class=com.webank.wedatasphere.linkis.engineplugin.spark.SparkEngineConnPlugin
 
 wds.linkis.engine.connector.hooks=com.webank.wedatasphere.linkis.engineconn.computation.executor.hook.ComputationEngineConnHook,com.webank.wedatasphere.linkis.engineconn.computation.executor.hook.PyUdfEngineHook,com.webank.wedatasphere.linkis.engineconn.computation.executor.hook.ScalaUdfEngineHook,com.webank.wedatasphere.linkis.engineconn.computation.executor.hook.JarUdfEngineHook,com.webank.wedatasphere.linkis.engineconn.computation.executor.hook.PyFunctionEngineHook,com.webank.wedatasphere.linkis.engineconn.computation.executor.hook.ScalaFunctionEngineHook
-
diff --git a/...n/scala/com/webank/wedatasphere/linkis/engineplugin/spark/config/SparkConfiguration.scala b/...n/scala/com/webank/wedatasphere/linkis/engineplugin/spark/config/SparkConfiguration.scala
@@ -79,7 +79,6 @@ object SparkConfiguration extends Logging {
 
   val IS_VIEWFS_ENV = CommonVars("wds.linkis.spark.engine.is.viewfs.env", true)
 
-
   private def getMainJarName(): String = {
     val somePath = ClassUtils.jarOfClass(classOf[SparkEngineConnFactory])
     if (somePath.isDefined) {

diff --git a/...tasphere/linkis/engineplugin/spark/launch/SparkSubmitProcessEngineConnLaunchBuilder.scala b/...tasphere/linkis/engineplugin/spark/launch/SparkSubmitProcessEngineConnLaunchBuilder.scala
@@ -32,9 +32,11 @@ import com.webank.wedatasphere.linkis.manager.label.entity.Label
 import com.webank.wedatasphere.linkis.manager.label.entity.engine.UserCreatorLabel
 import com.webank.wedatasphere.linkis.protocol.UserWithCreator
 import org.apache.commons.lang.StringUtils
-
 import scala.collection.JavaConverters._
 import scala.collection.mutable.ArrayBuffer
+import com.webank.wedatasphere.linkis.hadoop.common.conf.HadoopConf
+
+
 /**
 
  *
@@ -159,10 +161,14 @@ class SparkSubmitProcessEngineConnLaunchBuilder private extends JavaProcessEngin
       }
     }
 
-
     addOpt("--master", _master)
     addOpt("--deploy-mode", _deployMode)
     addOpt("--name", _name)
+
+    if (HadoopConf.KEYTAB_PROXYUSER_ENABLED.getValue && _proxyUser.nonEmpty) {
+      addOpt("--proxy-user", _proxyUser)
+    }
+
     //addOpt("--jars",Some(ENGINEMANAGER_JAR.getValue))
 //    info("No need to add jars for " + _jars.map(fromPath).exists(x => x.equals("hdfs:///")).toString())
     _jars = _jars.filter(_.isNotBlankPath())
@@ -364,7 +370,6 @@ class SparkSubmitProcessEngineConnLaunchBuilder private extends JavaProcessEngin
       val file = new java.io.File(x.path)
       file.isFile
     }).foreach(jar)
-    proxyUser(getValueAndRemove(properties, "proxyUser", ""))
     if (null != darResource) {
       this.queue(darResource.yarnResource.queueName)
     } else {
@@ -387,6 +392,13 @@ class SparkSubmitProcessEngineConnLaunchBuilder private extends JavaProcessEngin
       }
     }
     }
+
+    if (!HadoopConf.KEYTAB_PROXYUSER_ENABLED.getValue) {
+      this.proxyUser(getValueAndRemove(properties, "proxyUser", ""))
+    } else {
+      this.proxyUser(this._userWithCreator.user)
+    }
+
     //deal spark conf and spark.hadoop.*
     val iterator = properties.entrySet().iterator()
     val sparkConfKeys = ArrayBuffer[String]()