[SPARK-54553][VOLCANO] Supports receiving podgroup JSON format configurations when using Volcano.

Author: ruanwenjun

docs/running-on-kubernetes.md

@@ -1954,9 +1954,7 @@ Note that currently only driver/job level PodGroup is supported in Volcano Featu
 Volcano defines PodGroup spec using [CRD yaml](https://volcano.sh/en/docs/podgroup/#example).
 
 Similar to [Pod template](#pod-template), Spark users can use Volcano PodGroup Template to define the PodGroup spec configurations.
-To do so, specify the Spark property `spark.kubernetes.scheduler.volcano.podGroupTemplateFile` to point to files accessible to the `spark-submit` process.
 Below is an example of PodGroup template:
-
 ```yaml
 apiVersion: scheduling.volcano.sh/v1beta1
 kind: PodGroup
@@ -1975,6 +1973,17 @@ spec:
   queue: default
 ```
 
+You have two options to provide the PodGroup template in spark.
+1. Use `spark.kubernetes.scheduler.volcano.podGroupTemplateFile` to point to files accessible to the `spark-submit` process
+```bash
+--conf spark.kubernetes.scheduler.volcano.podGroupTemplateFile=/path/to/podgroup
+```
+
+2. Use `spark.kubernetes.scheduler.volcano.podGroupTemplateJson` to provide the template in json format.
+```bash
+--conf spark.kubernetes.scheduler.volcano.podGroupTemplateJson={"spec": {"minMember": 1,"minResources": {"cpu": "2","memory": "3Gi"},"priorityClassName": "system-node-critical","queue": "default"}}
+```
+
 #### Using Apache YuniKorn as Customized Scheduler for Spark on Kubernetes
 
 [Apache YuniKorn](https://yunikorn.apache.org/) is a resource scheduler for Kubernetes that provides advanced batch scheduling

resource-managers/kubernetes/core/volcano/src/main/scala/org/apache/spark/deploy/k8s/features/VolcanoFeatureStep.scala

@@ -16,15 +16,20 @@
  */
 package org.apache.spark.deploy.k8s.features
 
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.dataformat.yaml.YAMLFactory
 import io.fabric8.kubernetes.api.model._
 import io.fabric8.volcano.api.model.scheduling.v1beta1.{PodGroup, PodGroupSpec}
 import io.fabric8.volcano.client.DefaultVolcanoClient
 
+import org.apache.spark.SparkException
 import org.apache.spark.deploy.k8s.{KubernetesConf, KubernetesDriverConf, KubernetesExecutorConf, SparkPod}
 import org.apache.spark.internal.Logging
 
-private[spark] class VolcanoFeatureStep extends KubernetesDriverCustomFeatureConfigStep
-  with KubernetesExecutorCustomFeatureConfigStep with Logging {
+private[spark] class VolcanoFeatureStep
+    extends KubernetesDriverCustomFeatureConfigStep
+    with KubernetesExecutorCustomFeatureConfigStep
+    with Logging {
   import VolcanoFeatureStep._
 
   private var kubernetesConf: KubernetesConf = _
@@ -42,13 +47,15 @@ private[spark] class VolcanoFeatureStep extends KubernetesDriverCustomFeatureCon
 
   override def getAdditionalPreKubernetesResources(): Seq[HasMetadata] = {
     if (kubernetesConf.isInstanceOf[KubernetesExecutorConf]) {
-      logWarning("VolcanoFeatureStep#getAdditionalPreKubernetesResources() is not supported " +
-        "for executor.")
+      logWarning(
+        "VolcanoFeatureStep#getAdditionalPreKubernetesResources() is not supported " +
+          "for executor.")
       return Seq.empty
     }
     lazy val client = new DefaultVolcanoClient
-    val template = kubernetesConf.getOption(POD_GROUP_TEMPLATE_FILE_KEY)
-    val pg = template.map(client.podGroups.load(_).item).getOrElse(new PodGroup())
+    val pg = getPodGroupConfigByTemplateFile(client)
+      .orElse(getPodGroupByTemplateJson(client))
+      .getOrElse(new PodGroup())
     var metadata = pg.getMetadata
     if (metadata == null) metadata = new ObjectMeta
     metadata.setName(podGroupName)
@@ -59,13 +66,42 @@ private[spark] class VolcanoFeatureStep extends KubernetesDriverCustomFeatureCon
     if (spec == null) spec = new PodGroupSpec
     pg.setSpec(spec)
 
+    logDebug(s"Volcano PodGroup configuration: $pg")
     Seq(pg)
   }
 
+  private def getPodGroupConfigByTemplateFile(
+      volcanoClient: DefaultVolcanoClient): Option[PodGroup] = {
+    kubernetesConf.getOption(POD_GROUP_TEMPLATE_FILE_KEY).map { templateFile =>
+      logDebug("Loading Volcano PodGroup configuration from template file")
+      volcanoClient.podGroups.load(templateFile).item
+    }
+  }
+
+  private def getPodGroupByTemplateJson(volcanoClient: DefaultVolcanoClient): Option[PodGroup] = {
+    kubernetesConf.getOption(POD_GROUP_TEMPLATE_JSON_KEY).map { templateJson =>
+      logDebug("Loading Volcano PodGroup configuration from template json")
+      try {
+        val templateYaml = new ObjectMapper()
+          .writerWithDefaultPrettyPrinter()
+          .writeValueAsString(new ObjectMapper(new YAMLFactory()).readTree(templateJson))
+        volcanoClient
+          .podGroups()
+          .load(new java.io.ByteArrayInputStream(templateYaml.getBytes()))
+          .item()
+      } catch {
+        case e: Exception =>
+          throw new SparkException(
+            f"The ${POD_GROUP_TEMPLATE_JSON_KEY} provided is not validated",
+            e)
+      }
+    }
+  }
+
   override def configurePod(pod: SparkPod): SparkPod = {
     val k8sPodBuilder = new PodBuilder(pod.pod)
       .editMetadata()
-        .addToAnnotations(POD_GROUP_ANNOTATION, podGroupName)
+      .addToAnnotations(POD_GROUP_ANNOTATION, podGroupName)
       .endMetadata()
     val k8sPod = k8sPodBuilder.build()
     SparkPod(k8sPod, pod.container)
@@ -75,4 +111,5 @@ private[spark] class VolcanoFeatureStep extends KubernetesDriverCustomFeatureCon
 private[spark] object VolcanoFeatureStep {
   val POD_GROUP_ANNOTATION = "scheduling.k8s.io/group-name"
   val POD_GROUP_TEMPLATE_FILE_KEY = "spark.kubernetes.scheduler.volcano.podGroupTemplateFile"
+  val POD_GROUP_TEMPLATE_JSON_KEY = "spark.kubernetes.scheduler.volcano.podGroupTemplateJson"
 }

resource-managers/kubernetes/core/volcano/src/test/scala/org/apache/spark/deploy/k8s/features/VolcanoFeatureStepSuite.scala

@@ -34,7 +34,8 @@ class VolcanoFeatureStepSuite extends SparkFunSuite {
 
     val annotations = configuredPod.pod.getMetadata.getAnnotations
 
-    assert(annotations.get("scheduling.k8s.io/group-name") === s"${kubernetesConf.appId}-podgroup")
+    assert(
+      annotations.get("scheduling.k8s.io/group-name") === s"${kubernetesConf.appId}-podgroup")
     val podGroup = step.getAdditionalPreKubernetesResources().head.asInstanceOf[PodGroup]
     assert(podGroup.getMetadata.getName === s"${kubernetesConf.appId}-podgroup")
   }
@@ -46,12 +47,13 @@ class VolcanoFeatureStepSuite extends SparkFunSuite {
     step.init(kubernetesConf)
     val configuredPod = step.configurePod(SparkPod.initialPod())
     val annotations = configuredPod.pod.getMetadata.getAnnotations
-    assert(annotations.get("scheduling.k8s.io/group-name") === s"${kubernetesConf.appId}-podgroup")
+    assert(
+      annotations.get("scheduling.k8s.io/group-name") === s"${kubernetesConf.appId}-podgroup")
   }
 
   test("SPARK-38455: Support driver podgroup template") {
-    val templatePath = new File(
-      getClass.getResource("/driver-podgroup-template.yml").getFile).getAbsolutePath
+    val templatePath =
+      new File(getClass.getResource("/driver-podgroup-template.yml").getFile).getAbsolutePath
     val sparkConf = new SparkConf()
       .set(VolcanoFeatureStep.POD_GROUP_TEMPLATE_FILE_KEY, templatePath)
     val kubernetesConf = KubernetesTestConf.createDriverConf(sparkConf)
@@ -67,6 +69,34 @@ class VolcanoFeatureStepSuite extends SparkFunSuite {
     assert(podGroup.getSpec.getQueue == "driver-queue")
   }
 
+  test("SPARK-38455: Support create driver podgroup by attributes") {
+    val sparkConf = new SparkConf()
+      .set(
+        VolcanoFeatureStep.POD_GROUP_TEMPLATE_JSON_KEY,
+        "{\n" +
+          "  \"spec\": {\n" +
+          "    \"minMember\": 1,\n" +
+          "    \"minResources\": {\n" +
+          "      \"cpu\": \"2\",\n" +
+          "      \"memory\": \"2048Mi\"\n" +
+          "    },\n" +
+          "    \"priorityClassName\": \"driver-priority\",\n" +
+          "    \"queue\": \"driver-queue\"\n" +
+          "  }\n" +
+          "}")
+    val kubernetesConf = KubernetesTestConf.createDriverConf(sparkConf)
+    val step = new VolcanoFeatureStep()
+    step.init(kubernetesConf)
+    step.configurePod(SparkPod.initialPod())
+    val podGroup = step.getAdditionalPreKubernetesResources().head.asInstanceOf[PodGroup]
+    assert(podGroup.getSpec.getMinMember == 1)
+    assert(podGroup.getSpec.getMinResources.get("cpu").getAmount == "2")
+    assert(podGroup.getSpec.getMinResources.get("memory").getAmount == "2048")
+    assert(podGroup.getSpec.getMinResources.get("memory").getFormat == "Mi")
+    assert(podGroup.getSpec.getPriorityClassName == "driver-priority")
+    assert(podGroup.getSpec.getQueue == "driver-queue")
+  }
+
   test("SPARK-38503: return empty for executor pre resource") {
     val kubernetesConf = KubernetesTestConf.createExecutorConf(new SparkConf())
     val step = new VolcanoFeatureStep()