[KYUUBI #7495][KUBERNETES] Avoid overlong pod log directory names

externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/SparkSQLEngine.scala

@@ -230,9 +230,14 @@ object SparkSQLEngine extends Logging {
 
   private val sparkSessionCreated = new AtomicBoolean(false)
 
-  // Kubernetes pod name max length - '-exec-' - Int.MAX_VALUE.length
-  // 253 - 10 - 6
-  val EXECUTOR_POD_NAME_PREFIX_MAX_LENGTH = 237
+  final private val POD_NAME_MAX_LENGTH = 253
+  final private val POD_UID_MAX_LENGTH = 36
+  final private val POD_LOGS_DIRECTORY_SEPARATOR_LENGTH = 2
+  final private val EXECUTOR_POD_NAME_RESERVED_LENGTH =
+    "-exec-".length + Int.MaxValue.toString.length
+
+  val EXECUTOR_POD_NAME_PREFIX_MAX_LENGTH =
+    executorPodNamePrefixMaxLength(KUBERNETES_NAMESPACE.defaultValStr)
 
   SignalRegister.registerLogger(logger)
   setupConf()
@@ -292,7 +297,9 @@ object SparkSQLEngine extends Logging {
       // due to the long app name
       _sparkConf.setIfMissing(
         "spark.kubernetes.executor.podNamePrefix",
-        generateExecutorPodNamePrefixForK8s(user))
+        generateExecutorPodNamePrefixForK8s(
+          user,
+          _sparkConf.get("spark.kubernetes.namespace", KUBERNETES_NAMESPACE.defaultValStr)))
 
       if (!isOnK8sClusterMode) {
         // set driver host to ip instead of kyuubi pod name
@@ -464,17 +471,27 @@ object SparkSQLEngine extends Logging {
   }
 
   @VisibleForTesting
-  def generateExecutorPodNamePrefixForK8s(userName: String): String = {
+  def generateExecutorPodNamePrefixForK8s(
+      userName: String,
+      namespace: String = KUBERNETES_NAMESPACE.defaultValStr): String = {
     val resolvedUserName =
       userName.trim.toLowerCase(Locale.ROOT)
         .replaceAll("[^a-z0-9\\-]", "-")
         .replaceAll("-+", "-")
         .replaceAll("^-", "")
     val podNamePrefixWithUser = s"kyuubi-$resolvedUserName-${Instant.now().toEpochMilli}"
-    if (podNamePrefixWithUser.length <= EXECUTOR_POD_NAME_PREFIX_MAX_LENGTH) {
+    if (podNamePrefixWithUser.length <= executorPodNamePrefixMaxLength(namespace)) {
       podNamePrefixWithUser
     } else {
       s"kyuubi-${UUID.randomUUID()}"
     }
   }
+
+  private def executorPodNamePrefixMaxLength(namespace: String): Int = {
+    POD_NAME_MAX_LENGTH -
+      namespace.length -
+      POD_UID_MAX_LENGTH -
+      POD_LOGS_DIRECTORY_SEPARATOR_LENGTH -
+      EXECUTOR_POD_NAME_RESERVED_LENGTH
+  }
 }

externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/SparkSQLEngineSuite.scala

@@ -21,6 +21,9 @@ import org.apache.kyuubi.KyuubiFunSuite
 
 class SparkSQLEngineSuite extends KyuubiFunSuite {
 
+  private val namespace = "n" * 63
+  private val podUid = "u" * 36
+
   test("[KYUUBI #3385] generate executor pod name prefix with user or UUID") {
     val userName1 = "/kyuubi_user+-*"
     val executorPodNamePrefix1 = SparkSQLEngine.generateExecutorPodNamePrefixForK8s(userName1)
@@ -33,5 +36,16 @@ class SparkSQLEngineSuite extends KyuubiFunSuite {
     val executorPodNamePrefix2 = SparkSQLEngine.generateExecutorPodNamePrefixForK8s(userName2)
     assert(!executorPodNamePrefix2.contains(userName2))
     assert(executorPodNamePrefix2.length <= SparkSQLEngine.EXECUTOR_POD_NAME_PREFIX_MAX_LENGTH)
+
+    val userName3 = "l" * 160
+    val executorPodNamePrefix3 =
+      SparkSQLEngine.generateExecutorPodNamePrefixForK8s(userName3, namespace)
+    val executorPodName3 = s"$executorPodNamePrefix3-exec-${Int.MaxValue}"
+    assert(!executorPodNamePrefix3.contains(userName3))
+    assert(podLogsDirectoryNameLength(namespace, executorPodName3) <= 253)
+  }
+
+  private def podLogsDirectoryNameLength(namespace: String, podName: String): Int = {
+    s"${namespace}_${podName}_$podUid".length
   }
 }

kyuubi-server/src/main/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilder.scala

@@ -261,12 +261,16 @@ class SparkProcessBuilder(
 
   def appendPodNameConf(conf: Map[String, String]): Map[String, String] = {
     val appName = conf.getOrElse(APP_KEY, "spark")
+    val namespace = conf.get(KUBERNETES_NAMESPACE_KEY)
+      .orElse(kubernetesNamespace())
+      .getOrElse(KUBERNETES_NAMESPACE.defaultValStr)
     val map = mutable.Map.newBuilder[String, String]
     if (clusterManager().exists(cm => cm.toLowerCase(Locale.ROOT).startsWith("k8s"))) {
       if (!conf.contains(KUBERNETES_EXECUTOR_POD_NAME_PREFIX)) {
         val prefix = KubernetesUtils.generateExecutorPodNamePrefix(
           appName,
           engineRefId,
+          namespace,
           forciblyRewriteExecPodNamePrefix)
         map += (KUBERNETES_EXECUTOR_POD_NAME_PREFIX -> prefix)
       }
@@ -275,6 +279,7 @@ class SparkProcessBuilder(
           val name = KubernetesUtils.generateDriverPodName(
             appName,
             engineRefId,
+            namespace,
             forciblyRewriteDriverPodName)
           map += (KUBERNETES_DRIVER_POD_NAME -> name)
         }

kyuubi-server/src/main/scala/org/apache/kyuubi/util/KubernetesUtils.scala

@@ -33,10 +33,14 @@ import org.apache.kyuubi.config.KyuubiConf
 import org.apache.kyuubi.config.KyuubiConf._
 
 object KubernetesUtils extends Logging {
-  // Kubernetes pod name max length - '-exec-' - Int.MAX_VALUE.length
-  // 253 - 10 - 6
-  final val EXECUTOR_POD_NAME_PREFIX_MAX_LENGTH = 237
   final val DRIVER_POD_NAME_MAX_LENGTH = 253
+  final private val POD_UID_MAX_LENGTH = 36
+  final private val POD_LOGS_DIRECTORY_SEPARATOR_LENGTH = 2
+  final private val EXECUTOR_POD_NAME_RESERVED_LENGTH =
+    "-exec-".length + Int.MaxValue.toString.length
+
+  final val EXECUTOR_POD_NAME_PREFIX_MAX_LENGTH =
+    maxExecutorPodNamePrefixLength(KUBERNETES_NAMESPACE.defaultValStr)
 
   def buildKubernetesClient(conf: KyuubiConf): Option[KubernetesClient] = {
     val master = conf.get(KUBERNETES_MASTER)
@@ -138,6 +142,14 @@ object KubernetesUtils extends Logging {
       appName: String,
       engineRefId: String,
       forciblyRewrite: Boolean): String = {
+    generateDriverPodName(appName, engineRefId, KUBERNETES_NAMESPACE.defaultValStr, forciblyRewrite)
+  }
+
+  def generateDriverPodName(
+      appName: String,
+      engineRefId: String,
+      namespace: String,
+      forciblyRewrite: Boolean): String = {
     val resourceNamePrefix = if (appName.contains(engineRefId)) {
       getResourceNamePrefix(appName, None)
     } else {
@@ -148,7 +160,7 @@ object KubernetesUtils extends Logging {
     } else {
       s"kyuubi-$resourceNamePrefix-driver"
     }
-    if (forciblyRewrite || resolvedResourceName.length > DRIVER_POD_NAME_MAX_LENGTH) {
+    if (forciblyRewrite || resolvedResourceName.length > maxDriverPodNameLength(namespace)) {
       s"kyuubi-$engineRefId-driver"
     } else {
       resolvedResourceName
@@ -159,6 +171,18 @@ object KubernetesUtils extends Logging {
       appName: String,
       engineRefId: String,
       forciblyRewrite: Boolean): String = {
+    generateExecutorPodNamePrefix(
+      appName,
+      engineRefId,
+      KUBERNETES_NAMESPACE.defaultValStr,
+      forciblyRewrite)
+  }
+
+  def generateExecutorPodNamePrefix(
+      appName: String,
+      engineRefId: String,
+      namespace: String,
+      forciblyRewrite: Boolean): String = {
     val resourceNamePrefix = if (appName.contains(engineRefId)) {
       getResourceNamePrefix(appName, None)
     } else {
@@ -169,10 +193,22 @@ object KubernetesUtils extends Logging {
     } else {
       s"kyuubi-$resourceNamePrefix"
     }
-    if (forciblyRewrite || resolvedResourceName.length > EXECUTOR_POD_NAME_PREFIX_MAX_LENGTH) {
+    if (forciblyRewrite || resolvedResourceName.length > maxExecutorPodNamePrefixLength(
+        namespace)) {
       s"kyuubi-$engineRefId"
     } else {
       resolvedResourceName
     }
   }
+
+  private def maxDriverPodNameLength(namespace: String): Int = {
+    DRIVER_POD_NAME_MAX_LENGTH -
+      namespace.length -
+      POD_UID_MAX_LENGTH -
+      POD_LOGS_DIRECTORY_SEPARATOR_LENGTH
+  }
+
+  private def maxExecutorPodNamePrefixLength(namespace: String): Int = {
+    maxDriverPodNameLength(namespace) - EXECUTOR_POD_NAME_RESERVED_LENGTH
+  }
 }

kyuubi-server/src/test/scala/org/apache/kyuubi/util/KubernetesPodNameSuite.scala

@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.util
+
+import org.apache.kyuubi.KyuubiFunSuite
+import org.apache.kyuubi.config.KyuubiConf
+import org.apache.kyuubi.engine.spark.SparkProcessBuilder
+import org.apache.kyuubi.engine.spark.SparkProcessBuilder._
+
+class KubernetesPodNameSuite extends KyuubiFunSuite {
+
+  private val namespace = "n" * 63
+  private val engineRefId = "kyuubi-test-engine"
+  private val longAppName = "a" * 160
+  private val podUid = "u" * 36
+
+  test("driver pod name should reserve kubelet pod logs directory budget") {
+    val podName =
+      KubernetesUtils.generateDriverPodName(longAppName, engineRefId, namespace, false)
+
+    assert(podName === s"kyuubi-$engineRefId-driver")
+    assert(podLogsDirectoryNameLength(namespace, podName) <=
+      KubernetesUtils.DRIVER_POD_NAME_MAX_LENGTH)
+  }
+
+  test("executor pod name prefix should reserve kubelet pod logs directory budget") {
+    val prefix =
+      KubernetesUtils.generateExecutorPodNamePrefix(longAppName, engineRefId, namespace, false)
+    val podName = s"$prefix-exec-${Int.MaxValue}"
+
+    assert(prefix === s"kyuubi-$engineRefId")
+    assert(podLogsDirectoryNameLength(namespace, podName) <=
+      KubernetesUtils.DRIVER_POD_NAME_MAX_LENGTH)
+  }
+
+  test("SparkProcessBuilder should use spark kubernetes namespace for pod name budget") {
+    val builder = new SparkProcessBuilder(
+      "kyuubi",
+      true,
+      KyuubiConf().set(MASTER_KEY, "k8s://internal").set(DEPLOY_MODE_KEY, "cluster"),
+      engineRefId)
+    val conf = Map(APP_KEY -> longAppName, KUBERNETES_NAMESPACE_KEY -> namespace)
+    val podNameConf = builder.appendPodNameConf(conf)
+
+    assert(podNameConf(KUBERNETES_DRIVER_POD_NAME) === s"kyuubi-$engineRefId-driver")
+    assert(podNameConf(KUBERNETES_EXECUTOR_POD_NAME_PREFIX) === s"kyuubi-$engineRefId")
+  }
+
+  private def podLogsDirectoryNameLength(namespace: String, podName: String): Int = {
+    s"${namespace}_${podName}_$podUid".length
+  }
+}