From 6d89944775db2da829ab8f120c4f82765381409d Mon Sep 17 00:00:00 2001
From: Prashant Sharma <prashant@apache.org>
Date: Thu, 19 Dec 2019 13:22:04 +0530
Subject: [PATCH 01/15] [SPARK-25065] Allow setting up correct logging
 configuration on driver and executor.

---
 .../org/apache/spark/deploy/k8s/Config.scala  |  16 +-
 .../apache/spark/deploy/k8s/Constants.scala   |   6 +-
 .../features/MountLogConfFeatureStep.scala    | 141 ++++++++++++++++++
 .../k8s/submit/KubernetesDriverBuilder.scala  |   3 +-
 .../k8s/KubernetesExecutorBuilder.scala       |   3 +-
 .../MountLogConfFeatureStepSuite.scala        |  46 ++++++
 .../k8s/KubernetesExecutorBuilderSuite.scala  |   3 +-
 trigger                                       |   0
 8 files changed, 213 insertions(+), 5 deletions(-)
 create mode 100644 resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala
 create mode 100644 resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStepSuite.scala
 create mode 100644 trigger

diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala
index b326650476add..34c9da94209c8 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala
@@ -276,6 +276,20 @@ private[spark] object Config extends Logging {
       .stringConf
       .createOptional
 
+  val KUBERNETES_LOGGING_CONF_CONFIG_MAP =
+    ConfigBuilder("spark.kubernetes.loggingConf.configMapName")
+      .doc("Specify the name of the ConfigMap, containing the logging config files, " +
+        "to be mounted on the driver and executors for custom logger configuration.")
+      .stringConf
+      .createOptional
+
+  val KUBERNETES_LOGGING_CONF_FILE_NAME =
+    ConfigBuilder("spark.kubernetes.loggingConf.fileName")
+      .doc("Specify the name of the file, containing the logging configuraiton, " +
+        "to be mounted on the driver and executors for custom logger configuration.")
+      .stringConf
+      .createWithDefault("log4j.properties")
+
   val KUBERNETES_KERBEROS_DT_SECRET_NAME =
     ConfigBuilder("spark.kubernetes.kerberos.tokenSecret.name")
       .doc("Specify the name of the secret where your existing delegation tokens are stored. " +
@@ -359,7 +373,7 @@ private[spark] object Config extends Logging {
   val KUBERNETES_FILE_UPLOAD_PATH =
     ConfigBuilder("spark.kubernetes.file.upload.path")
       .doc("Hadoop compatible file system path where files from the local file system " +
-        "will be uploded to in cluster mode.")
+        "will be uploaded to in cluster mode.")
       .stringConf
       .createOptional
 
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Constants.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Constants.scala
index a3c74ff7b2885..d221291e84718 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Constants.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Constants.scala
@@ -67,7 +67,11 @@ private[spark] object Constants {
   val SPARK_CONF_FILE_NAME = "spark.properties"
   val SPARK_CONF_PATH = s"$SPARK_CONF_DIR_INTERNAL/$SPARK_CONF_FILE_NAME"
   val ENV_HADOOP_TOKEN_FILE_LOCATION = "HADOOP_TOKEN_FILE_LOCATION"
-
+  // The default logging conf mount is chosen to be at root, because in kubernetes,
+  // "subpath volume mount" has limitation that configmap updates are not applied.
+  val LOGGING_MOUNT_DIR = "/conf-logging"
+  // Logging configuration for containers.
+  val JAVA_OPT_FOR_LOGGING = s"-Dlog4j.configuration=file://$LOGGING_MOUNT_DIR/"
   // BINDINGS
   val ENV_PYSPARK_MAJOR_PYTHON_VERSION = "PYSPARK_MAJOR_PYTHON_VERSION"
 
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala
new file mode 100644
index 0000000000000..7c8039241d0af
--- /dev/null
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala
@@ -0,0 +1,141 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.k8s.features
+
+import java.net.URL
+import java.util.UUID
+
+import scala.collection.JavaConverters._
+import scala.io.Source
+
+import io.fabric8.kubernetes.api.model.{ConfigMap, ConfigMapBuilder, ContainerBuilder, EnvVar, EnvVarBuilder, HasMetadata, PodBuilder, VolumeMountBuilder}
+
+import org.apache.spark.deploy.k8s.{Config, KubernetesConf, SparkPod}
+import org.apache.spark.deploy.k8s.Constants._
+import org.apache.spark.internal.Logging
+import org.apache.spark.launcher.SparkLauncher.{DRIVER_EXTRA_JAVA_OPTIONS, EXECUTOR_EXTRA_JAVA_OPTIONS}
+
+/**
+ * Mounts the logger configuration from local configuration directory -
+ * on the spark job submitter's end or from a pre-defined config map.
+ */
+class MountLogConfFeatureStep(conf: KubernetesConf)
+  extends KubernetesFeatureConfigStep with Logging {
+  private val useExistingConfigMap = conf.get(Config.KUBERNETES_LOGGING_CONF_CONFIG_MAP).isDefined
+  private val configMapName: String = conf.get(Config.KUBERNETES_LOGGING_CONF_CONFIG_MAP)
+    .getOrElse(s"config-map-logging-conf-${UUID.randomUUID().toString.take(5)}")
+  private val loggingConfigFileName: String = conf.get(Config.KUBERNETES_LOGGING_CONF_FILE_NAME)
+  private val loggingConfURL: URL = this.getClass.getClassLoader.getResource(loggingConfigFileName)
+
+  private val featureEnabled: Boolean = {
+      (loggingConfURL != null || useExistingConfigMap)
+  }
+  private val loggerJVMProp = s"$JAVA_OPT_FOR_LOGGING${loggingConfigFileName}"
+
+  override def configurePod(pod: SparkPod): SparkPod = {
+    val logConfVolume = s"logger-conf-volume-${UUID.randomUUID().toString.take(5)}"
+    if (useExistingConfigMap) {
+      logInfo(s"Using an existing config map ${configMapName} for logging configuration.")
+    }
+    // Existing value of SPARK_CLASSPATH environment variable.
+    val sparkClasspath =
+      pod.container.getEnv.asScala.find(p => p.getName == ENV_CLASSPATH).map {x => x.getValue}
+    // All other environment variable except SPARK_CLASSPATH.
+    val envVars = pod.container.getEnv.asScala.filterNot(p => p.getName == ENV_CLASSPATH)
+    // Update the classpath with path to logger configuration file.
+    val updatedClasspath = if (sparkClasspath.isDefined) {
+      s"$LOGGING_MOUNT_DIR:${sparkClasspath.get}"
+    } else {
+      LOGGING_MOUNT_DIR
+    }
+    val sparkClasspathEnv =
+      new EnvVarBuilder().withName(ENV_CLASSPATH).withValue(updatedClasspath).build()
+    val podUpdated = if (featureEnabled) {
+      new PodBuilder(pod.pod)
+      .editSpec()
+        .addNewVolume()
+          .withName(logConfVolume)
+          .withNewConfigMap()
+            .withName(configMapName)
+            .endConfigMap()
+          .endVolume()
+        .endSpec()
+      .build()
+    } else {
+      logDebug(s"Logging configuration not found, mount not performed.")
+      pod.pod
+    }
+
+    val configMapVolumeMount = new VolumeMountBuilder()
+        .withName(logConfVolume)
+        .withReadOnly(true)
+      // We need a separate mounting dir for logging because,
+      // Mounting a ConfigMap has limitation that the mounted directory can hold only 1 file.
+        .withMountPath(LOGGING_MOUNT_DIR)
+        .build()
+    // As per https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.14/
+    // #configmapvolumesource-v1-core
+    // Since we do not want the pod creation to fail, in absence of the config map.
+    configMapVolumeMount.setAdditionalProperty("optional", "true")
+
+    val containerUpdated =
+      if (featureEnabled) {
+        new ContainerBuilder(pod.container)
+          .withEnv((envVars ++ Seq(sparkClasspathEnv)).asJava)
+          .withVolumeMounts(configMapVolumeMount)
+          .build()
+      } else {
+        pod.container
+      }
+    SparkPod(podUpdated, containerUpdated)
+  }
+
+  private def buildConfigMap(configMapName: String): ConfigMap = {
+    val loggerConfStream =
+      this.getClass.getClassLoader.getResourceAsStream(loggingConfigFileName)
+    logInfo(s"Logging configuration is picked up from: $loggingConfURL")
+    val loggerConfString = Source.createBufferedSource(loggerConfStream).getLines().mkString("\n")
+    new ConfigMapBuilder()
+      .withNewMetadata()
+        .withName(configMapName)
+        .endMetadata()
+      .addToData(loggingConfigFileName, loggerConfString)
+      .build()
+  }
+
+  override def getAdditionalPodSystemProperties(): Map[String, String] = {
+    if (featureEnabled) {
+      val executorJavaOpts =
+        s"$loggerJVMProp ${conf.get(EXECUTOR_EXTRA_JAVA_OPTIONS, "")}"
+      val driverJavaOpts =
+        s"$loggerJVMProp ${conf.get(DRIVER_EXTRA_JAVA_OPTIONS, "")}"
+      Map((EXECUTOR_EXTRA_JAVA_OPTIONS -> executorJavaOpts),
+        (DRIVER_EXTRA_JAVA_OPTIONS -> driverJavaOpts),
+        (Config.KUBERNETES_LOGGING_CONF_CONFIG_MAP.key -> configMapName))
+    } else {
+      Map.empty[String, String]
+    }
+  }
+  override def getAdditionalKubernetesResources(): Seq[HasMetadata] = {
+    if (featureEnabled && !useExistingConfigMap) {
+      Seq(buildConfigMap(configMapName))
+    } else {
+      Seq[HasMetadata]()
+    }
+  }
+}
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/KubernetesDriverBuilder.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/KubernetesDriverBuilder.scala
index 43639a3b7dc1b..54856d4390a7e 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/KubernetesDriverBuilder.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/KubernetesDriverBuilder.scala
@@ -48,7 +48,8 @@ private[spark] class KubernetesDriverBuilder {
       new HadoopConfDriverFeatureStep(conf),
       new KerberosConfDriverFeatureStep(conf),
       new PodTemplateConfigMapStep(conf),
-      new LocalDirsFeatureStep(conf))
+      new LocalDirsFeatureStep(conf),
+      new MountLogConfFeatureStep(conf))
 
     val spec = KubernetesDriverSpec(
       initialPod,
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesExecutorBuilder.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesExecutorBuilder.scala
index 22bff2c807330..a945ccd4ccb89 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesExecutorBuilder.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesExecutorBuilder.scala
@@ -45,7 +45,8 @@ private[spark] class KubernetesExecutorBuilder {
       new MountSecretsFeatureStep(conf),
       new EnvSecretsFeatureStep(conf),
       new MountVolumesFeatureStep(conf),
-      new LocalDirsFeatureStep(conf))
+      new LocalDirsFeatureStep(conf),
+      new MountLogConfFeatureStep(conf))
 
     features.foldLeft(initialPod) { case (pod, feature) => feature.configurePod(pod) }
   }
diff --git a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStepSuite.scala b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStepSuite.scala
new file mode 100644
index 0000000000000..1c71be0ae4e95
--- /dev/null
+++ b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStepSuite.scala
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.k8s.features
+
+import org.apache.spark.{SparkConf, SparkFunSuite}
+import org.apache.spark.deploy.k8s.{Config, KubernetesTestConf, SparkPod}
+import org.apache.spark.launcher.SparkLauncher
+
+class MountLogConfFeatureStepSuite extends SparkFunSuite {
+  // TODO add more tests.
+  test("Do not enable mount logging conf feature," +
+    " if neither of logging configuration file or user defined config map is configured.") {
+    val sparkConf = new SparkConf(false)
+      .set(SparkLauncher.DEPLOY_MODE, "cluster")
+      .set(Config.KUBERNETES_LOGGING_CONF_FILE_NAME, "some-non-existent-file-name.none")
+    val conf = KubernetesTestConf.createDriverConf(sparkConf = sparkConf)
+    val step = new MountLogConfFeatureStep(conf)
+    val initPod = SparkPod.initialPod()
+    val configuredPod = step.configurePod(initPod)
+    assert(configuredPod == initPod, "Pod should be unchanged.")
+  }
+
+  test("Do not create a config map if user provided configMap is configured.") {
+    val sparkConf = new SparkConf(false)
+      .set(SparkLauncher.DEPLOY_MODE, "cluster")
+      .set(Config.KUBERNETES_LOGGING_CONF_CONFIG_MAP, "user-created-config-map-name")
+    val conf = KubernetesTestConf.createDriverConf(sparkConf = sparkConf)
+    val step = new MountLogConfFeatureStep(conf)
+    assert(step.getAdditionalKubernetesResources() == Seq.empty)
+  }
+}
diff --git a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesExecutorBuilderSuite.scala b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesExecutorBuilderSuite.scala
index bd716174a8271..54543900ab419 100644
--- a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesExecutorBuilderSuite.scala
+++ b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesExecutorBuilderSuite.scala
@@ -28,7 +28,8 @@ class KubernetesExecutorBuilderSuite extends PodBuilderSuite {
     Config.KUBERNETES_EXECUTOR_PODTEMPLATE_FILE
   }
 
-  override protected def buildPod(sparkConf: SparkConf, client: KubernetesClient): SparkPod = {
+  override protected def buildPod(sparkConf: SparkConf,
+      client: KubernetesClient): SparkPod = {
     sparkConf.set("spark.driver.host", "https://driver.host.com")
     val conf = KubernetesTestConf.createExecutorConf(sparkConf = sparkConf)
     val secMgr = new SecurityManager(sparkConf)
diff --git a/trigger b/trigger
new file mode 100644
index 0000000000000..e69de29bb2d1d

From 195f81aec78e4e7b7ddb6e13832d781421707fb4 Mon Sep 17 00:00:00 2001
From: Prashant Sharma <prashant@apache.org>
Date: Wed, 8 Jan 2020 11:37:43 +0530
Subject: [PATCH 02/15] Improved the picking up mechanism of logger
 configuration.

---
 .../apache/spark/deploy/k8s/Constants.scala   |  2 -
 .../features/MountLogConfFeatureStep.scala    | 41 +++++++------------
 2 files changed, 15 insertions(+), 28 deletions(-)

diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Constants.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Constants.scala
index d221291e84718..c12e29f02fe49 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Constants.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Constants.scala
@@ -70,8 +70,6 @@ private[spark] object Constants {
   // The default logging conf mount is chosen to be at root, because in kubernetes,
   // "subpath volume mount" has limitation that configmap updates are not applied.
   val LOGGING_MOUNT_DIR = "/conf-logging"
-  // Logging configuration for containers.
-  val JAVA_OPT_FOR_LOGGING = s"-Dlog4j.configuration=file://$LOGGING_MOUNT_DIR/"
   // BINDINGS
   val ENV_PYSPARK_MAJOR_PYTHON_VERSION = "PYSPARK_MAJOR_PYTHON_VERSION"
 
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala
index 7c8039241d0af..2593d1e1b0e2c 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala
@@ -23,11 +23,12 @@ import java.util.UUID
 import scala.collection.JavaConverters._
 import scala.io.Source
 
-import io.fabric8.kubernetes.api.model.{ConfigMap, ConfigMapBuilder, ContainerBuilder, EnvVar, EnvVarBuilder, HasMetadata, PodBuilder, VolumeMountBuilder}
+import io.fabric8.kubernetes.api.model.{ConfigMap, ConfigMapBuilder, ContainerBuilder, EnvVarBuilder, HasMetadata, PodBuilder, VolumeMountBuilder}
 
 import org.apache.spark.deploy.k8s.{Config, KubernetesConf, SparkPod}
 import org.apache.spark.deploy.k8s.Constants._
 import org.apache.spark.internal.Logging
+import org.apache.spark.launcher.SparkLauncher
 import org.apache.spark.launcher.SparkLauncher.{DRIVER_EXTRA_JAVA_OPTIONS, EXECUTOR_EXTRA_JAVA_OPTIONS}
 
 /**
@@ -36,35 +37,28 @@ import org.apache.spark.launcher.SparkLauncher.{DRIVER_EXTRA_JAVA_OPTIONS, EXECU
  */
 class MountLogConfFeatureStep(conf: KubernetesConf)
   extends KubernetesFeatureConfigStep with Logging {
+  // Logging configuration for containers.
+  val JAVA_OPT_FOR_LOGGING = s"-Dlog4j.configuration=file://$LOGGING_MOUNT_DIR/"
+
   private val useExistingConfigMap = conf.get(Config.KUBERNETES_LOGGING_CONF_CONFIG_MAP).isDefined
   private val configMapName: String = conf.get(Config.KUBERNETES_LOGGING_CONF_CONFIG_MAP)
     .getOrElse(s"config-map-logging-conf-${UUID.randomUUID().toString.take(5)}")
   private val loggingConfigFileName: String = conf.get(Config.KUBERNETES_LOGGING_CONF_FILE_NAME)
   private val loggingConfURL: URL = this.getClass.getClassLoader.getResource(loggingConfigFileName)
+  private val loggerJVMProp = s"$JAVA_OPT_FOR_LOGGING${loggingConfigFileName}"
 
   private val featureEnabled: Boolean = {
-      (loggingConfURL != null || useExistingConfigMap)
+    (loggingConfURL != null &&
+      conf.get(SparkLauncher.DEPLOY_MODE).equalsIgnoreCase("cluster")) ||
+      useExistingConfigMap
   }
-  private val loggerJVMProp = s"$JAVA_OPT_FOR_LOGGING${loggingConfigFileName}"
 
   override def configurePod(pod: SparkPod): SparkPod = {
     val logConfVolume = s"logger-conf-volume-${UUID.randomUUID().toString.take(5)}"
     if (useExistingConfigMap) {
       logInfo(s"Using an existing config map ${configMapName} for logging configuration.")
     }
-    // Existing value of SPARK_CLASSPATH environment variable.
-    val sparkClasspath =
-      pod.container.getEnv.asScala.find(p => p.getName == ENV_CLASSPATH).map {x => x.getValue}
-    // All other environment variable except SPARK_CLASSPATH.
-    val envVars = pod.container.getEnv.asScala.filterNot(p => p.getName == ENV_CLASSPATH)
-    // Update the classpath with path to logger configuration file.
-    val updatedClasspath = if (sparkClasspath.isDefined) {
-      s"$LOGGING_MOUNT_DIR:${sparkClasspath.get}"
-    } else {
-      LOGGING_MOUNT_DIR
-    }
-    val sparkClasspathEnv =
-      new EnvVarBuilder().withName(ENV_CLASSPATH).withValue(updatedClasspath).build()
+
     val podUpdated = if (featureEnabled) {
       new PodBuilder(pod.pod)
       .editSpec()
@@ -77,26 +71,20 @@ class MountLogConfFeatureStep(conf: KubernetesConf)
         .endSpec()
       .build()
     } else {
-      logDebug(s"Logging configuration not found, mount not performed.")
+      logDebug(s"Logging configuration mount not performed.")
       pod.pod
     }
 
     val configMapVolumeMount = new VolumeMountBuilder()
-        .withName(logConfVolume)
-        .withReadOnly(true)
+      .withName(logConfVolume)
       // We need a separate mounting dir for logging because,
       // Mounting a ConfigMap has limitation that the mounted directory can hold only 1 file.
-        .withMountPath(LOGGING_MOUNT_DIR)
-        .build()
-    // As per https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.14/
-    // #configmapvolumesource-v1-core
-    // Since we do not want the pod creation to fail, in absence of the config map.
-    configMapVolumeMount.setAdditionalProperty("optional", "true")
+      .withMountPath(LOGGING_MOUNT_DIR)
+      .build()
 
     val containerUpdated =
       if (featureEnabled) {
         new ContainerBuilder(pod.container)
-          .withEnv((envVars ++ Seq(sparkClasspathEnv)).asJava)
           .withVolumeMounts(configMapVolumeMount)
           .build()
       } else {
@@ -131,6 +119,7 @@ class MountLogConfFeatureStep(conf: KubernetesConf)
       Map.empty[String, String]
     }
   }
+
   override def getAdditionalKubernetesResources(): Seq[HasMetadata] = {
     if (featureEnabled && !useExistingConfigMap) {
       Seq(buildConfigMap(configMapName))

From 5d5d5891b35ae293786604c81ac5982093c3e467 Mon Sep 17 00:00:00 2001
From: Prashant Sharma <prashant@apache.org>
Date: Thu, 16 Jan 2020 16:22:19 +0530
Subject: [PATCH 03/15] Added documentation and tests

---
 docs/running-on-kubernetes.md                 | 15 ++++++++++++
 .../apache/spark/deploy/k8s/SparkPod.scala    |  6 ++---
 .../features/MountLogConfFeatureStep.scala    |  3 +--
 .../MountLogConfFeatureStepSuite.scala        | 23 ++++++++++++++++++-
 4 files changed, 41 insertions(+), 6 deletions(-)

diff --git a/docs/running-on-kubernetes.md b/docs/running-on-kubernetes.md
index 61d6154ccb084..b37ec9cae2c45 100644
--- a/docs/running-on-kubernetes.md
+++ b/docs/running-on-kubernetes.md
@@ -344,7 +344,22 @@ $ kubectl -n=<namespace> logs -f <driver-pod-name>
 The same logs can also be accessed through the
 [Kubernetes dashboard](https://kubernetes.io/docs/tasks/access-application-cluster/web-ui-dashboard/) if installed on
 the cluster.
+### Setting up custom logging configuration
+Spark uses `log4j.properties` under the conf dir as the standard way of setting up a logger configuration.
+
+1) Kubernetes lets us define a
+ [config map](https://kubernetes.io/docs/concepts/storage/volumes/#configmap) and can be sometimes convenient
+ to provide custom logging configuration using a config map. For example, config map can be updated irrespective
+ of point of submission of spark job.
+ Once the user defined config map is created inside the kubernetes cluster, we can mount it and configure
+ for spark to use it be setting the property: `spark.kubernetes.loggingConf.configMapName`.
+ 
+2) For spark cluster mode of deployment, the configuration file present inside the `conf/` dir is
+ by default auto configured as a kubernetes config map and setup for all executors and driver as logging
+ configuration. The name of the logging configuration file is by default `log4j.properties` and can be
+ altered using the property: `spark.kubernetes.loggingConf.fileName`.
 
+Setting up a `configMapName` takes precedence over the other methods of setting logger configuration. 
 ### Accessing Driver UI
 
 The UI associated with any application can be accessed locally using
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/SparkPod.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/SparkPod.scala
index fd1196368a7ff..d32b83889df73 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/SparkPod.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/SparkPod.scala
@@ -25,11 +25,11 @@ private[spark] case class SparkPod(pod: Pod, container: Container) {
    *
    * Use it like:
    *
-   *     original.modify { case pod =>
+   *     original.transform { case pod =>
    *       // update pod and return new one
-   *     }.modify { case pod =>
+   *     }.transform { case pod =>
    *       // more changes that create a new pod
-   *     }.modify {
+   *     }.transform {
    *       case pod if someCondition => // new pod
    *     }
    *
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala
index 2593d1e1b0e2c..17448cc964101 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala
@@ -20,10 +20,9 @@ package org.apache.spark.deploy.k8s.features
 import java.net.URL
 import java.util.UUID
 
-import scala.collection.JavaConverters._
 import scala.io.Source
 
-import io.fabric8.kubernetes.api.model.{ConfigMap, ConfigMapBuilder, ContainerBuilder, EnvVarBuilder, HasMetadata, PodBuilder, VolumeMountBuilder}
+import io.fabric8.kubernetes.api.model.{ConfigMap, ConfigMapBuilder, ContainerBuilder, HasMetadata, PodBuilder, VolumeMountBuilder}
 
 import org.apache.spark.deploy.k8s.{Config, KubernetesConf, SparkPod}
 import org.apache.spark.deploy.k8s.Constants._
diff --git a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStepSuite.scala b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStepSuite.scala
index 1c71be0ae4e95..9b7183d8091b0 100644
--- a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStepSuite.scala
+++ b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStepSuite.scala
@@ -23,7 +23,7 @@ import org.apache.spark.launcher.SparkLauncher
 
 class MountLogConfFeatureStepSuite extends SparkFunSuite {
   // TODO add more tests.
-  test("Do not enable mount logging conf feature," +
+  test("Do not enable mount logger config feature," +
     " if neither of logging configuration file or user defined config map is configured.") {
     val sparkConf = new SparkConf(false)
       .set(SparkLauncher.DEPLOY_MODE, "cluster")
@@ -35,6 +35,27 @@ class MountLogConfFeatureStepSuite extends SparkFunSuite {
     assert(configuredPod == initPod, "Pod should be unchanged.")
   }
 
+  test("Do not enable mount logger config for client mode, unless an explicit" +
+    " config map is configured") {
+    // For client mode we cannot create k8s resources like config maps, so we rely on explicit
+    // setup by the user.
+    val sparkConf = new SparkConf(false)
+      .set(SparkLauncher.DEPLOY_MODE, "client")
+      .set(Config.KUBERNETES_LOGGING_CONF_FILE_NAME, "log4j.properties")
+    val conf = KubernetesTestConf.createDriverConf(sparkConf = sparkConf)
+    val step = new MountLogConfFeatureStep(conf)
+    val initPod = SparkPod.initialPod()
+    val configuredPod = step.configurePod(initPod)
+    assert(configuredPod == initPod, "Pod should be unchanged.")
+    val configMapName = "config-map-logging"
+    sparkConf.set(Config.KUBERNETES_LOGGING_CONF_CONFIG_MAP, configMapName)
+    val conf2 = KubernetesTestConf.createDriverConf(sparkConf = sparkConf)
+    val step2 = new MountLogConfFeatureStep(conf2)
+    assert(step2.getAdditionalPodSystemProperties().nonEmpty)
+    val configuredPod2 = step2.configurePod(initPod)
+    assert(configuredPod2.pod.getSpec.getVolumes.get(0).getConfigMap.getName == configMapName)
+  }
+
   test("Do not create a config map if user provided configMap is configured.") {
     val sparkConf = new SparkConf(false)
       .set(SparkLauncher.DEPLOY_MODE, "cluster")

From b6ec75676cf24c567361edaedfd1a69f719af892 Mon Sep 17 00:00:00 2001
From: Prashant Sharma <prashant@apache.org>
Date: Tue, 4 Feb 2020 11:43:01 +0530
Subject: [PATCH 04/15] Added more tests and cleaned up code.

---
 .../apache/spark/deploy/k8s/Constants.scala   |   2 +-
 .../features/MountLogConfFeatureStep.scala    |  56 +++++-----
 .../MountLogConfFeatureStepSuite.scala        |   2 +-
 .../log-config-test-log4j.properties          |  23 ++++
 .../k8s/integrationtest/KubernetesSuite.scala |   3 +-
 .../MountLoggerConfigMapSuite.scala           | 105 ++++++++++++++++++
 6 files changed, 161 insertions(+), 30 deletions(-)
 create mode 100644 resource-managers/kubernetes/integration-tests/src/test/resources/log-config-test-log4j.properties
 create mode 100644 resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/MountLoggerConfigMapSuite.scala

diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Constants.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Constants.scala
index c12e29f02fe49..aed627534dcf4 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Constants.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Constants.scala
@@ -69,7 +69,7 @@ private[spark] object Constants {
   val ENV_HADOOP_TOKEN_FILE_LOCATION = "HADOOP_TOKEN_FILE_LOCATION"
   // The default logging conf mount is chosen to be at root, because in kubernetes,
   // "subpath volume mount" has limitation that configmap updates are not applied.
-  val LOGGING_MOUNT_DIR = "/conf-logging"
+  val LOGGING_MOUNT_DIR = "/opt/spark/log"
   // BINDINGS
   val ENV_PYSPARK_MAJOR_PYTHON_VERSION = "PYSPARK_MAJOR_PYTHON_VERSION"
 
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala
index 17448cc964101..61bfa4441185b 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala
@@ -36,15 +36,14 @@ import org.apache.spark.launcher.SparkLauncher.{DRIVER_EXTRA_JAVA_OPTIONS, EXECU
  */
 class MountLogConfFeatureStep(conf: KubernetesConf)
   extends KubernetesFeatureConfigStep with Logging {
-  // Logging configuration for containers.
-  val JAVA_OPT_FOR_LOGGING = s"-Dlog4j.configuration=file://$LOGGING_MOUNT_DIR/"
 
   private val useExistingConfigMap = conf.get(Config.KUBERNETES_LOGGING_CONF_CONFIG_MAP).isDefined
   private val configMapName: String = conf.get(Config.KUBERNETES_LOGGING_CONF_CONFIG_MAP)
-    .getOrElse(s"config-map-logging-conf-${UUID.randomUUID().toString.take(5)}")
+    .getOrElse(s"config-map-logger-${UUID.randomUUID().toString.take(3)}")
   private val loggingConfigFileName: String = conf.get(Config.KUBERNETES_LOGGING_CONF_FILE_NAME)
   private val loggingConfURL: URL = this.getClass.getClassLoader.getResource(loggingConfigFileName)
-  private val loggerJVMProp = s"$JAVA_OPT_FOR_LOGGING${loggingConfigFileName}"
+  private val loggerJVMProp =
+    s"${MountLogConfFeatureStep.JAVA_OPT_FOR_LOGGING}${loggingConfigFileName}"
 
   private val featureEnabled: Boolean = {
     (loggingConfURL != null &&
@@ -53,7 +52,7 @@ class MountLogConfFeatureStep(conf: KubernetesConf)
   }
 
   override def configurePod(pod: SparkPod): SparkPod = {
-    val logConfVolume = s"logger-conf-volume-${UUID.randomUUID().toString.take(5)}"
+    val logConfVolume = s"log-conf-vol-${UUID.randomUUID().toString.take(3)}"
     if (useExistingConfigMap) {
       logInfo(s"Using an existing config map ${configMapName} for logging configuration.")
     }
@@ -74,17 +73,13 @@ class MountLogConfFeatureStep(conf: KubernetesConf)
       pod.pod
     }
 
-    val configMapVolumeMount = new VolumeMountBuilder()
-      .withName(logConfVolume)
-      // We need a separate mounting dir for logging because,
-      // Mounting a ConfigMap has limitation that the mounted directory can hold only 1 file.
-      .withMountPath(LOGGING_MOUNT_DIR)
-      .build()
-
     val containerUpdated =
       if (featureEnabled) {
         new ContainerBuilder(pod.container)
-          .withVolumeMounts(configMapVolumeMount)
+            .addNewVolumeMount()
+            .withName(logConfVolume)
+              .withMountPath(LOGGING_MOUNT_DIR)
+              .endVolumeMount()
           .build()
       } else {
         pod.container
@@ -92,19 +87,6 @@ class MountLogConfFeatureStep(conf: KubernetesConf)
     SparkPod(podUpdated, containerUpdated)
   }
 
-  private def buildConfigMap(configMapName: String): ConfigMap = {
-    val loggerConfStream =
-      this.getClass.getClassLoader.getResourceAsStream(loggingConfigFileName)
-    logInfo(s"Logging configuration is picked up from: $loggingConfURL")
-    val loggerConfString = Source.createBufferedSource(loggerConfStream).getLines().mkString("\n")
-    new ConfigMapBuilder()
-      .withNewMetadata()
-        .withName(configMapName)
-        .endMetadata()
-      .addToData(loggingConfigFileName, loggerConfString)
-      .build()
-  }
-
   override def getAdditionalPodSystemProperties(): Map[String, String] = {
     if (featureEnabled) {
       val executorJavaOpts =
@@ -121,9 +103,29 @@ class MountLogConfFeatureStep(conf: KubernetesConf)
 
   override def getAdditionalKubernetesResources(): Seq[HasMetadata] = {
     if (featureEnabled && !useExistingConfigMap) {
-      Seq(buildConfigMap(configMapName))
+      Seq(MountLogConfFeatureStep
+        .buildConfigMap(loggingConfURL, loggingConfigFileName, configMapName))
     } else {
       Seq[HasMetadata]()
     }
   }
 }
+
+private[k8s] object MountLogConfFeatureStep extends Logging {
+  // Logging configuration for containers.
+  val JAVA_OPT_FOR_LOGGING = s"-Dlog4j.configuration=file://$LOGGING_MOUNT_DIR/"
+
+  // exposed for testing.
+  private[k8s] def buildConfigMap(
+      loggingConfUrl: URL, loggingConfigFileName: String, configMapName: String): ConfigMap = {
+    logInfo(s"Logging configuration is picked up from: $loggingConfigFileName")
+    val loggerConfStream = loggingConfUrl.openStream()
+    val loggerConfString = Source.createBufferedSource(loggerConfStream).getLines().mkString("\n")
+    new ConfigMapBuilder()
+      .withNewMetadata()
+      .withName(configMapName)
+      .endMetadata()
+      .addToData(loggingConfigFileName, loggerConfString)
+      .build()
+  }
+}
\ No newline at end of file
diff --git a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStepSuite.scala b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStepSuite.scala
index 9b7183d8091b0..b6bf9fba9663b 100644
--- a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStepSuite.scala
+++ b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStepSuite.scala
@@ -22,7 +22,7 @@ import org.apache.spark.deploy.k8s.{Config, KubernetesTestConf, SparkPod}
 import org.apache.spark.launcher.SparkLauncher
 
 class MountLogConfFeatureStepSuite extends SparkFunSuite {
-  // TODO add more tests.
+
   test("Do not enable mount logger config feature," +
     " if neither of logging configuration file or user defined config map is configured.") {
     val sparkConf = new SparkConf(false)
diff --git a/resource-managers/kubernetes/integration-tests/src/test/resources/log-config-test-log4j.properties b/resource-managers/kubernetes/integration-tests/src/test/resources/log-config-test-log4j.properties
new file mode 100644
index 0000000000000..0c89ff17bd188
--- /dev/null
+++ b/resource-managers/kubernetes/integration-tests/src/test/resources/log-config-test-log4j.properties
@@ -0,0 +1,23 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# This logger file is for integration test MountLoggerConfigMapSuite.
+log4j.rootCategory=DEBUG, console
+log4j.appender.console=org.apache.log4j.ConsoleAppender
+log4j.appender.console.target=System.err
+log4j.appender.console.layout=org.apache.log4j.PatternLayout
+log4j.appender.console.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss} %p %c: %m%n
diff --git a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/KubernetesSuite.scala b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/KubernetesSuite.scala
index 0d4fcccc35cf9..42bbf594ea05a 100644
--- a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/KubernetesSuite.scala
+++ b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/KubernetesSuite.scala
@@ -42,7 +42,8 @@ import org.apache.spark.internal.config._
 class KubernetesSuite extends SparkFunSuite
   with BeforeAndAfterAll with BeforeAndAfter with BasicTestsSuite with SecretsTestsSuite
   with PythonTestsSuite with ClientModeTestsSuite with PodTemplateSuite with PVTestsSuite
-  with DepsTestsSuite with RTestsSuite with Logging with Eventually with Matchers {
+  with MountLoggerConfigMapSuite with DepsTestsSuite with RTestsSuite with Logging with Eventually
+  with Matchers {
 
   import KubernetesSuite._
 
diff --git a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/MountLoggerConfigMapSuite.scala b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/MountLoggerConfigMapSuite.scala
new file mode 100644
index 0000000000000..450cf7a72177c
--- /dev/null
+++ b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/MountLoggerConfigMapSuite.scala
@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.deploy.k8s.integrationtest
+
+import java.io.{BufferedWriter, File, FileWriter}
+import java.net.URL
+
+import io.fabric8.kubernetes.api.model.{ConfigMap, ConfigMapBuilder}
+import scala.io.{BufferedSource, Source}
+
+private[spark] trait MountLoggerConfigMapSuite { k8sSuite: KubernetesSuite =>
+  import KubernetesSuite._
+
+  private var configMap: ConfigMap = _
+
+  private def buildConfigMap(
+      loggingConfUrl: URL, loggingConfigFileName: String, configMapName: String): ConfigMap = {
+    logInfo(s"Logging configuration is picked up from: $loggingConfigFileName")
+    val loggerConfStream = loggingConfUrl.openStream()
+    val loggerConfString = Source.createBufferedSource(loggerConfStream).getLines().mkString("\n")
+    new ConfigMapBuilder()
+      .withNewMetadata()
+      .withName(configMapName)
+      .endMetadata()
+      .addToData(loggingConfigFileName, loggerConfString)
+      .build()
+  }
+
+  test("Run with logging configuration provided as a k8s config map", k8sTestTag) {
+    val loggingConfigFileName = "log-config-test-log4j.properties"
+    val configMapName = "test-logging-config-map"
+    val loggingConfURL: URL = this.getClass.getClassLoader.getResource(loggingConfigFileName)
+    assert(loggingConfURL != null, "Test not properly setup, logging configuration file not available.")
+    try {
+      configMap = buildConfigMap(loggingConfURL, loggingConfigFileName, configMapName)
+
+      kubernetesTestComponents.kubernetesClient.configMaps().createOrReplace(configMap)
+
+      sparkAppConf
+        .set("spark.kubernetes.loggingConf.configMapName", configMapName)
+        .set("spark.kubernetes.loggingConf.fileName", loggingConfigFileName)
+      runSparkApplicationAndVerifyCompletion(
+        appResource = containerLocalSparkDistroExamplesJar,
+        mainClass = SPARK_PI_MAIN_CLASS,
+        expectedLogOnCompletion = (Seq("DEBUG",
+          s"Using an existing config map ${configMapName}", "Pi is roughly 3")),
+        appArgs = Array.empty[String],
+        driverPodChecker = doBasicDriverPodCheck,
+        executorPodChecker = doBasicExecutorPodCheck,
+        appLocator = appLocator,
+        isJVM = true)
+    } finally {
+      kubernetesTestComponents.kubernetesClient.configMaps().delete(configMap)
+    }
+  }
+
+  test("Run with logging configuration provided from log4j properties file", k8sTestTag) {
+    val loggingConfigFileName = "log-config-test-log4j.properties"
+    val loggingConfURL: URL = this.getClass.getClassLoader.getResource(loggingConfigFileName)
+    assert(loggingConfURL != null, "Test not properly setup, logging configuration file not available.")
+
+    val content = Source.createBufferedSource(loggingConfURL.openStream()).getLines().mkString("\n")
+    val logConfFilePath = s"${sparkHomeDir.toFile}/conf/$loggingConfigFileName"
+
+    try {
+      val writer = new BufferedWriter(new FileWriter(logConfFilePath))
+      writer.write(content)
+      writer.close()
+
+      sparkAppConf
+        .set("spark.kubernetes.loggingConf.fileName", loggingConfigFileName)
+        .set("spark.driver.extraJavaOptions", "-Dlog4j.debug")
+
+      runSparkApplicationAndVerifyCompletion(
+        appResource = containerLocalSparkDistroExamplesJar,
+        mainClass = SPARK_PI_MAIN_CLASS,
+        expectedLogOnCompletion = (Seq("DEBUG",
+          s"log4j: Reading configuration from URL file:/opt/spark/log/$loggingConfigFileName",
+          "Pi is roughly 3")),
+        appArgs = Array.empty[String],
+        driverPodChecker = doBasicDriverPodCheck,
+        executorPodChecker = doBasicExecutorPodCheck,
+        appLocator = appLocator,
+        isJVM = true)
+    } finally {
+      new File(logConfFilePath).delete()
+    }
+  }
+
+
+}

From 05b427391f048240d3f7cd38b0055c70c3a80987 Mon Sep 17 00:00:00 2001
From: Prashant Sharma <prashant@apache.org>
Date: Tue, 4 Feb 2020 11:50:18 +0530
Subject: [PATCH 05/15] Clarified the usage a bit more in documentation.

---
 docs/running-on-kubernetes.md | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/docs/running-on-kubernetes.md b/docs/running-on-kubernetes.md
index b37ec9cae2c45..405f1b5d98820 100644
--- a/docs/running-on-kubernetes.md
+++ b/docs/running-on-kubernetes.md
@@ -352,7 +352,10 @@ Spark uses `log4j.properties` under the conf dir as the standard way of setting
  to provide custom logging configuration using a config map. For example, config map can be updated irrespective
  of point of submission of spark job.
  Once the user defined config map is created inside the kubernetes cluster, we can mount it and configure
- for spark to use it be setting the property: `spark.kubernetes.loggingConf.configMapName`.
+ for spark to use it, by setting the property: `spark.kubernetes.loggingConf.configMapName` and 
+ `spark.kubernetes.loggingConf.fileName`. `spark.kubernetes.loggingConf.configMapName` is the name of config map
+  in kubernetes and `spark.kubernetes.loggingConf.fileName` is the name of the file used to create the
+  config map.
  
 2) For spark cluster mode of deployment, the configuration file present inside the `conf/` dir is
  by default auto configured as a kubernetes config map and setup for all executors and driver as logging

From 692b7a68f255e7831b8ad197827072d152a3504d Mon Sep 17 00:00:00 2001
From: Prashant Sharma <prashant@apache.org>
Date: Tue, 4 Feb 2020 13:00:24 +0530
Subject: [PATCH 06/15] Delete accidentally committed file.

---
 trigger | 0
 1 file changed, 0 insertions(+), 0 deletions(-)
 delete mode 100644 trigger

diff --git a/trigger b/trigger
deleted file mode 100644
index e69de29bb2d1d..0000000000000

From 9af175f1ce2a67c61e6089aa0f1e3478fd0d8450 Mon Sep 17 00:00:00 2001
From: Prashant Sharma <prashant@apache.org>
Date: Tue, 4 Feb 2020 13:14:46 +0530
Subject: [PATCH 07/15] Fixed scalastyle issues and improved some comments.

---
 .../spark/deploy/k8s/features/MountLogConfFeatureStep.scala     | 2 +-
 .../src/test/resources/log-config-test-log4j.properties         | 2 +-
 .../deploy/k8s/integrationtest/MountLoggerConfigMapSuite.scala  | 2 --
 3 files changed, 2 insertions(+), 4 deletions(-)

diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala
index 61bfa4441185b..949ee7f2d4b6e 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala
@@ -128,4 +128,4 @@ private[k8s] object MountLogConfFeatureStep extends Logging {
       .addToData(loggingConfigFileName, loggerConfString)
       .build()
   }
-}
\ No newline at end of file
+}
diff --git a/resource-managers/kubernetes/integration-tests/src/test/resources/log-config-test-log4j.properties b/resource-managers/kubernetes/integration-tests/src/test/resources/log-config-test-log4j.properties
index 0c89ff17bd188..f9f3ce6350dc0 100644
--- a/resource-managers/kubernetes/integration-tests/src/test/resources/log-config-test-log4j.properties
+++ b/resource-managers/kubernetes/integration-tests/src/test/resources/log-config-test-log4j.properties
@@ -15,7 +15,7 @@
 # limitations under the License.
 #
 
-# This logger file is for integration test MountLoggerConfigMapSuite.
+# This log4j config file is for integration test MountLoggerConfigMapSuite.
 log4j.rootCategory=DEBUG, console
 log4j.appender.console=org.apache.log4j.ConsoleAppender
 log4j.appender.console.target=System.err
diff --git a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/MountLoggerConfigMapSuite.scala b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/MountLoggerConfigMapSuite.scala
index 450cf7a72177c..27f08079f2db2 100644
--- a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/MountLoggerConfigMapSuite.scala
+++ b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/MountLoggerConfigMapSuite.scala
@@ -100,6 +100,4 @@ private[spark] trait MountLoggerConfigMapSuite { k8sSuite: KubernetesSuite =>
       new File(logConfFilePath).delete()
     }
   }
-
-
 }

From 8ee88f64acadd384976cafb9aafcefb9dd6c08a9 Mon Sep 17 00:00:00 2001
From: Prashant Sharma <prashant@apache.org>
Date: Tue, 4 Feb 2020 13:41:20 +0530
Subject: [PATCH 08/15] Fixed comments

---
 .../src/main/scala/org/apache/spark/deploy/k8s/Constants.scala  | 2 --
 1 file changed, 2 deletions(-)

diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Constants.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Constants.scala
index aed627534dcf4..72e1bfca9422a 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Constants.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Constants.scala
@@ -67,8 +67,6 @@ private[spark] object Constants {
   val SPARK_CONF_FILE_NAME = "spark.properties"
   val SPARK_CONF_PATH = s"$SPARK_CONF_DIR_INTERNAL/$SPARK_CONF_FILE_NAME"
   val ENV_HADOOP_TOKEN_FILE_LOCATION = "HADOOP_TOKEN_FILE_LOCATION"
-  // The default logging conf mount is chosen to be at root, because in kubernetes,
-  // "subpath volume mount" has limitation that configmap updates are not applied.
   val LOGGING_MOUNT_DIR = "/opt/spark/log"
   // BINDINGS
   val ENV_PYSPARK_MAJOR_PYTHON_VERSION = "PYSPARK_MAJOR_PYTHON_VERSION"

From bd23cfea541cefa5599a97466f8a571e52f323b7 Mon Sep 17 00:00:00 2001
From: Prashant Sharma <prashant@apache.org>
Date: Tue, 4 Feb 2020 13:44:37 +0530
Subject: [PATCH 09/15] undo stray changes

---
 .../scheduler/cluster/k8s/KubernetesExecutorBuilderSuite.scala | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesExecutorBuilderSuite.scala b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesExecutorBuilderSuite.scala
index 54543900ab419..bd716174a8271 100644
--- a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesExecutorBuilderSuite.scala
+++ b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesExecutorBuilderSuite.scala
@@ -28,8 +28,7 @@ class KubernetesExecutorBuilderSuite extends PodBuilderSuite {
     Config.KUBERNETES_EXECUTOR_PODTEMPLATE_FILE
   }
 
-  override protected def buildPod(sparkConf: SparkConf,
-      client: KubernetesClient): SparkPod = {
+  override protected def buildPod(sparkConf: SparkConf, client: KubernetesClient): SparkPod = {
     sparkConf.set("spark.driver.host", "https://driver.host.com")
     val conf = KubernetesTestConf.createExecutorConf(sparkConf = sparkConf)
     val secMgr = new SecurityManager(sparkConf)

From 6b1e6c5219d84b0c335be440af1c2881bd8fa99a Mon Sep 17 00:00:00 2001
From: Prashant Sharma <prashant@apache.org>
Date: Tue, 4 Feb 2020 13:47:48 +0530
Subject: [PATCH 10/15] Fixed scalastyle issues and improved some comments.

---
 .../k8s/integrationtest/MountLoggerConfigMapSuite.scala       | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/MountLoggerConfigMapSuite.scala b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/MountLoggerConfigMapSuite.scala
index 27f08079f2db2..fd756c7941a46 100644
--- a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/MountLoggerConfigMapSuite.scala
+++ b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/MountLoggerConfigMapSuite.scala
@@ -44,7 +44,7 @@ private[spark] trait MountLoggerConfigMapSuite { k8sSuite: KubernetesSuite =>
     val loggingConfigFileName = "log-config-test-log4j.properties"
     val configMapName = "test-logging-config-map"
     val loggingConfURL: URL = this.getClass.getClassLoader.getResource(loggingConfigFileName)
-    assert(loggingConfURL != null, "Test not properly setup, logging configuration file not available.")
+    assert(loggingConfURL != null, "Logging configuration file not available.")
     try {
       configMap = buildConfigMap(loggingConfURL, loggingConfigFileName, configMapName)
 
@@ -71,7 +71,7 @@ private[spark] trait MountLoggerConfigMapSuite { k8sSuite: KubernetesSuite =>
   test("Run with logging configuration provided from log4j properties file", k8sTestTag) {
     val loggingConfigFileName = "log-config-test-log4j.properties"
     val loggingConfURL: URL = this.getClass.getClassLoader.getResource(loggingConfigFileName)
-    assert(loggingConfURL != null, "Test not properly setup, logging configuration file not available.")
+    assert(loggingConfURL != null, "Logging configuration file not available.")
 
     val content = Source.createBufferedSource(loggingConfURL.openStream()).getLines().mkString("\n")
     val logConfFilePath = s"${sparkHomeDir.toFile}/conf/$loggingConfigFileName"

From d44478b9b4d28409157b481cad291400e1fa07ee Mon Sep 17 00:00:00 2001
From: Prashant Sharma <prashant@apache.org>
Date: Tue, 4 Feb 2020 14:30:12 +0530
Subject: [PATCH 11/15] default value for deploymode

---
 .../spark/deploy/k8s/features/MountLogConfFeatureStep.scala    | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala
index 949ee7f2d4b6e..65e1929b1b4f2 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala
@@ -47,7 +47,8 @@ class MountLogConfFeatureStep(conf: KubernetesConf)
 
   private val featureEnabled: Boolean = {
     (loggingConfURL != null &&
-      conf.get(SparkLauncher.DEPLOY_MODE).equalsIgnoreCase("cluster")) ||
+      conf.getOption(SparkLauncher.DEPLOY_MODE).getOrElse("client")
+        .equalsIgnoreCase("cluster")) ||
       useExistingConfigMap
   }
 

From 0a13936e938a972d05f50304cb352ae5a295a08e Mon Sep 17 00:00:00 2001
From: Prashant Sharma <prashant@apache.org>
Date: Thu, 6 Feb 2020 11:31:29 +0530
Subject: [PATCH 12/15] Improved docs, added another test.

---
 docs/running-on-kubernetes.md                 |  9 +++---
 .../org/apache/spark/deploy/k8s/Config.scala  |  8 +++--
 .../features/MountLogConfFeatureStep.scala    |  1 -
 .../MountLogConfFeatureStepSuite.scala        | 32 +++++++++++++++----
 4 files changed, 36 insertions(+), 14 deletions(-)

diff --git a/docs/running-on-kubernetes.md b/docs/running-on-kubernetes.md
index 405f1b5d98820..d74b155d6a29c 100644
--- a/docs/running-on-kubernetes.md
+++ b/docs/running-on-kubernetes.md
@@ -345,12 +345,13 @@ The same logs can also be accessed through the
 [Kubernetes dashboard](https://kubernetes.io/docs/tasks/access-application-cluster/web-ui-dashboard/) if installed on
 the cluster.
 ### Setting up custom logging configuration
-Spark uses `log4j.properties` under the conf dir as the standard way of setting up a logger configuration.
+Spark uses `log4j.properties` under the `SPARK_HOME/conf` dir as the standard way of setting up a logger configuration.
 
 1) Kubernetes lets us define a
- [config map](https://kubernetes.io/docs/concepts/storage/volumes/#configmap) and can be sometimes convenient
- to provide custom logging configuration using a config map. For example, config map can be updated irrespective
- of point of submission of spark job.
+ [configMap](https://kubernetes.io/docs/concepts/storage/volumes/#configmap) and can be sometimes convenient
+ to provide custom logging configuration using a configMap. For example, configMap can be updated irrespective
+ of point of submission of spark job. This is particularly useful in automated deployment and building 
+ spark as a service. 
  Once the user defined config map is created inside the kubernetes cluster, we can mount it and configure
  for spark to use it, by setting the property: `spark.kubernetes.loggingConf.configMapName` and 
  `spark.kubernetes.loggingConf.fileName`. `spark.kubernetes.loggingConf.configMapName` is the name of config map
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala
index 34c9da94209c8..82eca02d32b9c 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala
@@ -278,15 +278,17 @@ private[spark] object Config extends Logging {
 
   val KUBERNETES_LOGGING_CONF_CONFIG_MAP =
     ConfigBuilder("spark.kubernetes.loggingConf.configMapName")
-      .doc("Specify the name of the ConfigMap, containing the logging config files, " +
+      .doc("Specify the name of the k8s ConfigMap, containing the logging configuration file, " +
         "to be mounted on the driver and executors for custom logger configuration.")
       .stringConf
       .createOptional
 
   val KUBERNETES_LOGGING_CONF_FILE_NAME =
     ConfigBuilder("spark.kubernetes.loggingConf.fileName")
-      .doc("Specify the name of the file, containing the logging configuraiton, " +
-        "to be mounted on the driver and executors for custom logger configuration.")
+      .doc("Specify the name of the file, containing the logging configuration, " +
+        "to be mounted on the driver and executors for custom logger configuration. " +
+        "This property needs to be set, even if logging is provided via a user defined" +
+        " ConfigMap, if the file name is not `log4j.properties`.")
       .stringConf
       .createWithDefault("log4j.properties")
 
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala
index 65e1929b1b4f2..e912408b2004e 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala
@@ -116,7 +116,6 @@ private[k8s] object MountLogConfFeatureStep extends Logging {
   // Logging configuration for containers.
   val JAVA_OPT_FOR_LOGGING = s"-Dlog4j.configuration=file://$LOGGING_MOUNT_DIR/"
 
-  // exposed for testing.
   private[k8s] def buildConfigMap(
       loggingConfUrl: URL, loggingConfigFileName: String, configMapName: String): ConfigMap = {
     logInfo(s"Logging configuration is picked up from: $loggingConfigFileName")
diff --git a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStepSuite.scala b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStepSuite.scala
index b6bf9fba9663b..074fc8d0c68f9 100644
--- a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStepSuite.scala
+++ b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStepSuite.scala
@@ -17,13 +17,17 @@
 
 package org.apache.spark.deploy.k8s.features
 
+import scala.collection.JavaConverters._
+
+import io.fabric8.kubernetes.api.model.Pod
+
 import org.apache.spark.{SparkConf, SparkFunSuite}
 import org.apache.spark.deploy.k8s.{Config, KubernetesTestConf, SparkPod}
 import org.apache.spark.launcher.SparkLauncher
 
 class MountLogConfFeatureStepSuite extends SparkFunSuite {
 
-  test("Do not enable mount logger config feature," +
+  test("Do not enable, mount logger configuration feature," +
     " if neither of logging configuration file or user defined config map is configured.") {
     val sparkConf = new SparkConf(false)
       .set(SparkLauncher.DEPLOY_MODE, "cluster")
@@ -35,10 +39,9 @@ class MountLogConfFeatureStepSuite extends SparkFunSuite {
     assert(configuredPod == initPod, "Pod should be unchanged.")
   }
 
-  test("Do not enable mount logger config for client mode, unless an explicit" +
-    " config map is configured") {
-    // For client mode we cannot create k8s resources like config maps, so we rely on explicit
-    // setup by the user.
+  test("Do not mount logger config for client mode," +
+    " unless an explicit config map is configured") {
+
     val sparkConf = new SparkConf(false)
       .set(SparkLauncher.DEPLOY_MODE, "client")
       .set(Config.KUBERNETES_LOGGING_CONF_FILE_NAME, "log4j.properties")
@@ -56,7 +59,7 @@ class MountLogConfFeatureStepSuite extends SparkFunSuite {
     assert(configuredPod2.pod.getSpec.getVolumes.get(0).getConfigMap.getName == configMapName)
   }
 
-  test("Do not create a config map if user provided configMap is configured.") {
+  test("Do not auto-create a ConfigMap if user provided ConfigMap is configured.") {
     val sparkConf = new SparkConf(false)
       .set(SparkLauncher.DEPLOY_MODE, "cluster")
       .set(Config.KUBERNETES_LOGGING_CONF_CONFIG_MAP, "user-created-config-map-name")
@@ -64,4 +67,21 @@ class MountLogConfFeatureStepSuite extends SparkFunSuite {
     val step = new MountLogConfFeatureStep(conf)
     assert(step.getAdditionalKubernetesResources() == Seq.empty)
   }
+
+  test("User provided ConfigMap takes precedence and is configured properly.") {
+    val configMapName = "user-created-config-map-name"
+    val sparkConf = new SparkConf(false)
+      .set(SparkLauncher.DEPLOY_MODE, "cluster")
+      .set(Config.KUBERNETES_LOGGING_CONF_CONFIG_MAP, configMapName)
+    val conf = KubernetesTestConf.createDriverConf(sparkConf = sparkConf)
+    val step = new MountLogConfFeatureStep(conf)
+    val configuredPod = step.configurePod(SparkPod.initialPod())
+    assert(hasConfigMap(configuredPod.pod, configMapName))
+  }
+
+  private def hasConfigMap(pod: Pod, configMapName: String): Boolean = {
+    pod.getSpec.getVolumes.asScala.exists { volume =>
+      volume.getConfigMap.getName == configMapName
+    }
+  }
 }

From e744c7eda70ff12da6caf227894274ccc5823263 Mon Sep 17 00:00:00 2001
From: Prashant Sharma <prashant@apache.org>
Date: Tue, 18 Feb 2020 13:55:08 +0530
Subject: [PATCH 13/15] Added new configuraiton introduced to configuration
 docs.

---
 docs/running-on-kubernetes.md                 | 24 ++++++++++++++++---
 .../org/apache/spark/deploy/k8s/Config.scala  |  8 +++----
 .../features/MountLogConfFeatureStep.scala    |  6 ++---
 3 files changed, 28 insertions(+), 10 deletions(-)

diff --git a/docs/running-on-kubernetes.md b/docs/running-on-kubernetes.md
index 463efe0df0d1d..28999c7879407 100644
--- a/docs/running-on-kubernetes.md
+++ b/docs/running-on-kubernetes.md
@@ -356,10 +356,10 @@ Spark uses `log4j.properties` under the `SPARK_HOME/conf` dir as the standard wa
  for spark to use it, by setting the property: `spark.kubernetes.loggingConf.configMapName` and 
  `spark.kubernetes.loggingConf.fileName`. `spark.kubernetes.loggingConf.configMapName` is the name of config map
   in kubernetes and `spark.kubernetes.loggingConf.fileName` is the name of the file used to create the
-  config map.
+  configMap or it is also referred to as the key of the configMap.
  
-2) For spark cluster mode of deployment, the configuration file present inside the `conf/` dir is
- by default auto configured as a kubernetes config map and setup for all executors and driver as logging
+2) For spark "cluster" mode of deployment, the configuration file present inside the `conf/` dir is
+ by default auto configured as a kubernetes configMap and setup for all executors and driver as logging
  configuration. The name of the logging configuration file is by default `log4j.properties` and can be
  altered using the property: `spark.kubernetes.loggingConf.fileName`.
 
@@ -593,6 +593,24 @@ See the [configuration page](configuration.html) for information on Spark config
     excessive CPU usage on the spark driver.
   </td>
 </tr>
+<tr>
+  <td><code>spark.kubernetes.loggingConf.configMapName</code></td>
+  <td>(none)</td>
+  <td>
+    Specify the name of the k8s ConfigMap, containing the logging configuration file to be mounted
+    on the driver and executors for custom logger configuration. If the key is not the default value
+    i.e. `log4j.properties`, then it should be specified via the property
+    <code>spark.kubernetes.loggingConf.fileName</code>.
+  </td>
+</tr>
+<tr>
+  <td><code>spark.kubernetes.loggingConf.fileName</code></td>
+  <td>log4j.properties</td>
+  <td>
+    Specify the name of the file, containing the logging configuration, to be mounted on the driver
+    and executors for custom logger configuration.
+  </td>
+</tr>
 <tr>
   <td><code>spark.kubernetes.authenticate.submission.caCertFile</code></td>
   <td>(none)</td>
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala
index 78723e089650f..63296e96daca0 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala
@@ -286,16 +286,16 @@ private[spark] object Config extends Logging {
   val KUBERNETES_LOGGING_CONF_CONFIG_MAP =
     ConfigBuilder("spark.kubernetes.loggingConf.configMapName")
       .doc("Specify the name of the k8s ConfigMap, containing the logging configuration file, " +
-        "to be mounted on the driver and executors for custom logger configuration.")
+        "to be mounted on the driver and executors for custom logger configuration. If the key " +
+          " is not the default value i.e. `log4j.properties` then please specify it via property" +
+            "`spark.kubernetes.loggingConf.fileName`.")
       .stringConf
       .createOptional
 
   val KUBERNETES_LOGGING_CONF_FILE_NAME =
     ConfigBuilder("spark.kubernetes.loggingConf.fileName")
       .doc("Specify the name of the file, containing the logging configuration, " +
-        "to be mounted on the driver and executors for custom logger configuration. " +
-        "This property needs to be set, even if logging is provided via a user defined" +
-        " ConfigMap, if the file name is not `log4j.properties`.")
+        "to be mounted on the driver and executors for custom logger configuration.")
       .stringConf
       .createWithDefault("log4j.properties")
 
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala
index e912408b2004e..fbf562df9f330 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala
@@ -21,9 +21,7 @@ import java.net.URL
 import java.util.UUID
 
 import scala.io.Source
-
-import io.fabric8.kubernetes.api.model.{ConfigMap, ConfigMapBuilder, ContainerBuilder, HasMetadata, PodBuilder, VolumeMountBuilder}
-
+import io.fabric8.kubernetes.api.model.{ConfigMap, ConfigMapBuilder, ContainerBuilder, HasMetadata, KeyToPath, PodBuilder, VolumeMountBuilder}
 import org.apache.spark.deploy.k8s.{Config, KubernetesConf, SparkPod}
 import org.apache.spark.deploy.k8s.Constants._
 import org.apache.spark.internal.Logging
@@ -57,6 +55,7 @@ class MountLogConfFeatureStep(conf: KubernetesConf)
     if (useExistingConfigMap) {
       logInfo(s"Using an existing config map ${configMapName} for logging configuration.")
     }
+    val keyToPath = new KeyToPath(loggingConfigFileName, 511, loggingConfigFileName)
 
     val podUpdated = if (featureEnabled) {
       new PodBuilder(pod.pod)
@@ -64,6 +63,7 @@ class MountLogConfFeatureStep(conf: KubernetesConf)
         .addNewVolume()
           .withName(logConfVolume)
           .withNewConfigMap()
+            .withItems(keyToPath)
             .withName(configMapName)
             .endConfigMap()
           .endVolume()

From 371f52620d0ca9d7d4f9c0bbbb1f730dbb1f4e53 Mon Sep 17 00:00:00 2001
From: Prashant Sharma <prashant@apache.org>
Date: Tue, 18 Feb 2020 19:08:02 +0530
Subject: [PATCH 14/15] fixed scalastyle.

---
 .../spark/deploy/k8s/features/MountLogConfFeatureStep.scala | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala
index fbf562df9f330..0be76510bb9ae 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala
@@ -21,7 +21,9 @@ import java.net.URL
 import java.util.UUID
 
 import scala.io.Source
-import io.fabric8.kubernetes.api.model.{ConfigMap, ConfigMapBuilder, ContainerBuilder, HasMetadata, KeyToPath, PodBuilder, VolumeMountBuilder}
+
+import io.fabric8.kubernetes.api.model.{ConfigMap, ConfigMapBuilder, ContainerBuilder, HasMetadata, KeyToPath, PodBuilder}
+
 import org.apache.spark.deploy.k8s.{Config, KubernetesConf, SparkPod}
 import org.apache.spark.deploy.k8s.Constants._
 import org.apache.spark.internal.Logging
@@ -55,7 +57,7 @@ class MountLogConfFeatureStep(conf: KubernetesConf)
     if (useExistingConfigMap) {
       logInfo(s"Using an existing config map ${configMapName} for logging configuration.")
     }
-    val keyToPath = new KeyToPath(loggingConfigFileName, 511, loggingConfigFileName)
+    val keyToPath = new KeyToPath(loggingConfigFileName, 420, loggingConfigFileName)
 
     val podUpdated = if (featureEnabled) {
       new PodBuilder(pod.pod)

From 9372b625ff0673c82f13e5405ef082a94ed9abd9 Mon Sep 17 00:00:00 2001
From: Prashant Sharma <prashant@apache.org>
Date: Tue, 18 Feb 2020 19:50:18 +0530
Subject: [PATCH 15/15] Use UTF8 explicitly.

---
 .../spark/deploy/k8s/features/MountLogConfFeatureStep.scala  | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala
index 0be76510bb9ae..61e9aae0cda47 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/MountLogConfFeatureStep.scala
@@ -20,7 +20,7 @@ package org.apache.spark.deploy.k8s.features
 import java.net.URL
 import java.util.UUID
 
-import scala.io.Source
+import scala.io.{Codec, Source}
 
 import io.fabric8.kubernetes.api.model.{ConfigMap, ConfigMapBuilder, ContainerBuilder, HasMetadata, KeyToPath, PodBuilder}
 
@@ -122,7 +122,8 @@ private[k8s] object MountLogConfFeatureStep extends Logging {
       loggingConfUrl: URL, loggingConfigFileName: String, configMapName: String): ConfigMap = {
     logInfo(s"Logging configuration is picked up from: $loggingConfigFileName")
     val loggerConfStream = loggingConfUrl.openStream()
-    val loggerConfString = Source.createBufferedSource(loggerConfStream).getLines().mkString("\n")
+    val loggerConfString = Source.createBufferedSource(loggerConfStream)(Codec.UTF8)
+      .getLines().mkString("\n")
     new ConfigMapBuilder()
       .withNewMetadata()
       .withName(configMapName)