use safelogging?

Author: yifeih

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

@@ -89,6 +89,12 @@
       <version>3.8.1</version>
     </dependency>
 
+    <dependency>
+      <groupId>com.palantir.safe-logging</groupId>
+      <artifactId>safe-logging</artifactId>
+      <version>1.5.1</version>
+    </dependency>
+
     <dependency>
       <groupId>org.mockito</groupId>
       <artifactId>mockito-core</artifactId>

resource-managers/kubernetes/core/pom.xml

@@ -0,0 +1,65 @@
+/*
+ * 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.logging
+
+import com.palantir.logsafe.Arg
+import org.slf4j.LoggerFactory
+
+trait SafeLogging {
+  private[this] val log_ = LoggerFactory.getLogger(this.getClass.getName)
+
+  def safeLogInfo(message: String, args: Arg[_]*): Unit = {
+    if (log_.isInfoEnabled) log_.info(message, args: _*)
+  }
+
+  def safeLogInfo(message: String, error: Throwable, args: Arg[_]*): Unit = {
+    if (log_.isInfoEnabled) log_.info(message, error, args: _*)
+  }
+
+  def safeLogDebug(message: String, args: Arg[_]*): Unit = {
+    if (log_.isDebugEnabled) log_.debug(message, args: _*)
+  }
+
+  def safeLogDebug(message: String, error: Throwable, args: Arg[_]*): Unit = {
+    if (log_.isDebugEnabled) log_.debug(message, error, args: _*)
+  }
+
+  def safeLogTrace(message: String, args: Arg[_]*): Unit = {
+    if (log_.isTraceEnabled) log_.trace(message, args: _*)
+  }
+
+  def safeLogTrace(message: String, error: Throwable, args: Arg[_]*): Unit = {
+    if (log_.isTraceEnabled) log_.trace(message, error, args: _*)
+  }
+
+  def safeLogWarning(message: String, args: Arg[_]*): Unit = {
+    if (log_.isWarnEnabled) log_.warn(message, args: _*)
+  }
+
+  def safeLogWarning(message: String, error: Throwable, args: Arg[_]*): Unit = {
+    if (log_.isWarnEnabled) log_.warn(message, error, args: _*)
+  }
+
+  def safeLogError(message: String, args: Arg[_]*): Unit = {
+    if (log_.isErrorEnabled) log_.error(message, args: _*)
+  }
+
+  def safeLogError(message: String, error: Throwable, args: Arg[_]*): Unit = {
+    if (log_.isErrorEnabled) log_.error(message, error, args: _*)
+  }
+}

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/logging/SafeLogging.scala

@@ -18,6 +18,7 @@ package org.apache.spark.scheduler.cluster.k8s
 
 import java.util.concurrent.atomic.{AtomicInteger, AtomicLong}
 
+import com.palantir.logsafe.SafeArg
 import io.fabric8.kubernetes.api.model.PodBuilder
 import io.fabric8.kubernetes.client.KubernetesClient
 import scala.collection.mutable
@@ -26,7 +27,7 @@ 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.SafeLogging
+import org.apache.spark.logging.SafeLogging
 import org.apache.spark.util.{Clock, Utils}
 
 private[spark] class ExecutorPodsAllocator(
@@ -82,10 +83,12 @@ private[spark] class ExecutorPodsAllocator(
     newlyCreatedExecutors.foreach { case (execId, timeCreated) =>
       val currentTime = clock.getTimeMillis()
       if (currentTime - timeCreated > podCreationTimeout) {
-        safeLogWarning(s"Executor with id $execId was not detected in the Kubernetes" +
-          s" cluster after $podCreationTimeout milliseconds despite the fact that a" +
+        safeLogWarning("Executor with id {execId} was not detected in the Kubernetes" +
+          " 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.")
+          " deleted but the application missed the deletion event.",
+          SafeArg.of("execId", execId),
+          SafeArg.of("podCreationTimeout", podCreationTimeout))
         Utils.tryLogNonFatalError {
           kubernetesClient
             .pods()
@@ -94,8 +97,10 @@ private[spark] class ExecutorPodsAllocator(
         }
         newlyCreatedExecutors -= execId
       } else {
-        safeLogDebug(s"Executor with id $execId was not found in the Kubernetes cluster since it" +
-          s" was created ${currentTime - timeCreated} milliseconds ago.")
+        safeLogDebug("Executor with id {execId} was not found in the Kubernetes cluster since it" +
+          " was created {timeSinceCreation} milliseconds ago.",
+          SafeArg.of("execId", execId),
+          SafeArg.of("timeSinceCreation", currentTime - timeCreated))
       }
     }
 
@@ -112,15 +117,19 @@ private[spark] class ExecutorPodsAllocator(
         case _ => false
       }
       val currentTotalExpectedExecutors = totalExpectedExecutors.get
-      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.")
+      safeLogDebug("Currently have {currentRunningExecutors} running executors and" +
+        " {currentPendingExecutors} pending executors. {newlyCreatedExecutors} executors" +
+        " have been requested but are pending appearance in the cluster.",
+        SafeArg.of("currentRunningExecutors", currentRunningExecutors),
+        SafeArg.of("currentPendingExecutors", currentPendingExecutors),
+        SafeArg.of("newlyCreatedExecutors", newlyCreatedExecutors))
       if (newlyCreatedExecutors.isEmpty
         && currentPendingExecutors == 0
         && currentRunningExecutors < currentTotalExpectedExecutors) {
         val numExecutorsToAllocate = math.min(
           currentTotalExpectedExecutors - currentRunningExecutors, podAllocationSize)
-        safeLogInfo(s"Going to request $numExecutorsToAllocate executors from Kubernetes.")
+        safeLogInfo("Going to request {numExecutorsToAllocate} executors from Kubernetes.",
+          SafeArg.of("numExecutorsToAllocate", numExecutorsToAllocate))
         for ( _ <- 0 until numExecutorsToAllocate) {
           val newExecutorId = EXECUTOR_ID_COUNTER.incrementAndGet()
           val executorConf = KubernetesConf.createExecutorConf(
@@ -136,18 +145,23 @@ private[spark] class ExecutorPodsAllocator(
             .build()
           kubernetesClient.pods().create(podWithAttachedContainer)
           newlyCreatedExecutors(newExecutorId) = clock.getTimeMillis()
-          safeLogDebug(s"Requested executor with id $newExecutorId from Kubernetes.")
+          safeLogDebug("Requested executor with id {newExecutorId} from Kubernetes.",
+            SafeArg.of("newExecutorId", newExecutorId))
         }
       } else if (currentRunningExecutors >= currentTotalExpectedExecutors) {
         // TODO handle edge cases if we end up with more running executors than expected.
         safeLogDebug("Current number of running executors is equal to the number of requested" +
           " executors. Not scaling up further.")
       } else if (newlyCreatedExecutors.nonEmpty || currentPendingExecutors != 0) {
-        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:" +
-          s" ${newlyCreatedExecutors.size}.")
+        safeLogDebug("Still waiting for {currentWaitingExecutors}" +
+          " executors to begin running before requesting for more executors. # of executors in" +
+          " pending status in the cluster: {currentPendingExecutors}. # of executors that we have" +
+          " created but we have not observed as being present in the cluster yet:" +
+          " ${newlyCreatedExecutors}.",
+          SafeArg.of("currentWaitingExecutors",
+            newlyCreatedExecutors.size + currentPendingExecutors),
+          SafeArg.of("currentPendingExecutors", currentPendingExecutors),
+          SafeArg.of("newlyCreatedExecutors", newlyCreatedExecutors.size))
       }
     }
   }

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

@@ -17,6 +17,7 @@
 package org.apache.spark.scheduler.cluster.k8s
 
 import com.google.common.cache.Cache
+import com.palantir.logsafe.SafeArg
 import io.fabric8.kubernetes.api.model.Pod
 import io.fabric8.kubernetes.client.KubernetesClient
 import scala.collection.JavaConverters._
@@ -25,7 +26,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.SafeLogging
+import org.apache.spark.logging.SafeLogging
 import org.apache.spark.scheduler.ExecutorExited
 import org.apache.spark.util.Utils
 
@@ -57,18 +58,24 @@ private[spark] class ExecutorPodsLifecycleManager(
       snapshot.executorPods.foreach { case (execId, state) =>
         state match {
           case deleted@PodDeleted(_) =>
-            safeLogDebug(s"Snapshot reported deleted executor with id $execId," +
-              s" pod name ${state.pod.getMetadata.getName}")
+            safeLogDebug(
+              "Snapshot reported deleted executor with id {execId} and pod name {podName}",
+              SafeArg.of("execId", execId),
+              SafeArg.of("podName", state.pod.getMetadata.getName))
             removeExecutorFromSpark(schedulerBackend, deleted, execId)
             execIdsRemovedInThisRound += execId
           case failed@PodFailed(_) =>
-            safeLogDebug(s"Snapshot reported failed executor with id $execId," +
-              s" pod name ${state.pod.getMetadata.getName}")
+            safeLogDebug(
+              "Snapshot reported failed executor with id {execId} and pod name {podName}",
+              SafeArg.of("execId", execId),
+              SafeArg.of("podName", state.pod.getMetadata.getName))
             onFinalNonDeletedState(failed, execId, schedulerBackend, execIdsRemovedInThisRound)
           case succeeded@PodSucceeded(_) =>
-            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.")
+            safeLogDebug(
+              "Snapshot reported succeeded executor with id {execId} and pod name {podName}." +
+              " Note that unusual unless Spark specifically informed the executor to exit.",
+              SafeArg.of("execId", execId),
+              SafeArg.of("podName", state.pod.getMetadata.getName))
             onFinalNonDeletedState(succeeded, execId, schedulerBackend, execIdsRemovedInThisRound)
           case _ =>
         }
@@ -89,7 +96,10 @@ 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."
-          safeLogDebug(exitReasonMessage)
+          safeLogDebug("The executor with ID {missingExecutorId} was not found in the" +
+            " cluster but we didn't get a reason why. Marking the executor as failed. The" +
+            " executor may have been deleted but the driver missed the deletion event.",
+            SafeArg.of("missingExecutorId", missingExecutorId))
           val exitReason = ExecutorExited(
             UNKNOWN_EXIT_CODE,
             exitCausedByApp = false,
@@ -101,8 +111,9 @@ private[spark] class ExecutorPodsLifecycleManager(
     }
 
     if (execIdsRemovedInThisRound.nonEmpty) {
-      safeLogDebug(s"Removed executors with ids ${execIdsRemovedInThisRound.mkString(",")}" +
-        s" from Spark that were either found to be deleted or non-existent in the cluster.")
+      safeLogDebug("Removed executors with ids {execIdsRemovedInThisRound}" +
+        " from Spark that were either found to be deleted or non-existent in the cluster.",
+        SafeArg.of("execIdsRemovedInThisRound", execIdsRemovedInThisRound.mkString(",")))
     }
   }