diff --git a/conf/livy.conf.template b/conf/livy.conf.template index c3fe73e86..6187c74d2 100644 --- a/conf/livy.conf.template +++ b/conf/livy.conf.template @@ -294,6 +294,12 @@ # 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. When disabled, +# /sessions/:id/log and /batches/:id/log will not include any Kubernetes driver log lines +# at all. Disable if driver logs are collected externally and those repeated log calls +# are 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 diff --git a/server/src/main/scala/org/apache/livy/LivyConf.scala b/server/src/main/scala/org/apache/livy/LivyConf.scala index 73c7ec3c3..80a4e8fb8 100644 --- a/server/src/main/scala/org/apache/livy/LivyConf.scala +++ b/server/src/main/scala/org/apache/livy/LivyConf.scala @@ -318,6 +318,14 @@ 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 will not include any Kubernetes driver log lines at all. Useful + // when driver logs are collected externally and the repeated pods//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") 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 fec93ca4e..d2beb26be 100644 --- a/server/src/main/scala/org/apache/livy/utils/SparkKubernetesApp.scala +++ b/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//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//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 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..b47017020 100644 --- a/server/src/test/scala/org/apache/livy/utils/SparkKubernetesAppSpec.scala +++ b/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") {