Spark on k8s : changes for using supporting multi tenant(multi namespace) k8s cluster

pom.xml

@@ -83,7 +83,7 @@
     <slf4j.version>1.7.36</slf4j.version>
     <spark.scala-2.12.version>3.3.4</spark.scala-2.12.version>
     <spark.version>${spark.scala-2.12.version}</spark.version>
-    <kubernetes.client.version>5.6.0</kubernetes.client.version>
+    <kubernetes.client.version>6.8.1</kubernetes.client.version>
     <hive.version>3.0.0</hive.version>
     <commons-codec.version>1.15</commons-codec.version>
     <commons-lang3.version>3.17.0</commons-lang3.version>
@@ -1417,6 +1417,13 @@
   </reporting>
 
   <profiles>
+    <profile>
+      <id>hadoop3</id>
+      <properties>
+        <hadoop.major-minor.version>3</hadoop.major-minor.version>
+        <hadoop.version>3.4.0</hadoop.version>
+      </properties>
+    </profile>
     <profile>
       <id>hadoop2</id>
       <properties>
@@ -1450,7 +1457,7 @@
         <java.version>1.8</java.version>
         <py4j.version>0.10.9.7</py4j.version>
         <json4s.version>3.7.0-M11</json4s.version>
-        <netty.version>4.1.96.Final</netty.version>
+        <netty.version>4.1.108.Final</netty.version>
         <jackson.version>2.15.2</jackson.version>
         <jackson-databind.version>2.15.2</jackson-databind.version>
         <spark.bin.name>spark-${spark.version}-bin-hadoop${hadoop.major-minor.version}</spark.bin.name>

server/src/main/scala/org/apache/livy/server/batch/BatchSession.scala

@@ -40,6 +40,9 @@ case class BatchRecoveryMetadata(
     appTag: String,
     owner: String,
     proxyUser: Option[String],
+    // Empty means "unknown": metadata persisted before multi-namespace support has no
+    // namespace and deserializes to null, which SparkKubernetesApp resolves at recovery.
+    namespace: String = "",
     version: Int = 1)
   extends RecoveryMetadata
 
@@ -64,7 +67,7 @@ object BatchSession extends Logging {
       mockApp: Option[SparkApp] = None): BatchSession = {
     val appTag = s"livy-batch-$id-${Random.alphanumeric.take(8).mkString}".toLowerCase()
     val impersonatedUser = accessManager.checkImpersonation(proxyUser, owner)
-
+    val namespace = SparkApp.getNamespace(request.conf, livyConf)
     def createSparkApp(s: BatchSession): SparkApp = {
       val conf = SparkApp.prepareSparkConf(
         appTag,
@@ -106,7 +109,8 @@ object BatchSession extends Logging {
           childProcesses.decrementAndGet()
         }
       }
-      SparkApp.create(appTag, None, Option(sparkSubmit), livyConf, Option(s))
+      val extrasMap: Map[String, String] = Map(SparkApp.SPARK_KUBERNETES_NAMESPACE_KEY -> namespace)
+      SparkApp.create(appTag, None, Option(sparkSubmit), livyConf, Option(s), extrasMap)
     }
 
     info(s"Creating batch session $id: [owner: $owner, request: $request]")
@@ -120,6 +124,7 @@ object BatchSession extends Logging {
       owner,
       impersonatedUser,
       sessionStore,
+      namespace,
       mockApp.map { m => (_: BatchSession) => m }.getOrElse(createSparkApp))
   }
 
@@ -137,8 +142,10 @@ object BatchSession extends Logging {
       m.owner,
       m.proxyUser,
       sessionStore,
+      m.namespace,
       mockApp.map { m => (_: BatchSession) => m }.getOrElse { s =>
-        SparkApp.create(m.appTag, m.appId, None, livyConf, Option(s))
+        val extrasMap = Map(SparkApp.SPARK_KUBERNETES_NAMESPACE_KEY -> m.namespace)
+        SparkApp.create(m.appTag, m.appId, None, livyConf, Option(s), extrasMap)
       })
   }
 }
