WIP [SPARK-29779][CORE] Compact old event log files and cleanup - no UTs yet

Author: HeartSaVioR

core/src/main/resources/META-INF/services/org.apache.spark.deploy.history.EventFilterBuilder

@@ -0,0 +1 @@
+org.apache.spark.deploy.history.BasicEventFilterBuilder

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

@@ -0,0 +1,146 @@
+/*
+ * 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.deploy.history
+
+import scala.collection.mutable
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.scheduler._
+
+class BasicEventFilterBuilder extends SparkListener with EventFilterBuilder {
+  val liveJobToStages = new mutable.HashMap[Int, Seq[Int]]
+  val stageToTasks = new mutable.HashMap[Int, mutable.Set[Long]]
+
+  override def onJobStart(jobStart: SparkListenerJobStart): Unit = {
+    liveJobToStages += jobStart.jobId -> jobStart.stageIds
+  }
+
+  override def onJobEnd(jobEnd: SparkListenerJobEnd): Unit = {
+    val stages = liveJobToStages.getOrElse(jobEnd.jobId, Seq.empty[Int])
+    liveJobToStages -= jobEnd.jobId
+    stages.foreach { stage => stageToTasks -= stage }
+  }
+
+  override def onTaskStart(taskStart: SparkListenerTaskStart): Unit = {
+    val curTasks = stageToTasks.getOrElseUpdate(taskStart.stageId,
+      mutable.HashSet[Long]())
+    curTasks += taskStart.taskInfo.taskId
+  }
+
+  override def createFilter(): EventFilter = new BasicEventFilter(this)
+}
+
+class BasicEventFilter(trackListener: BasicEventFilterBuilder) extends EventFilter with Logging {
+
+  private val liveTasks: Set[Long] = trackListener.stageToTasks.values match {
+    case xs if xs.isEmpty => Set.empty[Long]
+    case xs => xs.reduce(_ ++ _).toSet
+  }
+
+  if (log.isDebugEnabled) {
+    logDebug(s"live jobs : ${trackListener.liveJobToStages.keySet}")
+    logDebug(s"stages in jobs : ${trackListener.liveJobToStages.values.flatten}")
+    logDebug(s"stages : ${trackListener.stageToTasks.keySet}")
+    logDebug(s"tasks in stages : ${trackListener.stageToTasks.values.flatten}")
+  }
+
+  override def filterStageCompleted(event: SparkListenerStageCompleted): Option[Boolean] = {
+    Some(trackListener.stageToTasks.contains(event.stageInfo.stageId))
+  }
+
+  override def filterStageSubmitted(event: SparkListenerStageSubmitted): Option[Boolean] = {
+    Some(trackListener.stageToTasks.contains(event.stageInfo.stageId))
+  }
+
+  override def filterTaskStart(event: SparkListenerTaskStart): Option[Boolean] = {
+    Some(liveTasks.contains(event.taskInfo.taskId))
+  }
+
+  override def filterTaskGettingResult(event: SparkListenerTaskGettingResult): Option[Boolean] = {
+    Some(liveTasks.contains(event.taskInfo.taskId))
+  }
+
+  override def filterTaskEnd(event: SparkListenerTaskEnd): Option[Boolean] = {
+    Some(liveTasks.contains(event.taskInfo.taskId))
+  }
+
+  override def filterJobStart(event: SparkListenerJobStart): Option[Boolean] = {
+    Some(trackListener.liveJobToStages.contains(event.jobId))
+  }
+
+  override def filterJobEnd(event: SparkListenerJobEnd): Option[Boolean] = {
+    Some(trackListener.liveJobToStages.contains(event.jobId))
+  }
+
+  override def filterUnpersistRDD(event: SparkListenerUnpersistRDD): Option[Boolean] = {
+    // FIXME: need to track rdd ids?
+    None
+  }
+
+  override def filterExecutorMetricsUpdate(
+      event: SparkListenerExecutorMetricsUpdate): Option[Boolean] = {
+    // FIXME: need to track live executors?
+    None
+  }
+
+  override def filterStageExecutorMetrics(
+      event: SparkListenerStageExecutorMetrics): Option[Boolean] = {
+    // FIXME: need to track live executors?
+    // handle this regardless of liveness of stage - good to restore metrics
+    None
+  }
+
+  override def filterExecutorAdded(event: SparkListenerExecutorAdded): Option[Boolean] = {
+    // FIXME: need to track live executors?
+    None
+  }
+
+  override def filterExecutorRemoved(event: SparkListenerExecutorRemoved): Option[Boolean] = {
+    // FIXME: need to track live executors?
+    None
+  }
+
+  override def filterExecutorBlacklisted(
+      event: SparkListenerExecutorBlacklisted): Option[Boolean] = {
+    // FIXME: need to track live executors?
+    None
+  }
+
+  override def filterExecutorBlacklistedForStage(
+      event: SparkListenerExecutorBlacklistedForStage): Option[Boolean] = {
+    // FIXME: need to track live executors?
+    Some(trackListener.stageToTasks.contains(event.stageId))
+  }
+
+  override def filterNodeBlacklistedForStage(
+      event: SparkListenerNodeBlacklistedForStage): Option[Boolean] = {
+    // FIXME: need to track live executors?
+    Some(trackListener.stageToTasks.contains(event.stageId))
+  }
+
+  override def filterExecutorUnblacklisted(
+      event: SparkListenerExecutorUnblacklisted): Option[Boolean] = {
+    // FIXME: need to track live executors?
+    None
+  }
+
+  override def filterSpeculativeTaskSubmitted(
+      event: SparkListenerSpeculativeTaskSubmitted): Option[Boolean] = {
+    Some(trackListener.stageToTasks.contains(event.stageId))
+  }
+}

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

@@ -0,0 +1,82 @@
+/*
+ * 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.deploy.history
+
+import org.apache.spark.scheduler._
+
+trait EventFilterBuilder extends SparkListenerInterface {
+  def createFilter(): EventFilter
+}
+
+trait EventFilter {
+  def filterStageCompleted(event: SparkListenerStageCompleted): Option[Boolean] = None
+
+  def filterStageSubmitted(event: SparkListenerStageSubmitted): Option[Boolean] = None
+
+  def filterTaskStart(event: SparkListenerTaskStart): Option[Boolean] = None
+
+  def filterTaskGettingResult(event: SparkListenerTaskGettingResult): Option[Boolean] = None
+
+  def filterTaskEnd(event: SparkListenerTaskEnd): Option[Boolean] = None
+
+  def filterJobStart(event: SparkListenerJobStart): Option[Boolean] = None
+
+  def filterJobEnd(event: SparkListenerJobEnd): Option[Boolean] = None
+
+  def filterEnvironmentUpdate(event: SparkListenerEnvironmentUpdate): Option[Boolean] = None
+
+  def filterBlockManagerAdded(event: SparkListenerBlockManagerAdded): Option[Boolean] = None
+
+  def filterBlockManagerRemoved(event: SparkListenerBlockManagerRemoved): Option[Boolean] = None
+
+  def filterUnpersistRDD(event: SparkListenerUnpersistRDD): Option[Boolean] = None
+
+  def filterApplicationStart(event: SparkListenerApplicationStart): Option[Boolean] = None
+
+  def filterApplicationEnd(event: SparkListenerApplicationEnd): Option[Boolean] = None
+
+  def filterExecutorMetricsUpdate(event: SparkListenerExecutorMetricsUpdate): Option[Boolean] = None
+
+  def filterStageExecutorMetrics(event: SparkListenerStageExecutorMetrics): Option[Boolean] = None
+
+  def filterExecutorAdded(event: SparkListenerExecutorAdded): Option[Boolean] = None
+
+  def filterExecutorRemoved(event: SparkListenerExecutorRemoved): Option[Boolean] = None
+
+  def filterExecutorBlacklisted(event: SparkListenerExecutorBlacklisted): Option[Boolean] = None
+
+  def filterExecutorBlacklistedForStage(
+      event: SparkListenerExecutorBlacklistedForStage): Option[Boolean] = None
+
+  def filterNodeBlacklistedForStage(
+      event: SparkListenerNodeBlacklistedForStage): Option[Boolean] = None
+
+  def filterExecutorUnblacklisted(event: SparkListenerExecutorUnblacklisted): Option[Boolean] = None
+
+  def filterNodeBlacklisted(event: SparkListenerNodeBlacklisted): Option[Boolean] = None
+
+  def filterNodeUnblacklisted(event: SparkListenerNodeUnblacklisted): Option[Boolean] = None
+
+  def filterBlockUpdated(event: SparkListenerBlockUpdated): Option[Boolean] = None
+
+  def filterSpeculativeTaskSubmitted(
+      event: SparkListenerSpeculativeTaskSubmitted): Option[Boolean] = None
+
+  def filterOtherEvent(event: SparkListenerEvent): Option[Boolean] = None
+}
+