[KYUUBI #833] Check if spark.kubernetes.executor.podNamePrefix is invalid

### _Why are the changes needed?_
close #833

### _How was this patch tested?_
- [ ] Add some test cases that check the changes thoroughly including negative and positive cases if possible

- [ ] Add screenshots for manual tests if appropriate

- [x] [Run test](https://kyuubi.apache.org/docs/latest/develop_tools/testing.html#running-tests) locally before make a pull request

Closes #3164 from Kwafoor/master.

Closes #833

63995b5c [wangjunbo] [kyuubi-833]Check if spark.kubernetes.executor.podNamePrefix is invalid
ef6663dd [wangjunbo] [kyuubi-833]Check if spark.kubernetes.executor.podNamePrefix is invalid

Authored-by: wangjunbo <wangjunbo@qiyi.com>
Signed-off-by: ulysses-you <ulyssesyou@apache.org>

integration-tests/kyuubi-kubernetes-it/src/test/scala/org/apache/kyuubi/kubernetes/test/spark/SparkOnKubernetesTestsSuite.scala

@@ -23,14 +23,16 @@ import scala.concurrent.duration._
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.net.NetUtils
 
-import org.apache.kyuubi.{BatchTestHelper, Logging, Utils, WithKyuubiServer, WithSimpleDFSService}
+import org.apache.kyuubi.{BatchTestHelper, KyuubiException, Logging, Utils, WithKyuubiServer, WithSimpleDFSService}
 import org.apache.kyuubi.config.KyuubiConf
 import org.apache.kyuubi.config.KyuubiConf.{FRONTEND_CONNECTION_URL_USE_HOSTNAME, FRONTEND_THRIFT_BINARY_BIND_HOST}
 import org.apache.kyuubi.engine.{ApplicationInfo, ApplicationOperation, KubernetesApplicationOperation}
 import org.apache.kyuubi.engine.ApplicationState.{FAILED, NOT_FOUND, RUNNING}
+import org.apache.kyuubi.engine.spark.SparkProcessBuilder
 import org.apache.kyuubi.kubernetes.test.MiniKube
 import org.apache.kyuubi.operation.SparkQueryTests
 import org.apache.kyuubi.session.{KyuubiBatchSessionImpl, KyuubiSessionManager}
+import org.apache.kyuubi.util.Validator.KUBERNETES_EXECUTOR_POD_NAME_PREFIX
 import org.apache.kyuubi.zookeeper.ZookeeperConf.ZK_CLIENT_PORT_ADDRESS
 
 abstract class SparkOnKubernetesSuiteBase
@@ -173,6 +175,17 @@ class KyuubiOperationKubernetesClusterClusterModeSuite
   private def sessionManager: KyuubiSessionManager =
     server.backendService.sessionManager.asInstanceOf[KyuubiSessionManager]
 
+  test("Check if spark.kubernetes.executor.podNamePrefix is invalid") {
+    Seq("_123", "spark_exec", "spark@", "a" * 238).foreach { invalid =>
+      conf.set(KUBERNETES_EXECUTOR_POD_NAME_PREFIX, invalid)
+      val builder = new SparkProcessBuilder("test", conf)
+      val e = intercept[KyuubiException](builder.validateConf)
+      assert(e.getMessage === s"'$invalid' in spark.kubernetes.executor.podNamePrefix is" +
+        s" invalid. must conform https://kubernetes.io/docs/concepts/overview/" +
+        "working-with-objects/names/#dns-subdomain-names and the value length <= 237")
+    }
+  }
+
   test("Spark Cluster Mode On Kubernetes Kyuubi KubernetesApplicationOperation Suite") {
     val driverPodNamePrefix = "kyuubi-spark-driver"
     conf.set(

kyuubi-server/src/main/scala/org/apache/kyuubi/engine/EngineRef.scala

@@ -187,6 +187,7 @@ private[kyuubi] class EngineRef(
     try {
       val redactedCmd = builder.toString
       info(s"Launching engine:\n$redactedCmd")
+      builder.validateConf
       val process = builder.start
       var exitValue: Option[Int] = None
       while (engineRef.isEmpty) {

kyuubi-server/src/main/scala/org/apache/kyuubi/engine/ProcBuilder.scala

@@ -192,6 +192,8 @@ trait ProcBuilder {
     file
   }
 
+  def validateConf: Unit = {}
+
   final def start: Process = synchronized {
     process = processBuilder.start()
     processLaunched = true

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

@@ -31,6 +31,7 @@ import org.apache.kyuubi.engine.{KyuubiApplicationManager, ProcBuilder}
 import org.apache.kyuubi.ha.HighAvailabilityConf
 import org.apache.kyuubi.ha.client.AuthTypes
 import org.apache.kyuubi.operation.log.OperationLog
+import org.apache.kyuubi.util.Validator
 
 class SparkProcessBuilder(
     override val proxyUser: String,
@@ -164,6 +165,9 @@ class SparkProcessBuilder(
   override def clusterManager(): Option[String] = {
     conf.getOption(MASTER_KEY).orElse(defaultMaster)
   }
+
+  override def validateConf: Unit = Validator.validateConf(conf)
+
 }
 
 object SparkProcessBuilder {

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

@@ -0,0 +1,53 @@
+/*
+ * 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.KyuubiException
+import org.apache.kyuubi.config.KyuubiConf
+
+object Validator {
+
+  def validateConf(kyuubiConf: KyuubiConf): Unit = {
+    validateSparkK8sConf(kyuubiConf)
+  }
+
+  private def validateSparkK8sConf(kyuubiConf: KyuubiConf): Unit = {
+    val prefix = kyuubiConf.getOption(KUBERNETES_EXECUTOR_POD_NAME_PREFIX).orNull
+    if (prefix != null && !isValidExecutorPodNamePrefix(prefix)) {
+      throw new KyuubiException(s"'$prefix' " +
+        s"in spark.kubernetes.executor.podNamePrefix is invalid." +
+        s" must conform https://kubernetes.io/docs/concepts/overview/working-with-objects" +
+        "/names/#dns-subdomain-names and the value length <= 237")
+    }
+  }
+
+  private val dns1123LabelFmt = "[a-z0-9]([-a-z0-9]*[a-z0-9])?"
+
+  private val podConfValidator = (s"^$dns1123LabelFmt(\\.$dns1123LabelFmt)*$$").r.pattern
+
+  val KUBERNETES_DNS_SUBDOMAIN_NAME_MAX_LENGTH = 253
+
+  private def isValidExecutorPodNamePrefix(prefix: String): Boolean = {
+    // 6 is length of '-exec-'
+    val reservedLen = Int.MaxValue.toString.length + 6
+    val validLength = prefix.length + reservedLen <= KUBERNETES_DNS_SUBDOMAIN_NAME_MAX_LENGTH
+    validLength && podConfValidator.matcher(prefix).matches()
+  }
+
+  val KUBERNETES_EXECUTOR_POD_NAME_PREFIX = "spark.kubernetes.executor.podNamePrefix"
+}