Github user steveloughran commented on a diff in the pull request:
https://github.com/apache/spark/pull/9571#discussion_r76952864
--- Diff:
core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala ---
@@ -664,6 +707,116 @@ private[history] class FsHistoryProvider(conf:
SparkConf, clock: Clock)
prevFileSize < latest.fileSize
}
}
+
+ /**
+ * Time a closure, returning its output.
+ * The timer is updated with the duration, and if a counter is supplied,
it's count
+ * is incremented by the duration.
+ * @param timer timer
+ * @param counter counter: an optional counter of the duration
+ * @param fn function
+ * @tparam T type of return value of time
+ * @return the result of the function.
+ */
+ private def time[T](timer: Timer, counter: Option[Counter] = None)(fn:
=> T): T = {
+ val timeCtx = timer.time()
+ try {
+ fn
+ } finally {
+ val duration = timeCtx.stop()
+ counter.foreach(_.inc(duration))
+ }
+ }
+}
+
+/**
+ * Metrics integration: the various counters of activity.
+ */
+private[history] class FsHistoryProviderMetrics(owner: FsHistoryProvider,
prefix: String)
+ extends HistoryMetricSource(prefix) {
+
+ /**
+ * Function to return an average; if the count is 0, so is the average.
+ * @param value value to average
+ * @param count event count to divide by
+ * @return the average, or 0 if the counter is itself 0
+ */
+ private def average(value: Long, count: Long): Long = {
+ if (count> 0) value / count else 0
+ }
+
+ override val sourceName = "history.fs"
+
+ private val name = MetricRegistry.name(sourceName)
+
+ /** Number of updates. */
+ val updateCount = new Counter()
+
+ /** Number of update failures. */
+ val updateFailureCount = new Counter()
+
+ /** Number of events replayed as listing merge. */
+ val historyEventCount = new Counter()
+
+ /** Timer of listing merges. */
+ val historyMergeTimer = new Timer()
+
+ /** Total time to merge all histories. */
+ val historyTotalMergeTime = new Counter()
+
+ /** Average time to process an event in the history merge operation. */
+ val historyEventMergeTime = new LambdaLongGauge(() =>
+ average(historyTotalMergeTime.getCount, historyEventCount.getCount))
+
+ /** Number of events replayed. */
+ val appUIEventCount = new Counter()
+
+ /** Update duration timer. */
+ val updateTimer = new Timer()
+
+ private val clock = new SystemClock
+
+ /** Time the last update was attempted. */
+ val updateLastAttempted = new TimestampGauge(clock)
+
+ /** Time the last update succeded. */
+ val updateLastSucceeded = new TimestampGauge(clock)
+
+ /** Number of App UI load operations. */
+ val appUILoadCount = new Counter()
+
+ /** Number of App UI load operations that failed due to a
load/parse/replay problem. */
+ val appUILoadFailureCount = new Counter()
+
+ /** Number of App UI load operations that failed due to an unknown file.
*/
+ val appUILoadNotFoundCount = new Counter()
+
+ /** Statistics on time to load app UIs. */
+ val appUiLoadTimer = new Timer()
+
+ /** Total load time of all App UIs. */
+ val appUITotalLoadTime = new Counter()
+
+ /** Average time to load a single event in the App UI */
+ val appUIEventReplayTime = new LambdaLongGauge(() =>
+ average(appUITotalLoadTime.getCount, appUIEventCount.getCount))
+
+ register(Seq(
+ ("history.merge.event.count", historyEventCount),
+ ("history.merge.event.time", historyEventMergeTime),
--- End diff --
No problem, though I think the total load time does need to be measured and
logged somewhere. I'll make sure that happens.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
