Fix conversion from GB to MiB (#470)

* Fix conversion from GB to MiB

Previously we applied a value which was in units MiB to Kubernetes _as if_ it
were in MB units.  Now, place MiB units in variable names and apply them to
Kubernetes as the correct MiB units.

* Fix test

* Update tests

Author: ash211

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

@@ -101,5 +101,5 @@ package object constants {
   private[spark] val DRIVER_CONTAINER_NAME = "spark-kubernetes-driver"
   private[spark] val KUBERNETES_MASTER_INTERNAL_URL = "https://kubernetes.default.svc"
   private[spark] val MEMORY_OVERHEAD_FACTOR = 0.10
-  private[spark] val MEMORY_OVERHEAD_MIN = 384L
+  private[spark] val MEMORY_OVERHEAD_MIN_MIB = 384L
 }

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/submit/submitsteps/BaseDriverConfigurationStep.scala

@@ -46,13 +46,13 @@ private[spark] class BaseDriverConfigurationStep(
   private val driverLimitCores = submissionSparkConf.get(KUBERNETES_DRIVER_LIMIT_CORES)
 
   // Memory settings
-  private val driverMemoryMb = submissionSparkConf.get(
+  private val driverMemoryMiB = submissionSparkConf.get(
       org.apache.spark.internal.config.DRIVER_MEMORY)
-  private val memoryOverheadMb = submissionSparkConf
+  private val memoryOverheadMiB = submissionSparkConf
       .get(KUBERNETES_DRIVER_MEMORY_OVERHEAD)
-      .getOrElse(math.max((MEMORY_OVERHEAD_FACTOR * driverMemoryMb).toInt,
-          MEMORY_OVERHEAD_MIN))
-  private val driverContainerMemoryWithOverhead = driverMemoryMb + memoryOverheadMb
+      .getOrElse(math.max((MEMORY_OVERHEAD_FACTOR * driverMemoryMiB).toInt,
+          MEMORY_OVERHEAD_MIN_MIB))
+  private val driverContainerMemoryWithOverheadMiB = driverMemoryMiB + memoryOverheadMiB
   private val driverDockerImage = submissionSparkConf.get(DRIVER_DOCKER_IMAGE)
 
   override def configureDriver(
@@ -86,10 +86,10 @@ private[spark] class BaseDriverConfigurationStep(
       .withAmount(driverCpuCores)
       .build()
     val driverMemoryQuantity = new QuantityBuilder(false)
-      .withAmount(s"${driverMemoryMb}M")
+      .withAmount(s"${driverMemoryMiB}Mi")
       .build()
     val driverMemoryLimitQuantity = new QuantityBuilder(false)
-      .withAmount(s"${driverContainerMemoryWithOverhead}M")
+      .withAmount(s"${driverContainerMemoryWithOverheadMiB}Mi")
       .build()
     val maybeCpuLimitQuantity = driverLimitCores.map { limitCores =>
       ("cpu", new QuantityBuilder(false).withAmount(limitCores).build())
@@ -102,7 +102,7 @@ private[spark] class BaseDriverConfigurationStep(
       .addToEnv(driverExtraClasspathEnv.toSeq: _*)
       .addNewEnv()
         .withName(ENV_DRIVER_MEMORY)
-        .withValue(driverContainerMemoryWithOverhead + "m")
+        .withValue(driverContainerMemoryWithOverheadMiB + "M") // JVM treats the "M" unit as "Mi"
         .endEnv()
       .addNewEnv()
         .withName(ENV_DRIVER_MAIN_CLASS)

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesClusterSchedulerBackend.scala

@@ -113,16 +113,16 @@ private[spark] class KubernetesClusterSchedulerBackend(
       throw new SparkException("Must specify the driver pod name"))
   private val executorPodNamePrefix = conf.get(KUBERNETES_EXECUTOR_POD_NAME_PREFIX)
 
-  private val executorMemoryMb = conf.get(org.apache.spark.internal.config.EXECUTOR_MEMORY)
+  private val executorMemoryMiB = conf.get(org.apache.spark.internal.config.EXECUTOR_MEMORY)
   private val executorMemoryString = conf.get(
     org.apache.spark.internal.config.EXECUTOR_MEMORY.key,
     org.apache.spark.internal.config.EXECUTOR_MEMORY.defaultValueString)
 
-  private val memoryOverheadMb = conf
+  private val memoryOverheadMiB = conf
     .get(KUBERNETES_EXECUTOR_MEMORY_OVERHEAD)
-    .getOrElse(math.max((MEMORY_OVERHEAD_FACTOR * executorMemoryMb).toInt,
-      MEMORY_OVERHEAD_MIN))
-  private val executorMemoryWithOverhead = executorMemoryMb + memoryOverheadMb
+    .getOrElse(math.max((MEMORY_OVERHEAD_FACTOR * executorMemoryMiB).toInt,
+      MEMORY_OVERHEAD_MIN_MIB))
+  private val executorMemoryWithOverheadMiB = executorMemoryMiB + memoryOverheadMiB
 
   private val executorCores = conf.getDouble("spark.executor.cores", 1d)
   private val executorLimitCores = conf.getOption(KUBERNETES_EXECUTOR_LIMIT_CORES.key)
@@ -441,10 +441,10 @@ private[spark] class KubernetesClusterSchedulerBackend(
       SPARK_ROLE_LABEL -> SPARK_POD_EXECUTOR_ROLE) ++
       executorLabels
     val executorMemoryQuantity = new QuantityBuilder(false)
-      .withAmount(s"${executorMemoryMb}M")
+      .withAmount(s"${executorMemoryMiB}Mi")
       .build()
     val executorMemoryLimitQuantity = new QuantityBuilder(false)
-      .withAmount(s"${executorMemoryWithOverhead}M")
+      .withAmount(s"${executorMemoryWithOverheadMiB}Mi")
       .build()
     val executorCpuQuantity = new QuantityBuilder(false)
       .withAmount(executorCores.toString)

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/kubernetes/submit/submitsteps/BaseDriverConfigurationStepSuite.scala

@@ -81,17 +81,17 @@ private[spark] class BaseDriverConfigurationStepSuite extends SparkFunSuite {
         .toMap
     assert(envs.size === 6)
     assert(envs(ENV_SUBMIT_EXTRA_CLASSPATH) === "/opt/spark/spark-exmaples.jar")
-    assert(envs(ENV_DRIVER_MEMORY) === "456m")
+    assert(envs(ENV_DRIVER_MEMORY) === "456M")
     assert(envs(ENV_DRIVER_MAIN_CLASS) === MAIN_CLASS)
     assert(envs(ENV_DRIVER_ARGS) === "arg1 arg2")
     assert(envs(DRIVER_CUSTOM_ENV_KEY1) === "customDriverEnv1")
     assert(envs(DRIVER_CUSTOM_ENV_KEY2) === "customDriverEnv2")
     val resourceRequirements = preparedDriverSpec.driverContainer.getResources
     val requests = resourceRequirements.getRequests.asScala
     assert(requests("cpu").getAmount === "2")
-    assert(requests("memory").getAmount === "256M")
+    assert(requests("memory").getAmount === "256Mi")
     val limits = resourceRequirements.getLimits.asScala
-    assert(limits("memory").getAmount === "456M")
+    assert(limits("memory").getAmount === "456Mi")
     assert(limits("cpu").getAmount === "4")
     val driverPodMetadata = preparedDriverSpec.driverPod.getMetadata
     assert(driverPodMetadata.getName === "spark-driver-pod")