diff --git a/pom.xml b/pom.xml index 0bb42545a..b54b8e69f 100644 --- a/pom.xml +++ b/pom.xml @@ -82,7 +82,7 @@ compile 1.7.36 4.1.2 - 5.6.0 + 6.8.1 3.0.0 1.15 3.17.0 @@ -1461,7 +1461,7 @@ 1.8 0.10.9.7 3.7.0-M11 - 4.1.96.Final + 4.1.108.Final 2.15.2 2.15.2 spark-${spark.version}-bin-hadoop${hadoop.major-minor.version} diff --git a/server/src/main/scala/org/apache/livy/server/batch/BatchSession.scala b/server/src/main/scala/org/apache/livy/server/batch/BatchSession.scala index 0d9a0c644..1f8a49d6c 100644 --- a/server/src/main/scala/org/apache/livy/server/batch/BatchSession.scala +++ b/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, @@ -111,7 +114,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]") @@ -125,6 +129,7 @@ object BatchSession extends Logging { owner, impersonatedUser, sessionStore, + namespace, mockApp.map { m => (_: BatchSession) => m }.getOrElse(createSparkApp)) } @@ -142,8 +147,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) }) } } @@ -157,6 +164,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._ @@ -209,5 +217,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) } diff --git a/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSession.scala b/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSession.scala index b193dac80..276f63362 100644 --- a/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSession.scala +++ b/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) } @@ -443,6 +449,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 @@ -472,11 +479,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)) } } @@ -545,7 +555,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) { diff --git a/server/src/main/scala/org/apache/livy/utils/SparkApp.scala b/server/src/main/scala/org/apache/livy/utils/SparkApp.scala index acb2b0e1e..52cda895f 100644 --- a/server/src/main/scala/org/apache/livy/utils/SparkApp.scala +++ b/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,73 @@ 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-defaults.conf` in the Spark config directory (`$SPARK_CONF_DIR` if set, + * otherwise `$SPARK_HOME/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) + } + } + + /** + * Resolve the Spark configuration directory the same way Spark's launch scripts do: + * honor `$SPARK_CONF_DIR` if set, otherwise fall back to `$SPARK_HOME/conf`. Livy runs + * spark-submit as a child process, which reads spark-defaults.conf from this directory; + * resolving it identically here keeps the monitored namespace consistent with where the + * driver pod is actually created. + */ + private def sparkConfDir(livyConf: LivyConf): Option[String] = { + sys.env.get("SPARK_CONF_DIR").filter(_.nonEmpty) + .orElse(livyConf.sparkHome().map(home => s"$home${File.separator}conf")) + } + + private def namespaceFromSparkDefaults(livyConf: LivyConf): Option[String] = { + sparkConfDir(livyConf).flatMap { confDir => + val sparkDefaults = new File(confDir, "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 +215,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.") diff --git a/server/src/main/scala/org/apache/livy/utils/SparkKubernetesApp.scala b/server/src/main/scala/org/apache/livy/utils/SparkKubernetesApp.scala index bb8eb291b..5773ff93d 100644 --- a/server/src/main/scala/org/apache/livy/utils/SparkKubernetesApp.scala +++ b/server/src/main/scala/org/apache/livy/utils/SparkKubernetesApp.scala @@ -21,6 +21,8 @@ import java.util.Collections import java.util.concurrent._ import scala.annotation.tailrec +import scala.collection.JavaConverters.asScalaSetConverter +import scala.collection.mutable import scala.collection.mutable.ArrayBuffer import scala.concurrent._ import scala.concurrent.duration._ @@ -47,29 +49,47 @@ object SparkKubernetesApp extends Logging { override def run(): Unit = { import KubernetesExtensions._ while (true) { - if (!leakedAppTags.isEmpty) { - // kill the app if found it and remove it if exceeding a threshold - val iter = leakedAppTags.entrySet().iterator() - var isRemoved = false - val now = System.currentTimeMillis() - val apps = withRetry(kubernetesClient.getApplications()) - while (iter.hasNext) { - val entry = iter.next() - apps.find(_.getApplicationTag.contains(entry.getKey)) - .foreach({ - app => - info(s"Kill leaked app ${app.getApplicationId}") - withRetry(kubernetesClient.killApplication(app)) + // Guard the whole cycle: an unexpected error must never terminate this daemon + // thread, otherwise leaked-app GC would silently stop for the whole server. + try { + if (!leakedAppTags.isEmpty) { + // kill the app if found it and remove it if exceeding a threshold + val iter = leakedAppTags.entrySet().iterator() + var isRemoved = false + val now = System.currentTimeMillis() + val apps = appNamespaces.flatMap { namespace => + // Isolate per-namespace failures: losing access to (or deletion of) one + // namespace must not stop GC from sweeping the others this cycle. + try { + withRetry(kubernetesClient.getApplications(namespace)) + } catch { + case NonFatal(e) => + warn(s"Failed to list Spark applications in namespace '$namespace' during " + + s"leaked-app GC; skipping it this cycle: ${e.getMessage}") + Seq.empty[KubernetesApplication] + } + } + while (iter.hasNext) { + val entry = iter.next() + apps.find(_.getApplicationTag.contains(entry.getKey)) + .foreach({ + app => + info(s"Kill leaked app ${app.getApplicationId}") + withRetry(kubernetesClient.killApplication(app)) + iter.remove() + isRemoved = true + }) + if (!isRemoved) { + if ((entry.getValue - now) > sessionLeakageCheckTimeout) { iter.remove() - isRemoved = true - }) - if (!isRemoved) { - if ((entry.getValue - now) > sessionLeakageCheckTimeout) { - iter.remove() - info(s"Remove leaked Kubernetes app tag ${entry.getKey}") + info(s"Remove leaked Kubernetes app tag ${entry.getKey}") + } } } } + } catch { + case NonFatal(e) => + error("Unexpected error during leaked-application GC; retrying next cycle.", e) } Thread.sleep(sessionLeakageCheckInterval) } @@ -156,6 +176,8 @@ object SparkKubernetesApp extends Logging { private var sessionLeakageCheckInterval: Long = _ var kubernetesClient: DefaultKubernetesClient = _ + var appNamespaces: mutable.Set[String] = + ConcurrentHashMap.newKeySet[String]().asScala private var appLookupThreadPoolSize: Long = _ private var appLookupMaxFailedTimes: Long = _ @@ -164,8 +186,7 @@ object SparkKubernetesApp extends Logging { this.livyConf = livyConf // KubernetesClient is thread safe. Create once, share it across threads. - kubernetesClient = - KubernetesClientFactory.createKubernetesClient(livyConf) + kubernetesClient = KubernetesClientFactory.createKubernetesClient(livyConf) cacheLogSize = livyConf.getInt(LivyConf.SPARK_LOGS_SIZE) appLookupTimeout = livyConf.getTimeAsMs(LivyConf.KUBERNETES_APP_LOOKUP_TIMEOUT).milliseconds @@ -268,7 +289,9 @@ class SparkKubernetesApp private[utils] ( process: Option[LineBufferedProcess], listener: Option[SparkAppListener], livyConf: LivyConf, - kubernetesClient: => KubernetesClient = SparkKubernetesApp.kubernetesClient) // For unit test. + extrasMap: Map[String, String], + // For unit test. + kubernetesClient: => DefaultKubernetesClient = SparkKubernetesApp.kubernetesClient) extends SparkApp with Logging { @@ -285,6 +308,20 @@ class SparkKubernetesApp private[utils] ( private var kubernetesTagToAppIdFailedTimes: Int = _ private var kubernetesAppMonitorFailedTimes: Int = _ + // Recovery metadata written before multi-namespace support has no namespace and + // deserializes to null (jackson-module-scala fills absent reference params with null, + // ignoring Scala default values). Treat null/missing as "" ("unknown") and resolve the + // real namespace once the driver pod is discovered (see monitorSparkKubernetesApp). + private var namespace: String = + Option(extrasMap.getOrElse(SparkApp.SPARK_KUBERNETES_NAMESPACE_KEY, "")).getOrElse("") + // Only track a concrete namespace for the leaked-app GC sweep. An empty namespace means + // "unknown" (e.g. a session recovered from pre-multi-namespace metadata); adding it would + // make every GC cycle fall back to a cluster-wide inAnyNamespace scan, which a + // namespace-scoped service account is not permitted to do. The real namespace is adopted + // and tracked once the driver pod is discovered (see monitorSparkKubernetesApp). + if (namespace.nonEmpty) { + appNamespaces.add(namespace) + } private def failToMonitor(): Unit = { changeState(SparkApp.State.FAILED) process.foreach(_.destroy()) @@ -315,10 +352,11 @@ class SparkKubernetesApp private[utils] ( } // Get KubernetesApplication by appTag. val appOption: Option[KubernetesApplication] = try { - getAppFromTag(appTag, pollInterval, appLookupTimeout.fromNow) + getAppFromTag(appTag, pollInterval, appLookupTimeout.fromNow, namespace) } catch { case e: Exception => failToGetAppId() + error(s"Exception getting app from tag $appTag in namespace $namespace with message: ", e) appPromise.failure(e) return } @@ -327,6 +365,13 @@ class SparkKubernetesApp private[utils] ( return } val app: KubernetesApplication = appOption.get + // For a session recovered from pre-multi-namespace metadata the namespace is unknown + // until the driver pod is found. Adopt the pod's real namespace so the report, ingress + // and leaked-app GC operations below target the correct namespace. + if (namespace == null || namespace.isEmpty) { + namespace = app.getApplicationNamespace + appNamespaces.add(namespace) + } appPromise.trySuccess(app) val appId = app.getApplicationId @@ -463,10 +508,11 @@ class SparkKubernetesApp private[utils] ( private def getAppFromTag( appTag: String, pollInterval: duration.Duration, - deadline: Deadline): Option[KubernetesApplication] = { + deadline: Deadline, + namespace: String): Option[KubernetesApplication] = { import KubernetesExtensions._ - - withRetry(kubernetesClient.getApplications().find(_.getApplicationTag.contains(appTag))) + withRetry(kubernetesClient.getApplications(namespace) + .find(_.getApplicationTag.contains(appTag))) match { case Some(app) => Some(app) case None => @@ -705,19 +751,28 @@ private[utils] object KubernetesExtensions { """.stripMargin def getApplications( + namespace: String = "", labels: Map[String, String] = Map(SPARK_ROLE_LABEL -> SPARK_ROLE_DRIVER), appTagLabel: String = SPARK_APP_TAG_LABEL, appIdLabel: String = SPARK_APP_ID_LABEL ): Seq[KubernetesApplication] = { - client.pods.inAnyNamespace - .withLabels(labels.asJava) - .withLabel(appTagLabel) + // An empty namespace means "unknown" (e.g. a session recovered from pre-multi-namespace + // metadata); scan all namespaces so the driver pod can still be located, matching the + // pre-multi-namespace behavior. A concrete namespace scopes the LIST to that tenant. + val pods = if (namespace == null || namespace.isEmpty) { + client.pods.inAnyNamespace.withLabels(labels.asJava) + } else { + client.pods.inNamespace(namespace).withLabels(labels.asJava) + } + pods.withLabel(appTagLabel) .withLabel(appIdLabel) .list.getItems.asScala.map(new KubernetesApplication(_)).toSeq } def killApplication(app: KubernetesApplication): Boolean = { - client.pods.inAnyNamespace.delete(app.getApplicationPod) + // Scope the delete to the driver pod's own namespace so teardown works under a + // namespaced RBAC role (a cluster-wide inAnyNamespace delete would be forbidden). + client.pods.inNamespace(app.getApplicationNamespace).delete(app.getApplicationPod) } def getApplicationReport( @@ -742,7 +797,7 @@ private[utils] object KubernetesExtensions { appTagLabel -> app.getApplicationTag, SPARK_ROLE_LABEL -> SPARK_ROLE_EXECUTOR ).asJava) - .list.getItems.asScala + .list.getItems.asScala.toSeq } else { Seq.empty } diff --git a/server/src/test/scala/org/apache/livy/server/batch/BatchSessionSpec.scala b/server/src/test/scala/org/apache/livy/server/batch/BatchSessionSpec.scala index 2477eac73..57bfed3a1 100644 --- a/server/src/test/scala/org/apache/livy/server/batch/BatchSessionSpec.scala +++ b/server/src/test/scala/org/apache/livy/server/batch/BatchSessionSpec.scala @@ -199,7 +199,7 @@ class BatchSessionSpec val req = new CreateBatchRequest() val name = Some("Test Batch Session") val mockApp = mock[SparkApp] - val m = BatchRecoveryMetadata(99, name, None, "appTag", null, None) + val m = BatchRecoveryMetadata(99, name, None, "appTag", null, None, "") val batch = BatchSession.recover(m, conf, sessionStore, Some(mockApp)) batch.state shouldBe (SessionState.Recovering) @@ -216,5 +216,14 @@ class BatchSessionSpec testRecoverSession(name) } } + + it("should propagate the stored namespace through recovery metadata") { + val conf = new LivyConf() + val mockApp = mock[SparkApp] + val m = BatchRecoveryMetadata( + 101, Some("ns-session"), None, "appTag", "owner", None, "team-a") + val batch = BatchSession.recover(m, conf, sessionStore, Some(mockApp)) + batch.recoveryMetadata.asInstanceOf[BatchRecoveryMetadata].namespace shouldBe "team-a" + } } } diff --git a/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionSpec.scala b/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionSpec.scala index 1efa266ac..bb7eb921f 100644 --- a/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionSpec.scala +++ b/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionSpec.scala @@ -286,7 +286,7 @@ class InteractiveSessionSpec extends AnyFunSpec val m = InteractiveRecoveryMetadata( 78, Some("Test session"), None, "appTag", Spark, 0, null, None, None, None, None, None, None, Map.empty[String, String], List.empty[String], List.empty[String], - List.empty[String], None, List.empty[String], None, None, Some(URI.create(""))) + List.empty[String], None, List.empty[String], None, None, Some(URI.create("")), "") val s = InteractiveSession.recover(m, conf, sessionStore, None, Some(mockClient)) s.start() @@ -305,7 +305,7 @@ class InteractiveSessionSpec extends AnyFunSpec val m = InteractiveRecoveryMetadata( 78, None, None, "appTag", Spark, 0, null, None, None, None, None, None, None, Map.empty[String, String], List.empty[String], List.empty[String], - List.empty[String], None, List.empty[String], None, None, Some(URI.create(""))) + List.empty[String], None, List.empty[String], None, None, Some(URI.create("")), "") val s = InteractiveSession.recover(m, conf, sessionStore, None, Some(mockClient)) s.start() @@ -322,7 +322,7 @@ class InteractiveSessionSpec extends AnyFunSpec val m = InteractiveRecoveryMetadata( 78, None, Some("appId"), "appTag", Spark, 0, null, None, None, None, None, None, None, Map.empty[String, String], List.empty[String], List.empty[String], - List.empty[String], None, List.empty[String], None, None, None) + List.empty[String], None, List.empty[String], None, None, None, "") val s = InteractiveSession.recover(m, conf, sessionStore, None) s.start() s.state shouldBe a[SessionState.Dead] diff --git a/server/src/test/scala/org/apache/livy/sessions/SessionManagerSpec.scala b/server/src/test/scala/org/apache/livy/sessions/SessionManagerSpec.scala index 8e5557018..7ab557621 100644 --- a/server/src/test/scala/org/apache/livy/sessions/SessionManagerSpec.scala +++ b/server/src/test/scala/org/apache/livy/sessions/SessionManagerSpec.scala @@ -216,7 +216,7 @@ class SessionManagerSpec extends AnyFunSpec with Matchers with LivyBaseUnitTestS implicit def executor: ExecutionContext = ExecutionContext.global def makeMetadata(id: Int, appTag: String): BatchRecoveryMetadata = { - BatchRecoveryMetadata(id, Some(s"test-session-$id"), None, appTag, null, None) + BatchRecoveryMetadata(id, Some(s"test-session-$id"), None, appTag, null, None, "") } def mockSession(id: Int): BatchSession = { diff --git a/server/src/test/scala/org/apache/livy/utils/SparkAppSpec.scala b/server/src/test/scala/org/apache/livy/utils/SparkAppSpec.scala index eb2ba48e3..4a002dea9 100644 --- a/server/src/test/scala/org/apache/livy/utils/SparkAppSpec.scala +++ b/server/src/test/scala/org/apache/livy/utils/SparkAppSpec.scala @@ -17,6 +17,7 @@ package org.apache.livy.utils +import java.io.{File, PrintWriter} import java.nio.file.{Files, Path} import scala.collection.JavaConverters._ @@ -32,6 +33,39 @@ class SparkAppSpec extends AnyFunSpec with LivyBaseUnitTestSuite { private val providerPathKey = "spark.hadoop.hadoop.security.credential.provider.path" private val truststorePasswordKey = "spark.hadoop.hive.metastore.truststore.password" + private def k8sConf(sparkHome: Option[String] = None): LivyConf = { + val conf = new LivyConf(false) + conf.set(LivyConf.LIVY_SPARK_MASTER, "k8s://https://kubernetes.default.svc:443") + sparkHome.foreach(conf.set(LivyConf.SPARK_HOME, _)) + conf + } + + /** Create a throwaway SPARK_HOME with the given spark-defaults.conf contents (or none). */ + private def withSparkHome(defaultsContent: Option[String])(f: String => Unit): Unit = { + // These tests assert resolution from $SPARK_HOME/conf; an ambient SPARK_CONF_DIR takes + // precedence and would shadow the throwaway SPARK_HOME, so skip when it is set. + assume(sys.env.get("SPARK_CONF_DIR").forall(_.isEmpty)) + val sparkHome = Files.createTempDirectory("livy-spark-home").toFile + try { + val confDir = new File(sparkHome, "conf") + assert(confDir.mkdirs()) + defaultsContent.foreach { content => + val writer = new PrintWriter(new File(confDir, "spark-defaults.conf")) + try writer.write(content) finally writer.close() + } + f(sparkHome.getAbsolutePath) + } finally { + deleteRecursively(sparkHome) + } + } + + private def deleteRecursively(file: File): Unit = { + if (file.isDirectory) { + Option(file.listFiles()).foreach(_.foreach(deleteRecursively)) + } + file.delete() + } + private def deleteRecursively(path: Path): Unit = { if (Files.exists(path)) { Files.walk(path).iterator().asScala.toSeq.reverse.foreach(Files.deleteIfExists) @@ -61,6 +95,67 @@ class SparkAppSpec extends AnyFunSpec with LivyBaseUnitTestSuite { } } + describe("SparkApp.getNamespace") { + + it("should return an empty namespace when not running on Kubernetes") { + val conf = new LivyConf(false) + conf.set(LivyConf.LIVY_SPARK_MASTER, "yarn") + // A namespace in the conf must be ignored for non-Kubernetes masters. + val sparkConf = Map(SparkApp.SPARK_KUBERNETES_NAMESPACE_KEY -> "ignored") + assert(SparkApp.getNamespace(sparkConf, conf) === "") + } + + it("should prefer the namespace from the session Spark conf") { + val sparkConf = Map(SparkApp.SPARK_KUBERNETES_NAMESPACE_KEY -> "team-a") + assert(SparkApp.getNamespace(sparkConf, k8sConf()) === "team-a") + } + + it("should fall back to spark-defaults.conf when the conf has no namespace") { + withSparkHome(Some(s"${SparkApp.SPARK_KUBERNETES_NAMESPACE_KEY} team-b\n")) { sparkHome => + assert(SparkApp.getNamespace(Map.empty, k8sConf(Some(sparkHome))) === "team-b") + } + } + + it("should fall back to the default namespace when spark-defaults.conf is absent") { + withSparkHome(None) { sparkHome => + assert(SparkApp.getNamespace(Map.empty, k8sConf(Some(sparkHome))) === + SparkApp.DEFAULT_KUBERNETES_NAMESPACE) + } + } + + it("should fall back to the default namespace when spark-defaults.conf lacks the key") { + withSparkHome(Some("spark.executor.memory 1g\n")) { sparkHome => + assert(SparkApp.getNamespace(Map.empty, k8sConf(Some(sparkHome))) === + SparkApp.DEFAULT_KUBERNETES_NAMESPACE) + } + } + + it("should fall back to the default namespace when SPARK_HOME is not set") { + // Namespace resolution reads SPARK_CONF_DIR then falls back to the SPARK_HOME env var, + // so only assert the env-independent default when the ambient environment sets neither. + assume(sys.env.get("SPARK_HOME").forall(_.isEmpty)) + assume(sys.env.get("SPARK_CONF_DIR").forall(_.isEmpty)) + assert(SparkApp.getNamespace(Map.empty, k8sConf()) === + SparkApp.DEFAULT_KUBERNETES_NAMESPACE) + } + + it("should fall back to the default namespace when spark-defaults.conf is malformed") { + // An invalid unicode escape makes java.util.Properties.load throw; getNamespace must + // swallow it and fall back rather than aborting session creation. + withSparkHome(Some("spark.driver.extraJavaOptions=-Dp=\\uZZZZ\n")) { sparkHome => + assert(SparkApp.getNamespace(Map.empty, k8sConf(Some(sparkHome))) === + SparkApp.DEFAULT_KUBERNETES_NAMESPACE) + } + } + + it("should treat an empty namespace in the conf as unset") { + withSparkHome(Some(s"${SparkApp.SPARK_KUBERNETES_NAMESPACE_KEY} team-c\n")) { sparkHome => + val sparkConf = Map(SparkApp.SPARK_KUBERNETES_NAMESPACE_KEY -> "") + assert(SparkApp.getNamespace(sparkConf, k8sConf(Some(sparkHome))) === "team-c") + } + } + } + describe("SparkApp.prepareSparkConf") { it("should leave conf unchanged when no credential provider path is configured") { val livyConf = new LivyConf(false) diff --git a/server/src/test/scala/org/apache/livy/utils/SparkKubernetesAppSpec.scala b/server/src/test/scala/org/apache/livy/utils/SparkKubernetesAppSpec.scala index 759cbed10..7066588d9 100644 --- a/server/src/test/scala/org/apache/livy/utils/SparkKubernetesAppSpec.scala +++ b/server/src/test/scala/org/apache/livy/utils/SparkKubernetesAppSpec.scala @@ -27,7 +27,7 @@ import org.scalatest.funspec.AnyFunSpec import org.scalatestplus.mockito.MockitoSugar._ import org.apache.livy.{LivyBaseUnitTestSuite, LivyConf} -import org.apache.livy.utils.KubernetesConstants.SPARK_APP_TAG_LABEL +import org.apache.livy.utils.KubernetesConstants.{SPARK_APP_ID_LABEL, SPARK_APP_TAG_LABEL} class SparkKubernetesAppSpec extends AnyFunSpec with LivyBaseUnitTestSuite with BeforeAndAfterAll { @@ -318,4 +318,21 @@ class SparkKubernetesAppSpec extends AnyFunSpec with LivyBaseUnitTestSuite with } } + describe("KubernetesApplication") { + // The namespace/tag/id read here drive namespace adoption on recovery, the leaked-app + // GC sweep and the application report; verify they are surfaced from the driver pod. + it("should expose the driver pod's namespace, tag and id") { + val pod = new PodBuilder().withNewMetadata().withName("driver") + .withNamespace("team-b") + .addToLabels(SPARK_APP_TAG_LABEL, "tag-x") + .addToLabels(SPARK_APP_ID_LABEL, "app-x") + .endMetadata().withNewSpec().endSpec().build() + val app = new KubernetesApplication(pod) + assert(app.getApplicationNamespace === "team-b") + assert(app.getApplicationTag === "tag-x") + assert(app.getApplicationId === "app-x") + assert(app.getApplicationPod === pod) + } + } + }