@@ -152,6 +159,7 @@ class BatchSession(
     owner: String,
     override val proxyUser: Option[String],
     sessionStore: SessionStore,
+    namespace: String,
     sparkApp: BatchSession => SparkApp)
   extends Session(id, name, owner, livyConf) with SparkAppListener {
   import BatchSession._
@@ -204,5 +212,5 @@ class BatchSession(
   override def infoChanged(appInfo: AppInfo): Unit = { this.appInfo = appInfo }
 
   override def recoveryMetadata: RecoveryMetadata =
-    BatchRecoveryMetadata(id, name, appId, appTag, owner, proxyUser)
+    BatchRecoveryMetadata(id, name, appId, appTag, owner, proxyUser, namespace)
 }

server/src/main/scala/org/apache/livy/server/interactive/InteractiveSession.scala

@@ -70,6 +70,9 @@ case class InteractiveRecoveryMetadata(
     // proxyUser is deprecated. It is available here only for backward compatibility
     proxyUser: Option[String],
     rscDriverUri: Option[URI],
+    // Empty means "unknown": metadata persisted before multi-namespace support has no
+    // namespace and deserializes to null, which SparkKubernetesApp resolves at recovery.
+    namespace: String = "",
     version: Int = 1)
   extends RecoveryMetadata
 
@@ -99,6 +102,7 @@ object InteractiveSession extends Logging {
       mockClient: Option[RSCClient] = None): InteractiveSession = {
     val appTag = s"livy-session-$id-${Random.alphanumeric.take(8).mkString}".toLowerCase()
     val impersonatedUser = accessManager.checkImpersonation(proxyUser, owner)
+    val namespace = SparkApp.getNamespace(request.conf, livyConf)
 
     val client = mockClient.orElse {
       val conf = SparkApp.prepareSparkConf(appTag, livyConf, prepareConf(
@@ -159,6 +163,7 @@ object InteractiveSession extends Logging {
       request.numExecutors,
       request.pyFiles,
       request.queue.filterNot(_.isEmpty).orElse(livyConf.sparkYarnQueue()),
+      namespace,
       mockApp)
   }
 
@@ -199,6 +204,7 @@ object InteractiveSession extends Logging {
       metadata.numExecutors,
       metadata.pyFiles,
       metadata.queue,
+      metadata.namespace,
       mockApp)
   }
 
@@ -438,6 +444,7 @@ class InteractiveSession(
     val numExecutors: Option[Int],
     val pyFiles: List[String],
     val queue: Option[String],
+    val namespace: String,
     mockApp: Option[SparkApp]) // For unit test.
   extends Session(id, name, owner, ttl, idleTimeout, livyConf)
   with SessionHeartbeat
@@ -467,11 +474,14 @@ class InteractiveSession(
     app = mockApp.orElse {
       val driverProcess = client.flatMap { c => Option(c.getDriverProcess) }
         .map(new LineBufferedProcess(_, livyConf.getInt(LivyConf.SPARK_LOGS_SIZE)))
+      val extrasMap = Map(SparkApp.SPARK_KUBERNETES_NAMESPACE_KEY -> namespace)
       if (!livyConf.isRunningOnKubernetes()) {
-        driverProcess.map(_ => SparkApp.create(appTag, appId, driverProcess, livyConf, Some(this)))
+        driverProcess.map { _ =>
+          SparkApp.create(appTag, appId, driverProcess, livyConf, Some(this), extrasMap)
+        }
       } else {
         // Create SparkApp for Kubernetes anyway
-        Some(SparkApp.create(appTag, appId, driverProcess, livyConf, Some(this)))
+        Some(SparkApp.create(appTag, appId, driverProcess, livyConf, Some(this), extrasMap))
       }
     }
 
@@ -540,7 +550,7 @@ class InteractiveSession(
       heartbeatTimeout.toSeconds.toInt, owner, ttl, idleTimeout,
       driverMemory, driverCores, executorMemory, executorCores, conf,
       archives, files, jars, numExecutors, pyFiles, queue,
-      proxyUser, rscDriverUri)
+      proxyUser, rscDriverUri, namespace)
 
   override def state: SessionState = {
     if (serverSideState == SessionState.Running) {

server/src/main/scala/org/apache/livy/utils/SparkApp.scala

@@ -17,9 +17,11 @@
 
 package org.apache.livy.utils
 
-import java.io.IOException
+import java.io.{File, FileInputStream, IOException}
+import java.util.Properties
 
 import scala.collection.JavaConverters._
+import scala.util.control.NonFatal
 
 import org.apache.hadoop.conf.Configuration
 
@@ -60,12 +62,60 @@ trait SparkAppListener {
  */
 object SparkApp extends Logging {
   private val SPARK_YARN_TAG_KEY = "spark.yarn.tags"
-
+  val SPARK_KUBERNETES_NAMESPACE_KEY = "spark.kubernetes.namespace"
   object State extends Enumeration {
     val STARTING, RUNNING, FINISHED, FAILED, KILLED = Value
   }
   type State = State.Value
 
+  val DEFAULT_KUBERNETES_NAMESPACE = "default"
+
+  /**
+   * Resolve the Kubernetes namespace a Spark application should run in.
+   *
+   * The namespace is looked up, in order of precedence, from:
+   *   1. the session's Spark configuration ([[SPARK_KUBERNETES_NAMESPACE_KEY]]),
+   *   2. `$SPARK_HOME/conf/spark-defaults.conf` (if present),
+   *   3. the [[DEFAULT_KUBERNETES_NAMESPACE]] fallback.
+   *
+   * The namespace is only meaningful on Kubernetes, so for any other cluster
+   * manager (YARN, local) an empty string is returned without touching the
+   * filesystem.
+   */
+  def getNamespace(conf: Map[String, String], livyConf: LivyConf): String = {
+    if (!livyConf.isRunningOnKubernetes()) {
+      return ""
+    }
+    conf.get(SPARK_KUBERNETES_NAMESPACE_KEY).filter(_.nonEmpty).getOrElse {
+      namespaceFromSparkDefaults(livyConf).getOrElse(DEFAULT_KUBERNETES_NAMESPACE)
+    }
+  }
+
+  private def namespaceFromSparkDefaults(livyConf: LivyConf): Option[String] = {
+    livyConf.sparkHome().flatMap { sparkHome =>
+      val sparkDefaults = new File(sparkHome, s"conf${File.separator}spark-defaults.conf")
+      if (!sparkDefaults.isFile) {
+        None
+      } else {
+        val in = new FileInputStream(sparkDefaults)
+        try {
+          val properties = new Properties()
+          properties.load(in)
+          Option(properties.getProperty(SPARK_KUBERNETES_NAMESPACE_KEY)).filter(_.nonEmpty)
+        } catch {
+          case NonFatal(e) =>
+            // A malformed spark-defaults.conf (e.g. an invalid unicode escape, which
+            // java.util.Properties rejects) must not abort session creation; fall
+            // back to the default namespace instead.
+            warn(s"Could not read $sparkDefaults for the Kubernetes namespace; " +
+              s"falling back to the default namespace: ${e.getMessage}")
+            None
+        } finally {
+          in.close()
+        }
+      }
+    }
+  }
   /**
    * Return cluster manager dependent SparkConf.
    *
@@ -152,11 +202,12 @@ object SparkApp extends Logging {
       appId: Option[String],
       process: Option[LineBufferedProcess],
       livyConf: LivyConf,
-      listener: Option[SparkAppListener]): SparkApp = {
+      listener: Option[SparkAppListener],
+      extrasMap: Map[String, String]): SparkApp = {
     if (livyConf.isRunningOnYarn()) {
       new SparkYarnApp(uniqueAppTag, appId, process, listener, livyConf)
     } else if (livyConf.isRunningOnKubernetes()) {
-      new SparkKubernetesApp(uniqueAppTag, appId, process, listener, livyConf)
+      new SparkKubernetesApp(uniqueAppTag, appId, process, listener, livyConf, extrasMap)
     } else {
       require(process.isDefined, "process must not be None when Livy master is not YARN or" +
         "Kubernetes.")