Support ephemeral-storage resource

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala

@@ -298,6 +298,13 @@ private[spark] object Config extends Logging {
       .stringConf
       .createOptional
 
+  val KUBERNETES_DRIVER_REQUEST_EPHEMERAL_STORAGE =
+    ConfigBuilder("spark.kubernetes.driver.request.ephemeral.storage")
+      .doc("Specify the ephemeral storage request for the driver pod")
+      .version("4.1.0")
+      .stringConf
+      .createOptional
+
   val KUBERNETES_DRIVER_SUBMIT_CHECK =
     ConfigBuilder("spark.kubernetes.submitInDriver")
     .internal()
@@ -342,6 +349,13 @@ private[spark] object Config extends Logging {
       .stringConf
       .createOptional
 
+  val KUBERNETES_EXECUTOR_REQUEST_EPHEMERAL_STORAGE =
+    ConfigBuilder("spark.kubernetes.executor.request.ephemeral.storage")
+      .doc("Specify the ephemeral storage request for the executor pod")
+      .version("4.1.0")
+      .stringConf
+      .createOptional
+
   val KUBERNETES_DRIVER_POD_NAME =
     ConfigBuilder("spark.kubernetes.driver.pod.name")
       .doc("Name of the driver pod.")

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

@@ -54,6 +54,10 @@ private[spark] class BasicDriverFeatureStep(conf: KubernetesDriverConf)
   // Memory settings
   private val driverMemoryMiB = conf.get(DRIVER_MEMORY)
 
+  // Ephemeral storage
+  private val driverEphemeralStorageRequest = conf
+    .get(KUBERNETES_DRIVER_REQUEST_EPHEMERAL_STORAGE)
+
   // The default memory overhead factor to use, derived from the deprecated
   // `spark.kubernetes.memoryOverheadFactor` config or the default overhead values.
   // If the user has not set it, then use a different default for non-JVM apps. This value is
@@ -90,6 +94,9 @@ private[spark] class BasicDriverFeatureStep(conf: KubernetesDriverConf)
     val maybeCpuLimitQuantity = driverLimitCores.map { limitCores =>
       ("cpu", new Quantity(limitCores))
     }
+    val maybeEphemeralStorageQuantity = driverEphemeralStorageRequest.map { storageRequest =>
+      ("ephemeral-storage", new Quantity(storageRequest))
+    }
 
     val driverResourceQuantities =
       KubernetesUtils.buildResourcesQuantities(SPARK_DRIVER_PREFIX, conf.sparkConf)
@@ -135,6 +142,8 @@ private[spark] class BasicDriverFeatureStep(conf: KubernetesDriverConf)
         .addToLimits(maybeCpuLimitQuantity.toMap.asJava)
         .addToRequests("memory", driverMemoryQuantity)
         .addToLimits("memory", driverMemoryQuantity)
+        .addToRequests(maybeEphemeralStorageQuantity.toMap.asJava)
+        .addToLimits(maybeEphemeralStorageQuantity.toMap.asJava)
         .addToLimits(driverResourceQuantities.asJava)
         .endResources()
       .build()

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

@@ -84,6 +84,7 @@ private[spark] class BasicExecutorFeatureStep(
       execResources.cores.get.toString
     }
   private val executorLimitCores = kubernetesConf.get(KUBERNETES_EXECUTOR_LIMIT_CORES)
+  private val executorRequestEphemeralStorage = kubernetesConf.get(KUBERNETES_EXECUTOR_REQUEST_EPHEMERAL_STORAGE)
 
   private def buildExecutorResourcesQuantities(
       customResources: Set[ExecutorResourceRequest]): Map[String, Quantity] = {
@@ -120,6 +121,10 @@ private[spark] class BasicExecutorFeatureStep(
     val executorCpuQuantity = new Quantity(executorCoresRequest)
     val executorResourceQuantities =
       buildExecutorResourcesQuantities(execResources.customResources.values.toSet)
+    val maybeEphemeralStorageQuantity = executorRequestEphemeralStorage.map {
+      storageRequest =>
+      ("ephemeral-storage", new Quantity(storageRequest))
+    }
 
     val executorEnv: Seq[EnvVar] = {
       val sparkAuthSecret = Option(secMgr.getSecretKey()).map {
@@ -194,6 +199,8 @@ private[spark] class BasicExecutorFeatureStep(
         .addToRequests("memory", executorMemoryQuantity)
         .addToLimits("memory", executorMemoryQuantity)
         .addToRequests("cpu", executorCpuQuantity)
+        .addToRequests(maybeEphemeralStorageQuantity.toMap.asJava)
+        .addToLimits(maybeEphemeralStorageQuantity.toMap.asJava)
         .addToLimits(executorResourceQuantities.asJava)
         .endResources()
       .addNewEnv()

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

@@ -171,6 +171,20 @@ class BasicDriverFeatureStepSuite extends SparkFunSuite {
     }
   }
 
+  test("Check driver pod respects kubernetes driver request ephemeral-storage") {
+    val sparkConf = new SparkConf()
+      .set(KUBERNETES_DRIVER_POD_NAME, "spark-driver-pod")
+      .set(CONTAINER_IMAGE, "spark-driver:latest")
+
+    val basePod = SparkPod.initialPod()
+    sparkConf.set(KUBERNETES_DRIVER_REQUEST_EPHEMERAL_STORAGE, "5Gi")
+    val requests1 = new BasicDriverFeatureStep(KubernetesTestConf.createDriverConf(sparkConf))
+      .configurePod(basePod)
+      .container.getResources
+      .getRequests.asScala
+    assert(amountAndFormat(requests1("ephemeral-storage")) === "5Gi")
+  }
+
   test("Check appropriate entrypoint rerouting for various bindings") {
     val javaSparkConf = new SparkConf()
       .set(DRIVER_MEMORY.key, "4g")

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

@@ -144,6 +144,21 @@ class BasicExecutorFeatureStepSuite extends SparkFunSuite with BeforeAndAfter {
     }
   }
 
+  test("basic executor pod with ephemeral-storage") {
+    baseConf.set(KUBERNETES_EXECUTOR_REQUEST_EPHEMERAL_STORAGE, "5Gi")
+    initDefaultProfile(baseConf)
+    val step = new BasicExecutorFeatureStep(newExecutorConf(), new SecurityManager(baseConf),
+      defaultProfile)
+    val executor = step.configurePod(SparkPod.initialPod())
+
+    assert(executor.container.getResources.getRequests.size() === 3)
+    assert(executor.container.getResources.getLimits.size() === 2)
+    assert(amountAndFormat(executor.container.getResources
+      .getRequests.get("ephemeral-storage")) === "5Gi")
+    assert(amountAndFormat(executor.container.getResources
+      .getLimits.get("ephemeral-storage")) === "5Gi")
+  }
+
   test("basic executor pod has reasonable defaults") {
     val conf = newExecutorConf()
     val step = new BasicExecutorFeatureStep(conf, new SecurityManager(baseConf),