Add SafeLogging to more classes (apache-spark-on-k8s#466)

Add SafeLogging (similar to palantir#425) to more classes that can be useful:
- k8s classes
- CoarseGrainedSchedulerBackend
- SparkContext
- MemoryStore
- Executor
- CoarseGrainedExecutorBackend
- TorrentBroadCast and Broadcast

Author: dvogelbacher

FORK.md

@@ -10,12 +10,15 @@
 * [SPARK-18079](https://issues.apache.org/jira/browse/SPARK-18079) - CollectLimitExec.executeToIterator should perform per-partition limits
 
 * [SPARK-15777](https://issues.apache.org/jira/browse/SPARK-15777) (Partial fix) - Catalog federation
- * make ExternalCatalog configurable beyond in memory and hive
- * FileIndex for catalog tables is provided by external catalog instead of using default impl
+    * make ExternalCatalog configurable beyond in memory and hive
+    * FileIndex for catalog tables is provided by external catalog instead of using default impl
 
 * Better pushdown for IN expressions in parquet via UserDefinedPredicate ([SPARK-17091](https://issues.apache.org/jira/browse/SPARK-17091) for original issue)
-
+* SafeLogging implemented for the following files:
+    * core: Broadcast, CoarseGrainedExecutorBackend, CoarseGrainedSchedulerBackend, Executor, MemoryStore, SparkContext, TorrentBroadcast
+    * kubernetes: ExecutorPodsAllocator, ExecutorPodsLifecycleManager, ExecutorPodsPollingSnapshotSource, ExecutorPodsSnapshot, ExecutorPodsWatchSnapshotSource, KubernetesClusterSchedulerBackend
+    * yarn: YarnClusterSchedulerBackend, YarnSchedulerBackend
 # Added
 
 * Gradle plugin to easily create custom docker images for use with k8s
-* Filter rLibDir by exists so that daemon.R references the correct file [460](https://github.com/palantir/spark/pull/460) 
+* Filter rLibDir by exists so that daemon.R references the correct file [460](https://github.com/palantir/spark/pull/460)

core/src/main/scala/org/apache/spark/SparkContext.scala

@@ -21,8 +21,10 @@ import java.io.Serializable
 
 import scala.reflect.ClassTag
 
+import com.palantir.logsafe.{SafeArg, UnsafeArg}
+
 import org.apache.spark.SparkException
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.SafeLogging
 import org.apache.spark.util.Utils
 
 /**
@@ -54,7 +56,7 @@ import org.apache.spark.util.Utils
  * @param id A unique identifier for the broadcast variable.
  * @tparam T Type of the data contained in the broadcast variable.
  */
-abstract class Broadcast[T: ClassTag](val id: Long) extends Serializable with Logging {
+abstract class Broadcast[T: ClassTag](val id: Long) extends Serializable with SafeLogging {
 
   /**
    * Flag signifying whether the broadcast variable is valid
@@ -107,7 +109,10 @@ abstract class Broadcast[T: ClassTag](val id: Long) extends Serializable with Lo
     assertValid()
     _isValid = false
     _destroySite = Utils.getCallSite().shortForm
-    logInfo("Destroying %s (from %s)".format(toString, _destroySite))
+    safeLogInfo("Destroying",
+      SafeArg.of("id", id),
+      UnsafeArg.of("stringRepr", toString),
+      UnsafeArg.of("destroySite", _destroySite))
     doDestroy(blocking)
   }
 

core/src/main/scala/org/apache/spark/broadcast/Broadcast.scala

@@ -25,8 +25,10 @@ import scala.collection.JavaConverters._
 import scala.reflect.ClassTag
 import scala.util.Random
 
+import com.palantir.logsafe.SafeArg
+
 import org.apache.spark._
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.SafeLogging
 import org.apache.spark.io.CompressionCodec
 import org.apache.spark.serializer.Serializer
 import org.apache.spark.storage._
@@ -55,7 +57,7 @@ import org.apache.spark.util.io.{ChunkedByteBuffer, ChunkedByteBufferOutputStrea
  * @param id A unique identifier for the broadcast variable.
  */
 private[spark] class TorrentBroadcast[T: ClassTag](obj: T, id: Long)
-  extends Broadcast[T](id) with Logging with Serializable {
+  extends Broadcast[T](id) with SafeLogging with Serializable {
 
   /**
    * Value of the broadcast object on executors. This is reconstructed by [[readBroadcastBlock]],
@@ -150,7 +152,9 @@ private[spark] class TorrentBroadcast[T: ClassTag](obj: T, id: Long)
 
     for (pid <- Random.shuffle(Seq.range(0, numBlocks))) {
       val pieceId = BroadcastBlockId(id, "piece" + pid)
-      logDebug(s"Reading piece $pieceId of $broadcastId")
+      safeLogDebug("Reading piece",
+        SafeArg.of("pieceId", pieceId),
+        SafeArg.of("broadcastId", broadcastId))
       // First try getLocalBytes because there is a chance that previous attempts to fetch the
       // broadcast blocks have already fetched some of the blocks. In that case, some blocks
       // would be available locally (on this executor).
@@ -226,10 +230,13 @@ private[spark] class TorrentBroadcast[T: ClassTag](obj: T, id: Long)
               throw new SparkException(s"Failed to get locally stored broadcast data: $broadcastId")
             }
           case None =>
-            logInfo("Started reading broadcast variable " + id)
+            safeLogInfo("Started reading broadcast variable",
+              SafeArg.of("id", id))
             val startTimeMs = System.currentTimeMillis()
             val blocks = readBlocks()
-            logInfo("Reading broadcast variable " + id + " took" + Utils.getUsedTimeMs(startTimeMs))
+            safeLogInfo("Reading broadcast variable finished",
+              SafeArg.of("id", id),
+              SafeArg.of("timeUsed", Utils.getUsedTimeMs(startTimeMs)))
 
             try {
               val obj = TorrentBroadcast.unBlockifyObject[T](
@@ -276,7 +283,7 @@ private[spark] class TorrentBroadcast[T: ClassTag](obj: T, id: Long)
 }
 
 
-private object TorrentBroadcast extends Logging {
+private object TorrentBroadcast extends SafeLogging {
 
   def blockifyObject[T: ClassTag](
       obj: T,
@@ -317,7 +324,7 @@ private object TorrentBroadcast extends Logging {
    * If removeFromDriver is true, also remove these persisted blocks on the driver.
    */
   def unpersist(id: Long, removeFromDriver: Boolean, blocking: Boolean): Unit = {
-    logDebug(s"Unpersisting TorrentBroadcast $id")
+    safeLogDebug("Unpersisting TorrentBroadcast", SafeArg.of("id", id))
     SparkEnv.get.blockManager.master.removeBroadcast(id, removeFromDriver, blocking)
   }
 }

core/src/main/scala/org/apache/spark/broadcast/TorrentBroadcast.scala

@@ -26,11 +26,13 @@ import scala.collection.mutable
 import scala.util.{Failure, Success}
 import scala.util.control.NonFatal
 
+import com.palantir.logsafe.{SafeArg, UnsafeArg}
+
 import org.apache.spark._
 import org.apache.spark.TaskState.TaskState
 import org.apache.spark.deploy.SparkHadoopUtil
 import org.apache.spark.deploy.worker.WorkerWatcher
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, SafeLogging}
 import org.apache.spark.rpc._
 import org.apache.spark.scheduler.{ExecutorLossReason, TaskDescription}
 import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages._
@@ -45,7 +47,7 @@ private[spark] class CoarseGrainedExecutorBackend(
     cores: Int,
     userClassPath: Seq[URL],
     env: SparkEnv)
-  extends ThreadSafeRpcEndpoint with ExecutorBackend with Logging {
+  extends ThreadSafeRpcEndpoint with ExecutorBackend with SafeLogging {
 
   private[this] val stopping = new AtomicBoolean(false)
   var executor: Executor = null
@@ -56,7 +58,7 @@ private[spark] class CoarseGrainedExecutorBackend(
   private[this] val ser: SerializerInstance = env.closureSerializer.newInstance()
 
   override def onStart() {
-    logInfo("Connecting to driver: " + driverUrl)
+    safeLogInfo("Connecting to driver", UnsafeArg.of("driverUrl", driverUrl))
     rpcEnv.asyncSetupEndpointRefByURI(driverUrl).flatMap { ref =>
       // This is a very fast action so we can use "ThreadUtils.sameThread"
       driver = Some(ref)
@@ -78,7 +80,7 @@ private[spark] class CoarseGrainedExecutorBackend(
 
   override def receive: PartialFunction[Any, Unit] = {
     case RegisteredExecutor =>
-      logInfo("Successfully registered with driver")
+      safeLogInfo("Successfully registered with driver")
       try {
         executor = new Executor(executorId, hostname, env, userClassPath, isLocal = false)
       } catch {
@@ -94,7 +96,7 @@ private[spark] class CoarseGrainedExecutorBackend(
         exitExecutor(1, "Received LaunchTask command but executor was null")
       } else {
         val taskDesc = TaskDescription.decode(data.value)
-        logInfo("Got assigned task " + taskDesc.taskId)
+        safeLogInfo("Got assigned task", SafeArg.of("taskId", taskDesc.taskId))
         executor.launchTask(this, taskDesc)
       }
 
@@ -107,7 +109,7 @@ private[spark] class CoarseGrainedExecutorBackend(
 
     case StopExecutor =>
       stopping.set(true)
-      logInfo("Driver commanded a shutdown")
+      safeLogInfo("Driver commanded a shutdown")
       // Cannot shutdown here because an ack may need to be sent back to the caller. So send
       // a message to self to actually do the shutdown.
       self.send(Shutdown)
@@ -125,26 +127,29 @@ private[spark] class CoarseGrainedExecutorBackend(
       }.start()
 
     case UpdateDelegationTokens(tokenBytes) =>
-      logInfo(s"Received tokens of ${tokenBytes.length} bytes")
+      safeLogInfo("Received tokens", UnsafeArg.of("tokenBytesLength", tokenBytes.length))
       SparkHadoopUtil.get.addDelegationTokens(tokenBytes, env.conf)
   }
 
   override def onDisconnected(remoteAddress: RpcAddress): Unit = {
     if (stopping.get()) {
-      logInfo(s"Driver from $remoteAddress disconnected during shutdown")
+      safeLogInfo("Driver disconnected during shutdown",
+        UnsafeArg.of("remoteAddress", remoteAddress))
     } else if (driver.exists(_.address == remoteAddress)) {
       exitExecutor(1, s"Driver $remoteAddress disassociated! Shutting down.", null,
         notifyDriver = false)
     } else {
-      logWarning(s"An unknown ($remoteAddress) driver disconnected.")
+      safeLogWarning("An unknown driver disconnected.",
+        UnsafeArg.of("remoteAddress", remoteAddress))
     }
   }
 
   override def statusUpdate(taskId: Long, state: TaskState, data: ByteBuffer) {
     val msg = StatusUpdate(executorId, taskId, state, data)
     driver match {
       case Some(driverRef) => driverRef.send(msg)
-      case None => logWarning(s"Drop $msg because has not yet connected to driver")
+      case None => safeLogWarning("Drop message because has not yet connected to driver",
+        UnsafeArg.of("msg", msg))
     }
   }
 
@@ -157,11 +162,11 @@ private[spark] class CoarseGrainedExecutorBackend(
                              reason: String,
                              throwable: Throwable = null,
                              notifyDriver: Boolean = true) = {
-    val message = "Executor self-exiting due to : " + reason
+    val message = "Executor self-exiting"
     if (throwable != null) {
-      logError(message, throwable)
+      safeLogError(message, throwable, UnsafeArg.of("reason", reason))
     } else {
-      logError(message)
+      safeLogError(message, UnsafeArg.of("reason", reason))
     }
 
     if (notifyDriver && driver.nonEmpty) {