add safelogging

Author: yifeih

core/src/main/scala/org/apache/spark/internal/SafeLogging.scala

@@ -0,0 +1,79 @@
+/*
+ * 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.internal
+
+trait SafeLogging extends Logging {
+  // Method to get the logger name for this object
+  override protected def logName = {
+    // Ignore trailing $'s in the class names for Scala objects
+    s"com.palantir.${this.getClass.getName.stripSuffix("$")}"
+  }
+
+  protected def safeLogInfo(msg: => String) {
+    if (log.isInfoEnabled) log.info(msg)
+  }
+  protected def safeLogDebug(msg: => String) {
+    if (log.isDebugEnabled) log.debug(msg)
+  }
+
+  protected def safeLogTrace(msg: => String) {
+    if (log.isTraceEnabled) log.trace(msg)
+  }
+
+  protected def safeLogWarning(msg: => String) {
+    if (log.isWarnEnabled) log.warn(msg)
+  }
+
+  protected def safeLogError(msg: => String) {
+    if (log.isErrorEnabled) log.error(msg)
+  }
+
+
+  // You can only safelog to SafeLogging logs, everything else is unsupported
+  override protected def logInfo(msg: => String) {
+    if (log.isWarnEnabled) log.warn(msg)
+  }
+  override protected def logDebug(msg: => String) {
+    throw new UnsupportedOperationException
+  }
+  override protected def logTrace(msg: => String) {
+    throw new UnsupportedOperationException
+  }
+  override protected def logWarning(msg: => String) {
+    throw new UnsupportedOperationException
+  }
+  override protected def logError(msg: => String) {
+    throw new UnsupportedOperationException
+  }
+
+  override protected def logInfo(msg: => String, throwable: Throwable) {
+    throw new UnsupportedOperationException
+  }
+  override protected def logDebug(msg: => String, throwable: Throwable) {
+    throw new UnsupportedOperationException
+  }
+  override protected def logTrace(msg: => String, throwable: Throwable) {
+    throw new UnsupportedOperationException
+  }
+  override protected def logWarning(msg: => String, throwable: Throwable) {
+    throw new UnsupportedOperationException
+  }
+  override protected def logError(msg: => String, throwable: Throwable) {
+    throw new UnsupportedOperationException
+  }
+}

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

@@ -26,15 +26,15 @@ import org.apache.spark.{SparkConf, SparkException}
 import org.apache.spark.deploy.k8s.Config._
 import org.apache.spark.deploy.k8s.Constants._
 import org.apache.spark.deploy.k8s.KubernetesConf
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.SafeLogging
 import org.apache.spark.util.{Clock, Utils}
 
 private[spark] class ExecutorPodsAllocator(
     conf: SparkConf,
     executorBuilder: KubernetesExecutorBuilder,
     kubernetesClient: KubernetesClient,
     snapshotsStore: ExecutorPodsSnapshotsStore,
-    clock: Clock) extends Logging {
+    clock: Clock) extends SafeLogging {
 
   private val EXECUTOR_ID_COUNTER = new AtomicLong(0L)
 
@@ -82,7 +82,7 @@ private[spark] class ExecutorPodsAllocator(
     newlyCreatedExecutors.foreach { case (execId, timeCreated) =>
       val currentTime = clock.getTimeMillis()
       if (currentTime - timeCreated > podCreationTimeout) {
-        logWarning(s"Executor with id $execId was not detected in the Kubernetes" +
+        safeLogWarning(s"Executor with id $execId was not detected in the Kubernetes" +
           s" cluster after $podCreationTimeout milliseconds despite the fact that a" +
           " previous allocation attempt tried to create it. The executor may have been" +
           " deleted but the application missed the deletion event.")
@@ -94,7 +94,7 @@ private[spark] class ExecutorPodsAllocator(
         }
         newlyCreatedExecutors -= execId
       } else {
-        logDebug(s"Executor with id $execId was not found in the Kubernetes cluster since it" +
+        safeLogDebug(s"Executor with id $execId was not found in the Kubernetes cluster since it" +
           s" was created ${currentTime - timeCreated} milliseconds ago.")
       }
     }
@@ -112,15 +112,15 @@ private[spark] class ExecutorPodsAllocator(
         case _ => false
       }
       val currentTotalExpectedExecutors = totalExpectedExecutors.get
-      logDebug(s"Currently have $currentRunningExecutors running executors and" +
+      safeLogDebug(s"Currently have $currentRunningExecutors running executors and" +
         s" $currentPendingExecutors pending executors. $newlyCreatedExecutors executors" +
         s" have been requested but are pending appearance in the cluster.")
       if (newlyCreatedExecutors.isEmpty
         && currentPendingExecutors == 0
         && currentRunningExecutors < currentTotalExpectedExecutors) {
         val numExecutorsToAllocate = math.min(
           currentTotalExpectedExecutors - currentRunningExecutors, podAllocationSize)
-        logInfo(s"Going to request $numExecutorsToAllocate executors from Kubernetes.")
+        safeLogInfo(s"Going to request $numExecutorsToAllocate executors from Kubernetes.")
         for ( _ <- 0 until numExecutorsToAllocate) {
           val newExecutorId = EXECUTOR_ID_COUNTER.incrementAndGet()
           val executorConf = KubernetesConf.createExecutorConf(
@@ -136,14 +136,14 @@ private[spark] class ExecutorPodsAllocator(
             .build()
           kubernetesClient.pods().create(podWithAttachedContainer)
           newlyCreatedExecutors(newExecutorId) = clock.getTimeMillis()
-          logDebug(s"Requested executor with id $newExecutorId from Kubernetes.")
+          safeLogDebug(s"Requested executor with id $newExecutorId from Kubernetes.")
         }
       } else if (currentRunningExecutors >= currentTotalExpectedExecutors) {
         // TODO handle edge cases if we end up with more running executors than expected.
-        logDebug("Current number of running executors is equal to the number of requested" +
+        safeLogDebug("Current number of running executors is equal to the number of requested" +
           " executors. Not scaling up further.")
       } else if (newlyCreatedExecutors.nonEmpty || currentPendingExecutors != 0) {
-        logDebug(s"Still waiting for ${newlyCreatedExecutors.size + currentPendingExecutors}" +
+        safeLogDebug(s"Still waiting for ${newlyCreatedExecutors.size + currentPendingExecutors}" +
           s" executors to begin running before requesting for more executors. # of executors in" +
           s" pending status in the cluster: $currentPendingExecutors. # of executors that we have" +
           s" created but we have not observed as being present in the cluster yet:" +

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

@@ -25,7 +25,7 @@ import scala.collection.mutable
 import org.apache.spark.SparkConf
 import org.apache.spark.deploy.k8s.Config._
 import org.apache.spark.deploy.k8s.KubernetesUtils._
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.SafeLogging
 import org.apache.spark.scheduler.ExecutorExited
 import org.apache.spark.util.Utils
 
@@ -37,7 +37,7 @@ private[spark] class ExecutorPodsLifecycleManager(
     // job-breaking if we remove executors more than once but it's ideal if we make an attempt
     // to avoid doing so. Expire cache entries so that this data structure doesn't grow beyond
     // bounds.
-    removedExecutorsCache: Cache[java.lang.Long, java.lang.Long]) extends Logging {
+    removedExecutorsCache: Cache[java.lang.Long, java.lang.Long]) extends SafeLogging {
 
   import ExecutorPodsLifecycleManager._
 
@@ -57,16 +57,16 @@ private[spark] class ExecutorPodsLifecycleManager(
       snapshot.executorPods.foreach { case (execId, state) =>
         state match {
           case deleted@PodDeleted(_) =>
-            logDebug(s"Snapshot reported deleted executor with id $execId," +
+            safeLogDebug(s"Snapshot reported deleted executor with id $execId," +
               s" pod name ${state.pod.getMetadata.getName}")
             removeExecutorFromSpark(schedulerBackend, deleted, execId)
             execIdsRemovedInThisRound += execId
           case failed@PodFailed(_) =>
-            logDebug(s"Snapshot reported failed executor with id $execId," +
+            safeLogDebug(s"Snapshot reported failed executor with id $execId," +
               s" pod name ${state.pod.getMetadata.getName}")
             onFinalNonDeletedState(failed, execId, schedulerBackend, execIdsRemovedInThisRound)
           case succeeded@PodSucceeded(_) =>
-            logDebug(s"Snapshot reported succeeded executor with id $execId," +
+            safeLogDebug(s"Snapshot reported succeeded executor with id $execId," +
               s" pod name ${state.pod.getMetadata.getName}. Note that succeeded executors are" +
               s" unusual unless Spark specifically informed the executor to exit.")
             onFinalNonDeletedState(succeeded, execId, schedulerBackend, execIdsRemovedInThisRound)
@@ -89,7 +89,7 @@ private[spark] class ExecutorPodsLifecycleManager(
           val exitReasonMessage = s"The executor with ID $missingExecutorId was not found in the" +
             s" cluster but we didn't get a reason why. Marking the executor as failed. The" +
             s" executor may have been deleted but the driver missed the deletion event."
-          logDebug(exitReasonMessage)
+          safeLogDebug(exitReasonMessage)
           val exitReason = ExecutorExited(
             UNKNOWN_EXIT_CODE,
             exitCausedByApp = false,
@@ -101,7 +101,7 @@ private[spark] class ExecutorPodsLifecycleManager(
     }
 
     if (execIdsRemovedInThisRound.nonEmpty) {
-      logDebug(s"Removed executors with ids ${execIdsRemovedInThisRound.mkString(",")}" +
+      safeLogDebug(s"Removed executors with ids ${execIdsRemovedInThisRound.mkString(",")}" +
         s" from Spark that were either found to be deleted or non-existent in the cluster.")
     }
   }