Add ability to scale Ephemeral storage along with memory, similar to CPU

core/invoker/src/main/resources/application.conf

@@ -110,10 +110,13 @@ whisk {
     #}
 
     #if missing, the pod will be created without ephermal disk request/limit
-    #if specified, the pod will be created with ephemeral-storage request+limit set
+    #if specified, the pod will be created with ephemeral-storage request+limit set or using the scale factor
+    #as a multiple of the request memory. If the scaled value exceeds the limit, the limit will be used so, it's good
+    #practice to set the limit if you plan on using the scale-factor.
     #See: https://kubernetes.io/docs/concepts/configuration/manage-resources-containers/#local-ephemeral-storage
     #ephemeral-storage {
     #  limit = 2 g
+    #  scale-factor = 2.0
     #}
 
     #enable PodDisruptionBudget creation for pods? (will include same labels as pods, and specify minAvailable=1 to prevent termination of action pods during maintenance)

core/invoker/src/main/scala/org/apache/openwhisk/core/containerpool/kubernetes/KubernetesClient.scala

@@ -74,7 +74,7 @@ case class KubernetesCpuScalingConfig(millicpus: Int, memory: ByteSize, maxMilli
 /**
  * Configuration for kubernetes ephemeral storage limit for the action container
  */
-case class KubernetesEphemeralStorageConfig(limit: ByteSize)
+case class KubernetesEphemeralStorageConfig(limit: ByteSize, scaleFactor: Double)
 
 /**
  * Exception to indicate a pod took too long to become ready.

core/invoker/src/main/scala/org/apache/openwhisk/core/containerpool/kubernetes/WhiskPodBuilder.scala

@@ -108,7 +108,14 @@ class WhiskPodBuilder(client: NamespacedKubernetesClient, config: KubernetesClie
       .getOrElse(Map.empty)
 
     val diskLimit = config.ephemeralStorage
-      .map(diskConfig => Map("ephemeral-storage" -> new Quantity(diskConfig.limit.toMB + "Mi")))
+      .map(
+        diskConfig =>
+          // Scale the ephemeral storage unless it exceeds the limit, if it exceeds the limit use the limit.
+          if ((diskConfig.scaleFactor > 0) && (diskConfig.scaleFactor * memory.toMB < diskConfig.limit.toMB)) {
+            Map("ephemeral-storage" -> new Quantity(diskConfig.scaleFactor * memory.toMB + "Mi"))
+          } else {
+            Map("ephemeral-storage" -> new Quantity(diskConfig.limit.toMB + "Mi"))
+        })
       .getOrElse(Map.empty)
 
     //In container its assumed that env, port, resource limits are set explicitly

tests/src/test/scala/org/apache/openwhisk/core/containerpool/kubernetes/test/WhiskPodBuilderTests.scala

@@ -133,7 +133,47 @@ class WhiskPodBuilderTests extends FlatSpec with Matchers with KubeClientSupport
       Some(KubernetesCpuScalingConfig(300, 3.MB, 1000)),
       false,
       None,
-      Some(KubernetesEphemeralStorageConfig(1.GB)))
+      Some(KubernetesEphemeralStorageConfig(1.GB, 0.0)))
+    val builder = new WhiskPodBuilder(kubeClient, config)
+
+    val (pod, _) = builder.buildPodSpec(name, testImage, 2.MB, Map("foo" -> "bar"), Map("fooL" -> "barV"), config)
+    withClue(Serialization.asYaml(pod)) {
+      val c = getActionContainer(pod)
+      c.getResources.getLimits.asScala.get("ephemeral-storage").map(_.getAmount) shouldBe Some("1024Mi")
+      c.getResources.getRequests.asScala.get("ephemeral-storage").map(_.getAmount) shouldBe Some("1024Mi")
+    }
+  }
+  it should "scale ephemeral storage when scale factor is given" in {
+    val config = KubernetesClientConfig(
+      KubernetesClientTimeoutConfig(1.second, 1.second),
+      KubernetesInvokerNodeAffinity(false, "k", "v"),
+      true,
+      None,
+      None,
+      Some(KubernetesCpuScalingConfig(300, 3.MB, 1000)),
+      false,
+      None,
+      Some(KubernetesEphemeralStorageConfig(1.GB, 1.25)))
+    val builder = new WhiskPodBuilder(kubeClient, config)
+
+    val (pod, _) = builder.buildPodSpec(name, testImage, 2.MB, Map("foo" -> "bar"), Map("fooL" -> "barV"), config)
+    withClue(Serialization.asYaml(pod)) {
+      val c = getActionContainer(pod)
+      c.getResources.getLimits.asScala.get("ephemeral-storage").map(_.getAmount) shouldBe Some("2.5Mi")
+      c.getResources.getRequests.asScala.get("ephemeral-storage").map(_.getAmount) shouldBe Some("2.5Mi")
+    }
+  }
+  it should "use ephemeral storage limit when scale factor suggests larger size" in {
+    val config = KubernetesClientConfig(
+      KubernetesClientTimeoutConfig(1.second, 1.second),
+      KubernetesInvokerNodeAffinity(false, "k", "v"),
+      true,
+      None,
+      None,
+      Some(KubernetesCpuScalingConfig(300, 3.MB, 1000)),
+      false,
+      None,
+      Some(KubernetesEphemeralStorageConfig(1.GB, 1000)))
     val builder = new WhiskPodBuilder(kubeClient, config)
 
     val (pod, _) = builder.buildPodSpec(name, testImage, 2.MB, Map("foo" -> "bar"), Map("fooL" -> "barV"), config)