Skip to content

[SPARK-1768] Make event logger use a single file. #1

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
vanzin wants to merge 12 commits into from
Closed

[SPARK-1768] Make event logger use a single file. #1

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
f96aa78
Separate history server from history backend.
May 7, 2014
a1d4f1e
Rudimentary paging support for the history UI.
May 8, 2014
461fac8
Ensure server.stop() is called when shutting down.
May 9, 2014
d62cd43
Better handle failures when loading app info.
May 9, 2014
8e2aad0
Fix scalastyle issues.
May 9, 2014
81b69fb
Fix a comment.
May 9, 2014
0bb7b57
Review feedback.
Jun 2, 2014
bd790a0
Initialize new app array to expected size.
Jun 2, 2014
43e270f
Cosmetic change to listing header.
Jun 3, 2014
e043700
Make class package-private.
Jun 3, 2014
9b11205
Make event logger use a single file.
May 22, 2014
9dcfab9
Make history server parse old-style log directories.
May 23, 2014
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Review feedback. 
Use monotonic time, plus other stylistic things.
  • Loading branch information
Marcelo Vanzin committed Jun 2, 2014
commit 0bb7b57ead82d109b74cbb40bed698f0b1507014
18 changes: 10 additions & 8 deletions core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@ class FsHistoryProvider(conf: SparkConf) extends ApplicationHistoryProvider
private val fs = Utils.getHadoopFileSystem(logDir)

// A timestamp of when the disk was last accessed to check for log updates
private var lastLogCheckTime = -1L
private var lastLogCheckTimeMs = -1L

// List of applications, in order from newest to oldest.
private val appList = new AtomicReference[Seq[ApplicationHistoryInfo]](Nil)
Expand All @@ -55,13 +55,13 @@ class FsHistoryProvider(conf: SparkConf) extends ApplicationHistoryProvider
private val logCheckingThread = new Thread("LogCheckingThread") {
override def run() = Utils.logUncaughtExceptions {
while (true) {
val now = System.currentTimeMillis
if (now - lastLogCheckTime > UPDATE_INTERVAL_MS) {
val now = getMonotonicTime()
if (now - lastLogCheckTimeMs > UPDATE_INTERVAL_MS) {
Thread.sleep(UPDATE_INTERVAL_MS)
} else {
// If the user has manually checked for logs recently, wait until
// UPDATE_INTERVAL_MS after the last check time
Thread.sleep(lastLogCheckTime + UPDATE_INTERVAL_MS - now)
Thread.sleep(lastLogCheckTimeMs + UPDATE_INTERVAL_MS - now)
}
checkForLogs()
}
Expand Down Expand Up @@ -108,13 +108,12 @@ class FsHistoryProvider(conf: SparkConf) extends ApplicationHistoryProvider
* applications that hasn't been updated since last time the logs were checked.
*/
def checkForLogs() = synchronized {
lastLogCheckTime = System.currentTimeMillis
logDebug("Checking for logs. Time is now %d.".format(lastLogCheckTime))
lastLogCheckTimeMs = getMonotonicTime()
logDebug("Checking for logs. Time is now %d.".format(lastLogCheckTimeMs))
try {
val logStatus = fs.listStatus(new Path(logDir))
val logDirs = if (logStatus != null) logStatus.filter(_.isDir).toSeq else Seq[FileStatus]()
val logInfos = logDirs
.sortBy { dir => getModificationTime(dir) }
.filter {
dir => fs.isFile(new Path(dir.getPath(), EventLoggingListener.APPLICATION_COMPLETE))
}
Expand All @@ -125,7 +124,7 @@ class FsHistoryProvider(conf: SparkConf) extends ApplicationHistoryProvider
// For any application that either (i) is not listed or (ii) has changed since the last time
// the listing was created (defined by the log dir's modification time), load the app's info.
// Otherwise just reuse what's already in memory.
val newApps = new mutable.ListBuffer[ApplicationHistoryInfo]
val newApps = new mutable.ArrayBuffer[ApplicationHistoryInfo]
for (dir <- logInfos) {
val curr = currentApps.getOrElse(dir.getPath().getName(), null)
if (curr == null || curr.lastUpdated < getModificationTime(dir)) {
Expand Down Expand Up @@ -198,4 +197,7 @@ class FsHistoryProvider(conf: SparkConf) extends ApplicationHistoryProvider
}
}

/** Returns the system's mononotically increasing time. */
private def getMonotonicTime() = System.nanoTime() / (1000 * 1000)

}
11 changes: 5 additions & 6 deletions core/src/main/scala/org/apache/spark/deploy/history/HistoryServer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,14 +59,13 @@ class HistoryServer(
private val appLoader = new CacheLoader[String, SparkUI] {
override def load(key: String): SparkUI = {
val info = provider.getAppInfo(key)
if (info != null) {
info.ui.getSecurityManager.setUIAcls(uiAclsEnabled)
info.ui.getSecurityManager.setViewAcls(info.sparkUser, info.viewAcls)
attachSparkUI(info.ui)
info.ui
} else {
if (info == null) {
throw new NoSuchElementException()
}
info.ui.getSecurityManager.setUIAcls(uiAclsEnabled)
info.ui.getSecurityManager.setViewAcls(info.sparkUser, info.viewAcls)
attachSparkUI(info.ui)
info.ui
}
}

Expand Down