Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions conf/livy.conf.template
Original file line number Diff line number Diff line change
Expand Up @@ -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
Comment thread
gyogal marked this conversation as resolved.

# Weather to create Kubernetes Nginx Ingress for Spark UI. If set to true, configure the desired
# options below
# livy.server.kubernetes.ingress.create = false
Expand Down
7 changes: 7 additions & 0 deletions server/src/main/scala/org/apache/livy/LivyConf.scala
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down Expand Up @@ -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._

Expand Down Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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") {
Expand Down
Loading