Address incremental load on constructing EventFilterBuilder

Author: HeartSaVioR

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

@@ -21,13 +21,15 @@ import java.io.IOException
 import java.net.URI
 import java.util.ServiceLoader
 
+import scala.annotation.tailrec
 import scala.collection.JavaConverters._
 
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.{FileStatus, FileSystem, Path}
 
 import org.apache.spark.SparkConf
 import org.apache.spark.deploy.history.EventFilter.FilterStatistic
+import org.apache.spark.deploy.history.EventFilterBuildersLoader.LowerIndexLoadRequested
 import org.apache.spark.internal.Logging
 import org.apache.spark.internal.config.{EVENT_LOG_COMPACTION_SCORE_THRESHOLD, EVENT_LOG_ROLLING_MAX_FILES_TO_RETAIN}
 import org.apache.spark.scheduler.ReplayListenerBus
@@ -46,10 +48,6 @@ import org.apache.spark.util.Utils
  * represents approximate rate of filtered-out events. Score is being calculated via applying
  * heuristic; task events tend to take most size in event log.
  *
- * This class assumes caller will provide the sorted list of files which are sorted by the index of
- * event log file, with "at most" one compact file placed first if it exists. Caller should keep in
- * mind that this class doesn't care about the semantic of ordering.
- *
  * When compacting the files, the range of compaction for given file list is determined as:
  * (first ~ the file where there're `maxFilesToRetain` files on the right side)
  *
@@ -59,22 +57,43 @@ class EventLogFileCompactor(
     sparkConf: SparkConf,
     hadoopConf: Configuration,
     fs: FileSystem) extends Logging {
+  import EventFilterBuildersLoader._
+
   private val maxFilesToRetain: Int = sparkConf.get(EVENT_LOG_ROLLING_MAX_FILES_TO_RETAIN)
   private val compactionThresholdScore: Double = sparkConf.get(EVENT_LOG_COMPACTION_SCORE_THRESHOLD)
 
-  def compact(eventLogFiles: Seq[FileStatus]): (CompactionResult.Value, Option[Long]) = {
-    assertPrecondition(eventLogFiles)
+  private var filterBuildersLoader = new EventFilterBuildersLoader(fs)
+  private var loadedLogPath: Path = _
+
+  def compact(reader: EventLogFileReader): (CompactionResult.Value, Option[Long]) = {
+    doCompact(reader)
+  }
+
+  @tailrec
+  private def doCompact(reader: EventLogFileReader): (CompactionResult.Value, Option[Long]) = {
+    if (loadedLogPath == null) {
+      loadedLogPath = reader.rootPath
+    } else {
+      require(loadedLogPath == null || reader.rootPath == loadedLogPath,
+        "An instance of compactor should deal with same path of event log.")
+    }
+
+    if (reader.lastIndex.isEmpty) {
+      return (CompactionResult.NOT_ENOUGH_FILES, None)
+    }
 
+    val eventLogFiles = reader.listEventLogFiles
     if (eventLogFiles.length < maxFilesToRetain) {
       return (CompactionResult.NOT_ENOUGH_FILES, None)
     }
 
     val filesToCompact = findFilesToCompact(eventLogFiles)
     if (filesToCompact.isEmpty) {
-      (CompactionResult.NOT_ENOUGH_FILES, None)
-    } else {
-      val builders = initializeBuilders(fs, filesToCompact.map(_.getPath))
+      return (CompactionResult.NOT_ENOUGH_FILES, None)
+    }
 
+    try {
+      val builders = filterBuildersLoader.loadNewFiles(filesToCompact)
       val filters = builders.map(_.createFilter())
       val minScore = filters.flatMap(_.statistic()).map(calculateScore).min
 
@@ -87,37 +106,14 @@ class EventLogFileCompactor(
         (CompactionResult.SUCCESS, Some(RollingEventLogFilesWriter.getEventLogFileIndex(
           filesToCompact.last.getPath.getName)))
       }
+    } catch {
+      case _: LowerIndexLoadRequested =>
+        // reset loader and load again
+        filterBuildersLoader = new EventFilterBuildersLoader(fs)
+        doCompact(reader)
     }
   }
 
-  private def assertPrecondition(eventLogFiles: Seq[FileStatus]): Unit = {
-    val idxCompactedFiles = eventLogFiles.zipWithIndex.filter { case (file, _) =>
-      EventLogFileWriter.isCompacted(file.getPath)
-    }
-    require(idxCompactedFiles.size < 2 && idxCompactedFiles.headOption.forall(_._2 == 0),
-      "The number of compact files should be at most 1, and should be placed first if exists.")
-  }
-
-  /**
-   * Loads all available EventFilterBuilders in classloader via ServiceLoader, and initializes
-   * them via replaying events in given files.
-   */
-  private def initializeBuilders(fs: FileSystem, files: Seq[Path]): Seq[EventFilterBuilder] = {
-    val bus = new ReplayListenerBus()
-
-    val builders = ServiceLoader.load(classOf[EventFilterBuilder],
-      Utils.getContextOrSparkClassLoader).asScala.toSeq
-    builders.foreach(bus.addListener)
-
-    files.foreach { log =>
-      Utils.tryWithResource(EventLogFileReader.openEventLog(log, fs)) { in =>
-        bus.replay(in, log.getName)
-      }
-    }
-
-    builders
-  }
-
   private def calculateScore(stats: FilterStatistic): Double = {
     // For now it's simply measuring how many task events will be filtered out (rejected)
     // but it can be sophisticated later once we get more heuristic information and found
@@ -162,6 +158,68 @@ object CompactionResult extends Enumeration {
   val SUCCESS, NOT_ENOUGH_FILES, LOW_SCORE_FOR_COMPACTION = Value
 }
 
+class EventFilterBuildersLoader(fs: FileSystem) {
+  // the implementation of this bus is expected to be stateless
+  private val bus = new ReplayListenerBus()
+
+  /** Loads all available EventFilterBuilders in classloader via ServiceLoader */
+  private val filterBuilders: Seq[EventFilterBuilder] = ServiceLoader.load(
+    classOf[EventFilterBuilder], Utils.getContextOrSparkClassLoader).asScala.toSeq
+
+  filterBuilders.foreach(bus.addListener)
+
+  private var latestIndexLoaded: Long = -1L
+
+  /** only exposed for testing; simple metric to help testing */
+  private[history] var numFilesToLoad: Long = 0L
+
+  /**
+   * Initializes EventFilterBuilders via replaying events in given files. Loading files are done
+   * incrementally, via dropping indices which are already loaded and replaying remaining files.
+   * For example, If the last index of requested files is same as the last index being loaded,
+   * this will not replay any files.
+   *
+   * If the last index of requested files is smaller than the last index being loaded, it will
+   * throw [[LowerIndexLoadRequested]], which caller can decide whether ignoring it or
+   * invalidating loader and retrying.
+   */
+  def loadNewFiles(eventLogFiles: Seq[FileStatus]): Seq[EventFilterBuilder] = {
+    require(eventLogFiles.nonEmpty)
+
+    val idxToStatuses = eventLogFiles.map { status =>
+      val idx = RollingEventLogFilesWriter.getEventLogFileIndex(status.getPath.getName)
+      idx -> status
+    }
+
+    val newLatestIdx = idxToStatuses.last._1
+    if (newLatestIdx < latestIndexLoaded) {
+      throw new LowerIndexLoadRequested("Loader already loads higher index of event log than" +
+        " requested.")
+    }
+
+    val filesToLoad = idxToStatuses
+      .filter { case (idx, _) => idx > latestIndexLoaded }
+      .map { case (_, status) => status.getPath }
+
+    if (filesToLoad.nonEmpty) {
+      filesToLoad.foreach { log =>
+        Utils.tryWithResource(EventLogFileReader.openEventLog(log, fs)) { in =>
+          bus.replay(in, log.getName)
+        }
+        numFilesToLoad += 1
+      }
+
+      latestIndexLoaded = newLatestIdx
+    }
+
+    filterBuilders
+  }
+}
+
+object EventFilterBuildersLoader {
+  class LowerIndexLoadRequested(_msg: String) extends Exception(_msg)
+}
+
 /**
  * This class rewrites the event log files into one compact file: the compact file will only
  * contain the events which pass the filters. Events will be dropped only when all filters

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

@@ -159,7 +159,8 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
     new HistoryServerDiskManager(conf, path, listing, clock)
   }
 
-  private val fileCompactor = new EventLogFileCompactor(conf, hadoopConf, fs)
+  // Visible for testing.
+  private[history] val logToCompactor = new mutable.HashMap[String, EventLogFileCompactor]
 
   // Used to store the paths, which are being processed. This enable the replay log tasks execute
   // asynchronously and make sure that checkForLogs would not process a path repeatedly.
@@ -526,20 +527,26 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
               reader.fileSizeForLastIndex > 0
           }
         }
-        .sortWith { case (entry1, entry2) =>
-          entry1.modificationTime > entry2.modificationTime
+        .sortWith { case (reader1, reader2) =>
+          reader1.modificationTime > reader2.modificationTime
         }
 
       if (updated.nonEmpty) {
         logDebug(s"New/updated attempts found: ${updated.size} ${updated.map(_.rootPath)}")
       }
 
-      updated.foreach { entry =>
-        processing(entry.rootPath)
+      updated.foreach { reader =>
+        processing(reader.rootPath)
         try {
           val task: Runnable = () => {
-            val updatedLastCompactionIndex = compact(entry)
-            mergeApplicationListing(entry, newLastScanTime, true, updatedLastCompactionIndex)
+            val (shouldRenewReader, updatedLastCompactionIndex) = compact(reader)
+            // we should renew reader if the list of event log files are changed in `compact`
+            val newReader = if (shouldRenewReader) {
+              EventLogFileReader(fs, reader.rootPath).get
+            } else {
+              reader
+            }
+            mergeApplicationListing(newReader, newLastScanTime, true, updatedLastCompactionIndex)
           }
           replayExecutor.submit(task)
         } catch {
@@ -566,6 +573,7 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
         log.appId.foreach { appId =>
           cleanAppData(appId, log.attemptId, log.logPath)
           listing.delete(classOf[LogInfo], log.logPath)
+          cleanupCompactor(log.logPath)
         }
       }
 
@@ -576,26 +584,33 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
   }
 
   /** exposed for testing */
-  private[history] def compact(reader: EventLogFileReader): Option[Long] = {
+  private[history] def compact(reader: EventLogFileReader): (Boolean, Option[Long]) = {
     reader.lastIndex match {
       case Some(lastIndex) =>
         try {
-          val info = listing.read(classOf[LogInfo], reader.rootPath.toString)
+          val rootPath = reader.rootPath.toString
+          val info = listing.read(classOf[LogInfo], rootPath)
           if (info.lastCompactionIndex.isEmpty || info.lastCompactionIndex.get < lastIndex) {
             // haven't tried compaction for this index, do compaction
-            val (_, lastCompactionIndex) = fileCompactor.compact(reader.listEventLogFiles)
-            listing.write(info.copy(lastCompactionIndex = lastCompactionIndex))
-            Some(lastIndex)
+            val compactor = logToCompactor.getOrElseUpdate(rootPath,
+              new EventLogFileCompactor(conf, hadoopConf, fs))
+            val (compactionResult, lastCompactionIndex) = compactor.compact(reader)
+            if (compactionResult == CompactionResult.SUCCESS) {
+              listing.write(info.copy(lastCompactionIndex = lastCompactionIndex))
+              (true, Some(lastIndex))
+            } else {
+              (false, Some(lastIndex))
+            }
           } else {
-            info.lastCompactionIndex
+            (false, info.lastCompactionIndex)
           }
         } catch {
           case _: NoSuchElementException =>
             // this should exist, but ignoring doesn't hurt much
-            None
+            (false, None)
         }
 
-      case None => None // This is not applied to single event log file.
+      case None => (false, None) // This is not applied to single event log file.
     }
   }
 
@@ -702,12 +717,14 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
       case e: InterruptedException =>
         throw e
       case e: AccessControlException =>
+        val rootPath = reader.rootPath
         // We don't have read permissions on the log file
-        logWarning(s"Unable to read log ${reader.rootPath}", e)
-        blacklist(reader.rootPath)
+        logWarning(s"Unable to read log ${rootPath}", e)
+        blacklist(rootPath)
         // SPARK-28157 We should remove this blacklisted entry from the KVStore
         // to handle permission-only changes with the same file sizes later.
-        listing.delete(classOf[LogInfo], reader.rootPath.toString)
+        listing.delete(classOf[LogInfo], rootPath.toString)
+        cleanupCompactor(rootPath.toString)
       case e: Exception =>
         logError("Exception while merging application listings", e)
     } finally {
@@ -864,6 +881,7 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
       logInfo(s"Deleting invalid / corrupt event log ${log.logPath}")
       deleteLog(fs, new Path(log.logPath))
       listing.delete(classOf[LogInfo], log.logPath)
+      cleanupCompactor(log.logPath)
     }
 
     log.appId.foreach { appId =>
@@ -911,6 +929,7 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
         logInfo(s"Deleting invalid / corrupt event log ${log.logPath}")
         deleteLog(fs, new Path(log.logPath))
         listing.delete(classOf[LogInfo], log.logPath)
+        cleanupCompactor(log.logPath)
       }
     }
 
@@ -953,6 +972,7 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
       logInfo(s"Deleting expired event log for ${attempt.logPath}")
       val logPath = new Path(logDir, attempt.logPath)
       listing.delete(classOf[LogInfo], logPath.toString())
+      cleanupCompactor(logPath.toString)
       cleanAppData(app.id, attempt.info.attemptId, logPath.toString())
       if (deleteLog(fs, logPath)) {
         countDeleted += 1
@@ -1240,6 +1260,10 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
     }
     deleted
   }
+
+  private def cleanupCompactor(logPath: String): Unit = {
+    logToCompactor -= logPath
+  }
 }
 
 private[history] object FsHistoryProvider {