|
| 1 | +/* |
| 2 | + * Licensed to the Apache Software Foundation (ASF) under one or more |
| 3 | + * contributor license agreements. See the NOTICE file distributed with |
| 4 | + * this work for additional information regarding copyright ownership. |
| 5 | + * The ASF licenses this file to You under the Apache License, Version 2.0 |
| 6 | + * (the "License"); you may not use this file except in compliance with |
| 7 | + * the License. You may obtain a copy of the License at |
| 8 | + * |
| 9 | + * http://www.apache.org/licenses/LICENSE-2.0 |
| 10 | + * |
| 11 | + * Unless required by applicable law or agreed to in writing, software |
| 12 | + * distributed under the License is distributed on an "AS IS" BASIS, |
| 13 | + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 14 | + * See the License for the specific language governing permissions and |
| 15 | + * limitations under the License. |
| 16 | + */ |
| 17 | + |
| 18 | +package org.apache.spark.deploy.history |
| 19 | + |
| 20 | +import java.io.FileNotFoundException |
| 21 | +import java.util.concurrent.atomic.AtomicReference |
| 22 | + |
| 23 | +import scala.collection.mutable |
| 24 | + |
| 25 | +import org.apache.hadoop.fs.{FileStatus, Path} |
| 26 | + |
| 27 | +import org.apache.spark.{Logging, SecurityManager, SparkConf} |
| 28 | +import org.apache.spark.scheduler._ |
| 29 | +import org.apache.spark.ui.SparkUI |
| 30 | +import org.apache.spark.util.Utils |
| 31 | + |
| 32 | +class FsHistoryProvider(conf: SparkConf) extends ApplicationHistoryProvider |
| 33 | + with Logging { |
| 34 | + |
| 35 | + // Interval between each check for event log updates |
| 36 | + private val UPDATE_INTERVAL_MS = conf.getInt("spark.history.fs.updateInterval", 10) * 1000 |
| 37 | + |
| 38 | + private val logDir = conf.get("spark.history.fs.logDirectory") |
| 39 | + private val fs = Utils.getHadoopFileSystem(logDir) |
| 40 | + |
| 41 | + // A timestamp of when the disk was last accessed to check for log updates |
| 42 | + private var lastLogCheckTime = -1L |
| 43 | + |
| 44 | + // List of applications, in order from newest to oldest. |
| 45 | + private val appList = new AtomicReference[Seq[ApplicationHistoryInfo]](Nil) |
| 46 | + |
| 47 | + /** |
| 48 | + * A background thread that periodically checks for event log updates on disk. |
| 49 | + * |
| 50 | + * If a log check is invoked manually in the middle of a period, this thread re-adjusts the |
| 51 | + * time at which it performs the next log check to maintain the same period as before. |
| 52 | + * |
| 53 | + * TODO: Add a mechanism to update manually. |
| 54 | + */ |
| 55 | + private val logCheckingThread = new Thread("LogCheckingThread") { |
| 56 | + override def run() = Utils.logUncaughtExceptions { |
| 57 | + while (!stopped) { |
| 58 | + val now = System.currentTimeMillis |
| 59 | + if (now - lastLogCheckTime > UPDATE_INTERVAL_MS) { |
| 60 | + Thread.sleep(UPDATE_INTERVAL_MS) |
| 61 | + } else { |
| 62 | + // If the user has manually checked for logs recently, wait until |
| 63 | + // UPDATE_INTERVAL_MS after the last check time |
| 64 | + Thread.sleep(lastLogCheckTime + UPDATE_INTERVAL_MS - now) |
| 65 | + } |
| 66 | + checkForLogs() |
| 67 | + } |
| 68 | + } |
| 69 | + } |
| 70 | + |
| 71 | + @volatile private var stopped = false |
| 72 | + |
| 73 | + initialize() |
| 74 | + |
| 75 | + private def initialize() { |
| 76 | + // Validate the log directory. |
| 77 | + val path = new Path(logDir) |
| 78 | + if (!fs.exists(path)) { |
| 79 | + throw new IllegalArgumentException("Logging directory specified does not exist: %s".format(logDir)) |
| 80 | + } |
| 81 | + if (!fs.getFileStatus(path).isDir) { |
| 82 | + throw new IllegalArgumentException("Logging directory specified is not a directory: %s".format(logDir)) |
| 83 | + } |
| 84 | + |
| 85 | + checkForLogs() |
| 86 | + logCheckingThread.start() |
| 87 | + } |
| 88 | + |
| 89 | + override def stop() = { |
| 90 | + stopped = true |
| 91 | + logCheckingThread.interrupt() |
| 92 | + logCheckingThread.join() |
| 93 | + } |
| 94 | + |
| 95 | + override def getListing(offset: Int, limit: Int): Seq[ApplicationHistoryInfo] = { |
| 96 | + appList.get() |
| 97 | + } |
| 98 | + |
| 99 | + override def getAppInfo(appId: String): ApplicationHistoryInfo = { |
| 100 | + try { |
| 101 | + val appLogDir = fs.getFileStatus(new Path(logDir, appId)) |
| 102 | + loadAppInfo(appLogDir, true) |
| 103 | + } catch { |
| 104 | + case e: FileNotFoundException => null |
| 105 | + } |
| 106 | + } |
| 107 | + |
| 108 | + /** |
| 109 | + * Check for any updates to event logs in the base directory. This is only effective once |
| 110 | + * the server has been bound. |
| 111 | + * |
| 112 | + * If a new completed application is found, the server renders the associated SparkUI |
| 113 | + * from the application's event logs, attaches this UI to itself, and stores metadata |
| 114 | + * information for this application. |
| 115 | + * |
| 116 | + * If the logs for an existing completed application are no longer found, the server |
| 117 | + * removes all associated information and detaches the SparkUI. |
| 118 | + */ |
| 119 | + def checkForLogs() = synchronized { |
| 120 | + lastLogCheckTime = System.currentTimeMillis |
| 121 | + logDebug("Checking for logs. Time is now %d.".format(lastLogCheckTime)) |
| 122 | + try { |
| 123 | + val logStatus = fs.listStatus(new Path(logDir)) |
| 124 | + val logDirs = if (logStatus != null) logStatus.filter(_.isDir).toSeq else Seq[FileStatus]() |
| 125 | + val logInfos = logDirs |
| 126 | + .sortBy { dir => getModificationTime(dir) } |
| 127 | + .filter { |
| 128 | + dir => fs.isFile(new Path(dir.getPath(), EventLoggingListener.APPLICATION_COMPLETE)) |
| 129 | + } |
| 130 | + |
| 131 | + var currentApps = Map[String, ApplicationHistoryInfo]( |
| 132 | + appList.get().map(app => (app.id -> app)):_*) |
| 133 | + |
| 134 | + // For any application that either (i) is not listed or (ii) has changed since the last time |
| 135 | + // the listing was created (defined by the log dir's modification time), load the app's info. |
| 136 | + // Otherwise just reuse what's already in memory. |
| 137 | + appList.set(logInfos |
| 138 | + .map { dir => |
| 139 | + val curr = currentApps.getOrElse(dir.getPath().getName(), null) |
| 140 | + if (curr == null || curr.lastUpdated < getModificationTime(dir)) { |
| 141 | + loadAppInfo(dir, false) |
| 142 | + } else { |
| 143 | + curr |
| 144 | + } |
| 145 | + } |
| 146 | + .sortBy { info => -info.lastUpdated }) |
| 147 | + } catch { |
| 148 | + case t: Throwable => logError("Exception in checking for event log updates", t) |
| 149 | + } |
| 150 | + } |
| 151 | + |
| 152 | + /** |
| 153 | + * Parse the application's logs to find out the information we need to build the |
| 154 | + * listing page. |
| 155 | + */ |
| 156 | + private def loadAppInfo(logDir: FileStatus, renderUI: Boolean): ApplicationHistoryInfo = { |
| 157 | + val elogInfo = EventLoggingListener.parseLoggingInfo(logDir.getPath(), fs) |
| 158 | + val path = logDir.getPath |
| 159 | + val appId = path.getName |
| 160 | + val replayBus = new ReplayListenerBus(elogInfo.logPaths, fs, elogInfo.compressionCodec) |
| 161 | + val appListener = new ApplicationEventListener |
| 162 | + replayBus.addListener(appListener) |
| 163 | + |
| 164 | + val ui: SparkUI = if (renderUI) { |
| 165 | + val conf = this.conf.clone() |
| 166 | + val appSecManager = new SecurityManager(conf) |
| 167 | + new SparkUI(conf, appSecManager, replayBus, appId, "/history/" + appId) |
| 168 | + // Do not call ui.bind() to avoid creating a new server for each application |
| 169 | + } else { |
| 170 | + null |
| 171 | + } |
| 172 | + |
| 173 | + replayBus.replay() |
| 174 | + val appName = appListener.appName |
| 175 | + val sparkUser = appListener.sparkUser |
| 176 | + val startTime = appListener.startTime |
| 177 | + val endTime = appListener.endTime |
| 178 | + val lastUpdated = getModificationTime(logDir) |
| 179 | + ApplicationHistoryInfo(appId, |
| 180 | + appListener.appName, |
| 181 | + appListener.startTime, |
| 182 | + appListener.endTime, |
| 183 | + getModificationTime(logDir), |
| 184 | + appListener.sparkUser, |
| 185 | + if (renderUI) appListener.viewAcls else null, |
| 186 | + ui) |
| 187 | + } |
| 188 | + |
| 189 | + /** Return when this directory was last modified. */ |
| 190 | + private def getModificationTime(dir: FileStatus): Long = { |
| 191 | + try { |
| 192 | + val logFiles = fs.listStatus(dir.getPath) |
| 193 | + if (logFiles != null && !logFiles.isEmpty) { |
| 194 | + logFiles.map(_.getModificationTime).max |
| 195 | + } else { |
| 196 | + dir.getModificationTime |
| 197 | + } |
| 198 | + } catch { |
| 199 | + case t: Throwable => |
| 200 | + logError("Exception in accessing modification time of %s".format(dir.getPath), t) |
| 201 | + -1L |
| 202 | + } |
| 203 | + } |
| 204 | + |
| 205 | +} |
0 commit comments