[LIVY-1072] Reduce Kubernetes API load in SparkKubernetesApp polling

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

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")

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)

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") {