Move spark.executor.ExecutorLogUrlHandler to spark.LogUrlHandler

Author: EnricoMi

core/src/main/scala/org/apache/spark/executor/ExecutorLogUrlHandler.scala renamed to core/src/main/scala/org/apache/spark/LogUrlHandler.scala

@@ -15,17 +15,17 @@
  * limitations under the License.
  */
 
-package org.apache.spark.executor
+package org.apache.spark
 
 import java.util.concurrent.atomic.AtomicBoolean
 
 import scala.util.matching.Regex
 
-import org.apache.spark.internal.{Logging, MDC}
-import org.apache.spark.internal.LogKeys
+import org.apache.spark.internal.{Logging, LogKeys, MDC}
 
-private[spark] class ExecutorLogUrlHandler(logUrlPattern: Option[String]) extends Logging {
-  import ExecutorLogUrlHandler._
+
+private[spark] class LogUrlHandler(logUrlPattern: Option[String]) extends Logging {
+  import LogUrlHandler._
 
   private val informedForMissingAttributes = new AtomicBoolean(false)
 
@@ -83,14 +83,14 @@ private[spark] class ExecutorLogUrlHandler(logUrlPattern: Option[String]) extend
       allPatterns: Set[String],
       allAttributes: Set[String]): Unit = {
     if (informedForMissingAttributes.compareAndSet(false, true)) {
-      logInfo(log"Fail to renew executor log urls: ${MDC(LogKeys.REASON, reason)}." +
+      logInfo(log"Fail to renew log urls: ${MDC(LogKeys.REASON, reason)}." +
         log" Required: ${MDC(LogKeys.REGEX, allPatterns)} / " +
         log"available: ${MDC(LogKeys.ATTRIBUTE_MAP, allAttributes)}." +
         log" Falling back to show app's original log urls.")
     }
   }
 }
 
-private[spark] object ExecutorLogUrlHandler {
+private[spark] object LogUrlHandler {
   val CUSTOM_URL_PATTERN_REGEX: Regex = "\\{\\{([A-Za-z0-9_\\-]+)\\}\\}".r
 }

core/src/main/scala/org/apache/spark/deploy/history/HistoryAppStatusStore.scala

@@ -17,8 +17,7 @@
 
 package org.apache.spark.deploy.history
 
-import org.apache.spark.SparkConf
-import org.apache.spark.executor.ExecutorLogUrlHandler
+import org.apache.spark.{LogUrlHandler, SparkConf}
 import org.apache.spark.internal.Logging
 import org.apache.spark.internal.config.History._
 import org.apache.spark.status.AppStatusStore
@@ -40,7 +39,7 @@ private[spark] class HistoryAppStatusStore(
     }
   }
 
-  private val logUrlHandler = new ExecutorLogUrlHandler(logUrlPattern)
+  private val logUrlHandler = new LogUrlHandler(logUrlPattern)
 
   override def executorList(activeOnly: Boolean): Seq[v1.ExecutorSummary] = {
     val execList = super.executorList(activeOnly)

core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala

@@ -27,11 +27,10 @@ import scala.concurrent.Future
 import com.google.common.cache.CacheBuilder
 import org.apache.hadoop.security.UserGroupInformation
 
-import org.apache.spark.{ExecutorAllocationClient, SparkEnv, TaskState}
+import org.apache.spark.{ExecutorAllocationClient, LogUrlHandler, SparkEnv, TaskState}
 import org.apache.spark.deploy.SparkHadoopUtil
 import org.apache.spark.deploy.security.HadoopDelegationTokenManager
 import org.apache.spark.errors.SparkCoreErrors
-import org.apache.spark.executor.ExecutorLogUrlHandler
 import org.apache.spark.internal.{config, Logging, MDC}
 import org.apache.spark.internal.LogKeys
 import org.apache.spark.internal.LogKeys._
@@ -156,7 +155,7 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
       .filter { case (k, _) => k.startsWith("spark.") }
       .toImmutableArraySeq
 
-    private val logUrlHandler: ExecutorLogUrlHandler = new ExecutorLogUrlHandler(
+    private val logUrlHandler: LogUrlHandler = new LogUrlHandler(
       conf.get(UI.CUSTOM_EXECUTOR_LOG_URL))
 
     override def onStart(): Unit = {

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

@@ -27,7 +27,7 @@ import io.fabric8.kubernetes.api.model.Pod
 import io.fabric8.kubernetes.api.model.PodBuilder
 import io.fabric8.kubernetes.client.KubernetesClient
 
-import org.apache.spark.SparkContext
+import org.apache.spark.{LogUrlHandler, SparkContext}
 import org.apache.spark.deploy.k8s.{KubernetesConf, KubernetesUtils}
 import org.apache.spark.deploy.k8s.Config._
 import org.apache.spark.deploy.k8s.Constants._
@@ -115,7 +115,7 @@ private[spark] class KubernetesClusterSchedulerBackend(
     Some(Map("LOG_FILES" -> "log") ++ extractAttributes)
 
   override def getDriverLogUrls: Option[Map[String, String]] = {
-    val logUrlHandler = new ExecutorLogUrlHandler(conf.get(UI.CUSTOM_DRIVER_LOG_URL))
+    val logUrlHandler = new LogUrlHandler(conf.get(UI.CUSTOM_DRIVER_LOG_URL))
     getDriverAttributes.map(attr => logUrlHandler.applyPattern(Map.empty, attr)).filter(_.nonEmpty)
   }