Read cluster configs from configMap (apache#275)

Author: avovchenko

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

@@ -299,6 +299,12 @@ private[spark] object Config extends Logging {
       .stringConf
       .createOptional
 
+  val MAPR_CLUSTER_CONFIGMAP =
+    ConfigBuilder("spark.mapr.cluster.configMap")
+      .doc("Name of the mapr cluster config map")
+      .stringConf
+      .createWithDefault("mapr-cluster-configmap")
+
   val KUBERNETES_AUTH_SUBMISSION_CONF_PREFIX =
     "spark.kubernetes.authenticate.submission"
 

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

@@ -78,4 +78,9 @@ private[spark] object Constants {
   val KUBERNETES_MASTER_INTERNAL_URL = "https://kubernetes.default.svc"
   val DRIVER_CONTAINER_NAME = "spark-kubernetes-driver"
   val MEMORY_OVERHEAD_MIN_MIB = 384L
+
+  // Mapr specific Constants
+  val CLDB_HOSTS = "MAPR_CLDB_HOSTS"
+  val ZK_HOSTS = "MAPR_ZK_HOSTS"
+  val CLUSTER_NAME = "MAPR_CLUSTER"
 }

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/steps/BasicDriverConfigurationStep.scala

@@ -18,7 +18,7 @@ package org.apache.spark.deploy.k8s.submit.steps
 
 import scala.collection.JavaConverters._
 
-import io.fabric8.kubernetes.api.model.{ContainerBuilder, EnvVarBuilder, EnvVarSourceBuilder, PodBuilder, QuantityBuilder}
+import io.fabric8.kubernetes.api.model._
 
 import org.apache.spark.{SparkConf, SparkException}
 import org.apache.spark.deploy.k8s.Config._
@@ -111,6 +111,8 @@ private[spark] class BasicDriverConfigurationStep(
       ("cpu", new QuantityBuilder(false).withAmount(limitCores).build())
     }
 
+    val clusterConfMap = sparkConf.get(MAPR_CLUSTER_CONFIGMAP).toString
+
     val driverContainer = new ContainerBuilder(driverSpec.driverContainer)
       .withName(DRIVER_CONTAINER_NAME)
       .withImage(driverContainerImage)
@@ -136,6 +138,27 @@ private[spark] class BasicDriverConfigurationStep(
           .withNewFieldRef("v1", "status.podIP")
           .build())
         .endEnv()
+      .addNewEnv()
+        .withName(CLDB_HOSTS)
+        .withNewValueFrom()
+        .withConfigMapKeyRef(
+          new ConfigMapKeySelector(CLDB_HOSTS, clusterConfMap, true))
+        .endValueFrom()
+        .endEnv()
+      .addNewEnv()
+        .withName(ZK_HOSTS)
+        .withNewValueFrom()
+        .withConfigMapKeyRef(
+          new ConfigMapKeySelector(ZK_HOSTS, clusterConfMap, true))
+        .endValueFrom()
+        .endEnv()
+      .addNewEnv()
+        .withName(CLUSTER_NAME)
+        .withNewValueFrom()
+        .withConfigMapKeyRef(
+          new ConfigMapKeySelector(CLUSTER_NAME, clusterConfMap, true))
+        .endValueFrom()
+        .endEnv()
       .withNewResources()
         .addToRequests("cpu", driverCpuQuantity)
         .addToRequests("memory", driverMemoryQuantity)

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

@@ -178,11 +178,34 @@ private[spark] class ExecutorPodFactory(
           .build()
       }
 
+    val clusterConfMap = sparkConf.get(MAPR_CLUSTER_CONFIGMAP).toString
+
     val executorContainer = new ContainerBuilder()
       .withName("executor")
       .withImage(executorContainerImage)
       .withImagePullPolicy(imagePullPolicy)
       .addAllToEnv(clusterEnvs.asJava)
+      .addNewEnv()
+        .withName(CLDB_HOSTS)
+        .withNewValueFrom()
+        .withConfigMapKeyRef(
+           new ConfigMapKeySelector(CLDB_HOSTS, clusterConfMap, false))
+        .endValueFrom()
+        .endEnv()
+      .addNewEnv()
+        .withName(ZK_HOSTS)
+        .withNewValueFrom()
+        .withConfigMapKeyRef(
+          new ConfigMapKeySelector(ZK_HOSTS, clusterConfMap, false))
+        .endValueFrom()
+        .endEnv()
+      .addNewEnv()
+        .withName(CLUSTER_NAME)
+        .withNewValueFrom()
+        .withConfigMapKeyRef(
+          new ConfigMapKeySelector(CLUSTER_NAME, clusterConfMap, false))
+        .endValueFrom()
+        .endEnv()
       .withNewResources()
         .addToRequests("memory", executorMemoryQuantity)
         .addToLimits("memory", executorMemoryLimitQuantity)