diff --git a/conf/livy.conf.template b/conf/livy.conf.template index f489d76fc..3e5aea12f 100644 --- a/conf/livy.conf.template +++ b/conf/livy.conf.template @@ -275,6 +275,10 @@ # details, routes, etc...) # livy.server.kubernetes.poll-interval = 15s +# Whether Livy fetches executor pods each poll cycle (used for executor log URLs and +# per-executor diagnostics). Disable on large clusters; driver pod is polled regardless. +# livy.server.kubernetes.executor-tracking.enabled = true + # Weather to create Kubernetes Nginx Ingress for Spark UI. If set to true, configure the desired # options below # livy.server.kubernetes.ingress.create = false diff --git a/server/src/main/scala/org/apache/livy/LivyConf.scala b/server/src/main/scala/org/apache/livy/LivyConf.scala index 03bdf0fca..ea6bee692 100644 --- a/server/src/main/scala/org/apache/livy/LivyConf.scala +++ b/server/src/main/scala/org/apache/livy/LivyConf.scala @@ -306,6 +306,11 @@ object LivyConf { // How often Livy polls Kubernetes to refresh Kubernetes app state. val KUBERNETES_POLL_INTERVAL = Entry("livy.server.kubernetes.poll-interval", "15s") + // Whether Livy fetches executor pods each poll cycle (used for executor log URLs + // and per-executor diagnostics). Driver pod is polled regardless. + val KUBERNETES_EXECUTOR_TRACKING_ENABLED = + Entry("livy.server.kubernetes.executor-tracking.enabled", true) + // How long to check livy session leakage. val KUBERNETES_APP_LEAKAGE_CHECK_TIMEOUT = Entry("livy.server.kubernetes.app-leakage.check-timeout", "600s") 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 838de20f7..bb8eb291b 100644 --- a/server/src/main/scala/org/apache/livy/utils/SparkKubernetesApp.scala +++ b/server/src/main/scala/org/apache/livy/utils/SparkKubernetesApp.scala @@ -178,6 +178,11 @@ object SparkKubernetesApp extends Logging { livyConf.getTimeAsMs(LivyConf.KUBERNETES_APP_LEAKAGE_CHECK_INTERVAL) sessionLeakageCheckTimeout = livyConf.getTimeAsMs(LivyConf.KUBERNETES_APP_LEAKAGE_CHECK_TIMEOUT) + if (!livyConf.getBoolean(LivyConf.KUBERNETES_EXECUTOR_TRACKING_ENABLED)) { + info("Kubernetes executor tracking is disabled. Per-executor log URLs and " + + "per-executor entries in session diagnostics will be omitted.") + } + leakedAppsGCThread.setDaemon(true) leakedAppsGCThread.setName("LeakedAppsGCThread") leakedAppsGCThread.start() @@ -721,12 +726,27 @@ private[utils] object KubernetesExtensions { cacheLogSize: Int, appTagLabel: String = SPARK_APP_TAG_LABEL ): KubernetesAppReport = { - val pods = client.pods.inNamespace(app.getApplicationNamespace) - .withLabels(Map(appTagLabel -> app.getApplicationTag).asJava) - .list.getItems.asScala.toSeq - val driver = pods.find(_.getMetadata.getLabels.get(SPARK_ROLE_LABEL) == SPARK_ROLE_DRIVER) - val executors = - pods.filter(_.getMetadata.getLabels.get(SPARK_ROLE_LABEL) == SPARK_ROLE_EXECUTOR) + // Narrow the LIST to the driver pod; application state does not depend on executors. + val driver = client.pods.inNamespace(app.getApplicationNamespace) + .withLabels(Map( + appTagLabel -> app.getApplicationTag, + SPARK_ROLE_LABEL -> SPARK_ROLE_DRIVER + ).asJava) + .list.getItems.asScala.headOption + + // Executors are used only for log URLs and per-executor diagnostics; skip when disabled. + val executors: Seq[Pod] = + if (livyConf.getBoolean(LivyConf.KUBERNETES_EXECUTOR_TRACKING_ENABLED)) { + client.pods.inNamespace(app.getApplicationNamespace) + .withLabels(Map( + appTagLabel -> app.getApplicationTag, + SPARK_ROLE_LABEL -> SPARK_ROLE_EXECUTOR + ).asJava) + .list.getItems.asScala + } else { + Seq.empty + } + val appLog = Try( client.pods.inNamespace(app.getApplicationNamespace) .withName(app.getApplicationPod.getMetadata.getName) 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 5de2ac83d..759cbed10 100644 --- a/server/src/test/scala/org/apache/livy/utils/SparkKubernetesAppSpec.scala +++ b/server/src/test/scala/org/apache/livy/utils/SparkKubernetesAppSpec.scala @@ -149,6 +149,32 @@ class SparkKubernetesAppSpec extends AnyFunSpec with LivyBaseUnitTestSuite with .getExecutorsLogUrls.isEmpty) } + it("should return diagnostics without executor entries when executors is empty") { + // When livy.server.kubernetes.executor-tracking.enabled=false, the + // executor LIST in getApplicationReport is skipped and executors is + // passed in as Seq.empty. Diagnostics must still render the driver + // section without error. + val driverStatus = when(mock[PodStatus].getPhase).thenReturn("Running") + .getMock[PodStatus] + val driverMeta = when(mock[ObjectMeta].getName).thenReturn("driver-pod") + .getMock[ObjectMeta] + when(driverMeta.getNamespace).thenReturn("ns") + when(driverMeta.getLabels).thenReturn(Map.empty[String, String].asJava) + val driverSpec = when(mock[PodSpec].getNodeName).thenReturn("node-1") + .getMock[PodSpec] + when(driverSpec.getContainers).thenReturn(java.util.Collections.emptyList[Container]) + when(driverStatus.getConditions).thenReturn(java.util.Collections.emptyList[PodCondition]) + val driver = when(mock[Pod].getStatus).thenReturn(driverStatus).getMock[Pod] + when(driver.getMetadata).thenReturn(driverMeta) + when(driver.getSpec).thenReturn(driverSpec) + + val diagnostics = KubernetesAppReport( + Some(driver), Seq.empty, IndexedSeq.empty, None, new LivyConf(false) + ).getApplicationDiagnostics + assert(diagnostics.exists(_.contains("driver-pod"))) + assert(!diagnostics.exists(_.contains("executor"))) + } + it("should return driver ingress url") { def livyConf(protocol: Option[String]): LivyConf = { @@ -236,6 +262,13 @@ class SparkKubernetesAppSpec extends AnyFunSpec with LivyBaseUnitTestSuite with KubernetesClientFactory.createKubernetesClient(conf) } } + + it("should enable executor tracking by default") { + // Preserve existing behavior: operators must opt in to skipping the + // executor LIST. This guards against an accidental default flip that + // would silently drop executor entries from session diagnostics. + assert(new LivyConf(false).getBoolean(LivyConf.KUBERNETES_EXECUTOR_TRACKING_ENABLED)) + } } describe("KubernetesClientExtensions") {