[LIVY-1082] Add config to disable Kubernetes driver pod log polling

conf/livy.conf.template

@@ -294,6 +294,10 @@
 # per-executor diagnostics). Disable on large clusters; driver pod is polled regardless.
 # livy.server.kubernetes.executor-tracking.enabled = true
 
+# Whether Livy fetches the driver pod's Kubernetes logs each poll cycle. Disable if driver
+# logs are collected externally and freshness in /sessions/:id/log, /batches/:id/log is not needed.
+# livy.server.kubernetes.driver-log-polling.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

@@ -318,6 +318,13 @@ object LivyConf {
   val KUBERNETES_EXECUTOR_TRACKING_ENABLED =
     Entry("livy.server.kubernetes.executor-tracking.enabled", true)
 
+  // Whether Livy fetches the driver pod's Kubernetes logs each poll cycle. When
+  // disabled, Livy still polls pod state and diagnostics, but /sessions/:id/log and
+  // /batches/:id/log won't include fresh Kubernetes driver logs. Useful when driver
+  // logs are collected externally and the repeated pods/<driver>/log calls are not needed.
+  val KUBERNETES_DRIVER_LOG_POLLING_ENABLED =
+    Entry("livy.server.kubernetes.driver-log-polling.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

@@ -182,6 +182,10 @@ object SparkKubernetesApp extends Logging {
       info("Kubernetes executor tracking is disabled. Per-executor log URLs and " +
         "per-executor entries in session diagnostics will be omitted.")
     }
+    if (!livyConf.getBoolean(LivyConf.KUBERNETES_DRIVER_LOG_POLLING_ENABLED)) {
+      info("Kubernetes driver log polling is disabled. Livy will keep polling pod state " +
+        "and diagnostics, but live driver log lines will be omitted from session/batch logs.")
+    }
 
     leakedAppsGCThread.setDaemon(true)
     leakedAppsGCThread.setName("LeakedAppsGCThread")
@@ -693,6 +697,18 @@ private[utils] case class KubernetesAppReport(driver: Option[Pod], executors: Se
 private[utils] object KubernetesExtensions {
   import KubernetesConstants._
 
+  // Skips the driver pod's Kubernetes log fetch when driver log polling is disabled,
+  // rather than paying for the pods/<driver>/log call just to discard the result.
+  private[utils] def resolveDriverAppLog(
+      livyConf: LivyConf,
+      fetchLog: () => IndexedSeq[String]): IndexedSeq[String] = {
+    if (livyConf.getBoolean(LivyConf.KUBERNETES_DRIVER_LOG_POLLING_ENABLED)) {
+      fetchLog()
+    } else {
+      IndexedSeq.empty
+    }
+  }
+
   implicit class KubernetesClientExtensions(client: KubernetesClient) {
     import scala.collection.JavaConverters._
 
@@ -747,11 +763,13 @@ private[utils] object KubernetesExtensions {
           Seq.empty
         }
 
-      val appLog = Try(
+      // The driver pod's Kubernetes logs are used only to populate live log lines in
+      // session/batch responses; skip the pods/<driver>/log call when disabled.
+      val appLog = resolveDriverAppLog(livyConf, () => Try(
         client.pods.inNamespace(app.getApplicationNamespace)
           .withName(app.getApplicationPod.getMetadata.getName)
           .tailingLines(cacheLogSize).getLog.split("\n").toIndexedSeq
-      ).getOrElse(IndexedSeq.empty)
+      ).getOrElse(IndexedSeq.empty))
       val ingress = client.network.v1.ingresses.inNamespace(app.getApplicationNamespace)
         .withLabel(SPARK_APP_TAG_LABEL, app.getApplicationTag)
         .list.getItems.asScala.headOption

server/src/test/scala/org/apache/livy/utils/SparkKubernetesAppSpec.scala

@@ -269,6 +269,37 @@ class SparkKubernetesAppSpec extends AnyFunSpec with LivyBaseUnitTestSuite with
       // would silently drop executor entries from session diagnostics.
       assert(new LivyConf(false).getBoolean(LivyConf.KUBERNETES_EXECUTOR_TRACKING_ENABLED))
     }
+
+    it("should enable driver log polling by default") {
+      // Preserve existing behavior unless operators explicitly disable driver log polling.
+      assert(new LivyConf(false).getBoolean(LivyConf.KUBERNETES_DRIVER_LOG_POLLING_ENABLED))
+    }
+  }
+
+  describe("resolveDriverAppLog") {
+    it("should skip fetching the log when driver log polling is disabled") {
+      var fetchCalled = false
+      val livyConf = new LivyConf(false)
+        .set(LivyConf.KUBERNETES_DRIVER_LOG_POLLING_ENABLED, false)
+
+      val result = KubernetesExtensions.resolveDriverAppLog(livyConf, () => {
+        fetchCalled = true
+        IndexedSeq("should-not-be-returned")
+      })
+
+      assert(!fetchCalled)
+      assert(result.isEmpty)
+    }
+
+    it("should fetch the log when driver log polling is enabled") {
+      val livyConf = new LivyConf(false)
+        .set(LivyConf.KUBERNETES_DRIVER_LOG_POLLING_ENABLED, true)
+
+      val result = KubernetesExtensions.resolveDriverAppLog(
+        livyConf, () => IndexedSeq("line-1", "line-2"))
+
+      assertResult(IndexedSeq("line-1", "line-2"))(result)
+    }
   }
 
   describe("KubernetesClientExtensions") {