Conda improvements to logging safety and customization (apache-spark-on-k8s#185)

* Fix token leak when logging in CondaEnvironmentManager, and allow configuration of higher priority pkgs_dirs through the conf

* copy yarn integration test logdirs to CIRCLE_ARTIFACTS otherwise we have no idea why they failed

* oops fix envs_dirs, should be a list

* only copy yarn test artifacts if they exist

* argh syntax

* omg bash you killin me

Author: dansanduleac

circle.yml

@@ -63,6 +63,14 @@ test:
   post:
     - find . -name unit-tests.log -exec rsync -R {} $CIRCLE_ARTIFACTS \;:
         parallel: true
+    - ? |
+        shopt -s nullglob
+        files=(resource-managers/yarn/target/./org.apache.spark.deploy.yarn.*/*-logDir-*)
+        if [[ ${#files[@]} != 0 ]]; then
+          rsync -Rrm "${files[@]}" $CIRCLE_ARTIFACTS
+        fi
+      :
+        parallel: true
 
 deployment:
   release:

core/pom.xml

@@ -261,6 +261,10 @@
       <groupId>com.fasterxml.jackson.core</groupId>
       <artifactId>jackson-databind</artifactId>
     </dependency>
+    <dependency>
+      <groupId>com.fasterxml.jackson.dataformat</groupId>
+      <artifactId>jackson-dataformat-yaml</artifactId>
+    </dependency>
     <dependency>
       <groupId>com.fasterxml.jackson.module</groupId>
       <artifactId>jackson-module-scala_${scala.binary.version}</artifactId>

core/src/main/scala/org/apache/spark/api/conda/CondaEnvironment.scala

@@ -66,7 +66,8 @@ final class CondaEnvironment(val manager: CondaEnvironmentManager,
       List("install", "-n", envName, "-y", "--override-channels")
         ::: channels.iterator.flatMap(Iterator("--channel", _)).toList
         ::: "--" :: packages.toList,
-      description = s"install dependencies in conda env $condaEnvDir"
+      description = s"install dependencies in conda env $condaEnvDir",
+      channels = channels.toList
     )
 
     this.packages ++= packages

core/src/main/scala/org/apache/spark/api/conda/CondaEnvironmentManager.scala

@@ -25,19 +25,52 @@ import scala.sys.process.BasicIO
 import scala.sys.process.Process
 import scala.sys.process.ProcessBuilder
 import scala.sys.process.ProcessIO
+import scala.sys.process.ProcessLogger
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.dataformat.yaml.YAMLFactory
+import org.json4s.JsonAST.JValue
+import org.json4s.jackson.Json4sScalaModule
+import org.json4s.jackson.JsonMethods
 
 import org.apache.spark.SparkConf
 import org.apache.spark.SparkException
 import org.apache.spark.internal.Logging
 import org.apache.spark.internal.config.CONDA_BINARY_PATH
-import org.apache.spark.internal.config.CONDA_CHANNEL_URLS
+import org.apache.spark.internal.config.CONDA_GLOBAL_PACKAGE_DIRS
 import org.apache.spark.internal.config.CONDA_VERBOSITY
 import org.apache.spark.util.Utils
 
-final class CondaEnvironmentManager(condaBinaryPath: String, verbosity: Int = 0) extends Logging {
+final class CondaEnvironmentManager(condaBinaryPath: String,
+                                    verbosity: Int = 0,
+                                    packageDirs: Seq[String] = Nil) extends Logging {
 
   require(verbosity >= 0 && verbosity <= 3, "Verbosity must be between 0 and 3 inclusively")
 
+  lazy val defaultInfo: Map[String, JValue] = {
+    logInfo("Retrieving the conda installation's info")
+    val command = Process(List(condaBinaryPath, "info", "--json"), None)
+
+    val buffer = new StringBuffer
+    val io = BasicIO(withIn = false,
+      buffer,
+      Some(ProcessLogger(line => logDebug(s"<conda> $line"))))
+
+    val exitCode = command.run(io).exitValue()
+    if (exitCode != 0) {
+      throw new SparkException(s"Attempt to retrieve initial conda info exited with code: "
+        + f"$exitCode%nCommand was: $command%nOutput was:%n${buffer.toString}")
+    }
+
+    implicit val format = org.json4s.DefaultFormats
+    JsonMethods.parse(buffer.toString).extract[Map[String, JValue]]
+  }
+
+  lazy val defaultPkgsDirs: List[String] = {
+    implicit val format = org.json4s.DefaultFormats
+    defaultInfo("pkgs_dirs").extract[List[String]]
+  }
+
   def create(
               baseDir: String,
               condaPackages: Seq[String],
@@ -58,11 +91,11 @@ final class CondaEnvironmentManager(condaBinaryPath: String, verbosity: Int = 0)
     // Attempt to create environment
     runCondaProcess(
       linkedBaseDir,
-      List("create", "-n", name, "-y", "--override-channels", "--no-default-packages")
+      List("create", "-n", name, "-y", "--no-default-packages")
         ::: verbosityFlags
-        ::: condaChannelUrls.flatMap(Iterator("--channel", _)).toList
         ::: "--" :: condaPackages.toList,
-      description = "create conda env"
+      description = "create conda env",
+      channels = condaChannelUrls.toList
     )
 
     new CondaEnvironment(this, linkedBaseDir, name, condaPackages, condaChannelUrls)
@@ -77,28 +110,37 @@ final class CondaEnvironmentManager(condaBinaryPath: String, verbosity: Int = 0)
    *
    * This hack is necessary otherwise conda tries to use the homedir for pkgs cache.
    */
-  private[this] def generateCondarc(baseRoot: Path): Path = {
-    val condaPkgsPath = Paths.get(condaBinaryPath).getParent.getParent.resolve("pkgs")
+  private[this] def generateCondarc(baseRoot: Path, channelUrls: Seq[String]): Path = {
+
     val condarc = baseRoot.resolve("condarc")
-    val condarcContents =
-      s"""pkgs_dirs:
-         | - $baseRoot/pkgs
-         | - $condaPkgsPath
-         |envs_dirs:
-         | - $baseRoot/envs
-         |show_channel_urls: false
-         |channels: []
-         |default_channels: []
-      """.stripMargin
-    Files.write(condarc, List(condarcContents).asJava)
-    logInfo(f"Using condarc at $condarc:%n$condarcContents")
+
+    import org.json4s.JsonAST._
+    import org.json4s.JsonDSL._
+
+    // building it in json4s AST since it gives us more control over how it will be mapped
+    val condarcNode = JObject(
+      "pkgs_dirs" -> (packageDirs ++: s"$baseRoot/pkgs" +: defaultPkgsDirs),
+      "envs_dirs" -> List(s"$baseRoot/envs"),
+      "show_channel_urls" -> false,
+      "default_channels" -> JArray(Nil),
+      "channels" -> channelUrls
+    )
+    val mapper = new ObjectMapper(new YAMLFactory()).registerModule(Json4sScalaModule)
+
+    Files.write(condarc, mapper.writeValueAsBytes(condarcNode))
+
+    val sanitizedCondarc = condarcNode removeField { case (name, _) => name == "channels" }
+    logInfo(f"Using condarc at $condarc (channels have been edited out):%n"
+        + mapper.writeValueAsString(sanitizedCondarc))
+
     condarc
   }
 
   private[conda] def runCondaProcess(baseRoot: Path,
                                      args: List[String],
+                                     channels: List[String],
                                      description: String): Unit = {
-    val condarc = generateCondarc(baseRoot)
+    val condarc = generateCondarc(baseRoot, channels)
     val fakeHomeDir = baseRoot.resolve("home")
     // Attempt to create fake home dir
     Files.createDirectories(fakeHomeDir)
@@ -142,6 +184,7 @@ object CondaEnvironmentManager {
     val condaBinaryPath = sparkConf.get(CONDA_BINARY_PATH).getOrElse(
       sys.error(s"Expected config ${CONDA_BINARY_PATH.key} to be set"))
     val verbosity = sparkConf.get(CONDA_VERBOSITY)
-    new CondaEnvironmentManager(condaBinaryPath, verbosity)
+    val packageDirs = sparkConf.get(CONDA_GLOBAL_PACKAGE_DIRS)
+    new CondaEnvironmentManager(condaBinaryPath, verbosity, packageDirs)
   }
 }

core/src/main/scala/org/apache/spark/internal/config/package.scala

@@ -193,6 +193,13 @@ package object config {
     .intConf
     .createWithDefault(0)
 
+  private[spark] val CONDA_GLOBAL_PACKAGE_DIRS = ConfigBuilder("spark.conda.packageDirs")
+    .doc("Custom pkgs_dirs that should be prepended to the default set of pkgs_dirs that comes " +
+      "with conda")
+    .stringConf
+    .toSequence
+    .createWithDefault(Nil)
+
   // To limit memory usage, we only track information for a fixed number of tasks
   private[spark] val UI_RETAINED_TASKS = ConfigBuilder("spark.ui.retainedTasks")
     .intConf