[SPARK-20654][core] Add config to limit disk usage of the history server. #20011

vanzin · 2017-12-18T21:29:44Z

This change adds a new configuration option and support code that limits
how much disk space the SHS will use. The default value is pretty generous
so that applications will, hopefully, only rarely need to be replayed
because of their disk stored being evicted.

This works by keeping track of how much data each application is using.
Also, because it's not possible to know, before replaying, how much space
will be needed, it's possible that usage will exceed the configured limit
temporarily. The code uses the concept of a "lease" to try to limit how
much the SHS will exceed the limit in those cases.

Active UIs are also tracked, so they're never deleted. This works in
tandem with the existing option of how many active UIs are loaded; because
unused UIs will be unloaded, their disk stores will also become candidates
for deletion. If the data is not deleted, though, re-loading the UI is
pretty quick.

…ver. This change adds a new configuration option and support code that limits how much disk space the SHS will use. The default value is pretty generous so that applications will, hopefully, only rarely need to be replayed because of their disk stored being evicted. This works by keeping track of how much data each application is using. Also, because it's not possible to know, before replaying, how much space will be needed, it's possible that usage will exceed the configured limit temporarily. The code uses the concept of a "lease" to try to limit how much the SHS will exceed the limit in those cases. Active UIs are also tracked, so they're never deleted. This works in tandem with the existing option of how many active UIs are loaded; because unused UIs will be unloaded, their disk stores will also become candidates for deletion. If the data is not deleted, though, re-loading the UI is pretty quick.

vanzin · 2017-12-18T22:07:53Z

Some previous comments for this PR can be found at: vanzin#39

SparkQA · 2017-12-19T00:53:28Z

Test build #85072 has finished for PR 20011 at commit 8b43f4d.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

vanzin · 2017-12-19T01:00:59Z

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

+   * event log. By default it's 30% of the event log size.
+   */
+  private def approximateSize(eventLogSize: Long): Long = {
+    math.ceil(eventLogSizeRatio * eventLogSize).toLong


I'll probably need to look at a better heuristic here. This is probably not a good approximation if the logs are compressed, and #20013 also affects this.

Also remove the config to tweak the heuristic since it doesn't make a lot of sense anymore.

SparkQA · 2017-12-19T23:05:18Z

Test build #85129 has finished for PR 20011 at commit 31c2755.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-12-22T00:34:45Z

Test build #85284 has finished for PR 20011 at commit 2d72dd1.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

squito · 2017-12-21T22:14:44Z

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

+
+    // Cap the value at 10% of the max size; this assumes that element cleanup will put a cap on
+    // how large the disk store can get, which may not always be the case.
+    math.min(expectedSize, maxUsage / 10)


this seems really arbitrary, and could have a ton of error. Do we really need it at all?

Well, all of these heuristics can have a lot of error. Increasing this increases the odds that an existing store will be deleted if another log starts to be parsed in parallel.

Don't really feel strongly either way, so away it goes.

squito · 2017-12-22T15:26:19Z

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

+ * @param listing The listing store, used to persist usage data.
+ * @param clock Clock instance to use.
+ */
+private class DiskStoreManager(


there is another very different DiskStore, can we name this something else so its easier to tell they're unrelated? KVStoreDiskManager?

squito · 2017-12-22T15:43:46Z

core/src/test/scala/org/apache/spark/deploy/history/DiskStoreManagerSuite.scala

+  private def mockManager(): DiskStoreManager = {
+    val conf = new SparkConf().set(MAX_LOCAL_DISK_USAGE, MAX_USAGE)
+    val manager = spy(new DiskStoreManager(conf, testDir, store, new ManualClock()))
+    doReturn(0L).when(manager).sizeOf(any(classOf[File]))


tests pass without this line ... and its kinda strange, I tried removing it because I was wondering why you would want this

Just added in case the method was called with an unexpected file. Will remove.

squito · 2017-12-22T15:48:59Z

core/src/test/scala/org/apache/spark/deploy/history/DiskStoreManagerSuite.scala

+    lease3.rollback()
+    lease4.rollback()
+    assert(hasFreeSpace(manager, 1))
+    assert(!hasFreeSpace(manager, 2))


nit: I don't think hasFreeSpace actually helps at all, here you could replace both of these with assert(manager.free() === 1) and everywhere else do assert(manager.free() === 0) etc.

squito · 2017-12-22T15:59:17Z

core/src/test/scala/org/apache/spark/deploy/history/DiskStoreManagerSuite.scala

+    val dst2 = lease2.commit("app2", None)
+    assert(!hasFreeSpace(manager, 1))
+
+    // Rollback 3 and 4, now there should be 1 left.


can you switch the naming scheme to be leaseA, leaseB, etc.? its a bit hard to tell when you're talking about sizes and when you're referring to specific leases.

squito · 2017-12-22T16:04:20Z

core/src/test/scala/org/apache/spark/deploy/history/DiskStoreManagerSuite.scala

+    val lease5 = manager.lease(1)
+    doReturn(3L).when(manager).sizeOf(meq(lease5.path))
+    lease5.commit("app2", None)
+    assert(dst2.exists())


would help here to have a comment explaining the case this really represents -- replaying an app because the event log has been updated, so replacing the old kvstore. Also just to make that clear you could add

val dst5 = lease5.commit("app2", None) assert(dst5 === dst2)

(at first I thought "app2" was a copy/paste mistake here)

squito · 2017-12-22T16:08:54Z

core/src/test/scala/org/apache/spark/deploy/history/DiskStoreManagerSuite.scala

+    // Try a big lease that should cause the committed app to be evicted.
+    val lease6 = manager.lease(6)
+    assert(!dst2.exists())
+    assert(!hasFreeSpace(manager, 1))


add a case with getting another lease, even though you're already beyond the size limit (don't think that is covered?)

squito · 2017-12-22T16:37:07Z

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

+  /** Visible for testing. Return the size of a directory. */
+  private[history] def sizeOf(path: File): Long = FileUtils.sizeOf(path)
+
+  private[history] class Lease(val path: File, private val leased: Long) {


can you rename path to tmpPath

squito · 2017-12-22T16:49:04Z

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

+
+      active.synchronized {
+        require(!active.contains(appId -> attemptId),
+          s"Cannot commit lease for active application $appId / $attemptId")


what is the scenario you're worried about here? you could have two threads both trying to call commit, and they might both get past this check, and then clobber each other later on, so this is more limited protection. The idea is just to make sure that you never do a commit while the UI has a store open, as a check of the whole UI cache eviction process?

This is more of a sanity check that the situation you describe should not happen. The SHS code should be ensuring that there's only a single thread parsing logs for an application, and that while that happens, the app's UI is not loaded. This just asserts that's the case.

Also fix a couple of bugs found in the process.

squito · 2017-12-22T23:18:45Z

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

+        logInfo(s"Deleting store for ${info.appId}/${info.attemptId}.")
+        deleteStore(new File(info.path))
+        updateUsage(-info.size, committed = true)
+      }


I like the change to track committed space. I'm wondering if we should log more here, even at Info level, summarizing the available space (I can see users complaining that they configured it to 10GB but its taking 12GB and wanting an explanation -- would be nice if the logs already had all the relevant info). Maybe something like

val totalNeeded = size var activeSize = 0 ... if (!isActive) { evicted += info needed -= info.size } else { activeSize += info.size } ... logInfo(s"Deleted ${evicted.size} apps to free ${Utils.bytesToString(needed)}) if (needed > 0) { val current = currentUsage.get() val leased = Utils.bytesToString(current - committedUsage.get()) logInfo(s"Did not free requested ${Utils.bytesToString(totalNeeded)}. Current usage is ${Utils.bytesToString(current). $leased (estimated) used by apps actively updating their kvstores; ${Utils.bytesToString(active)} used by active applications.") }

Yeah, I'll add some more logs.

SparkQA · 2017-12-23T01:31:02Z

Test build #85320 has finished for PR 20011 at commit be3f130.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-12-23T03:19:09Z

Test build #85326 has finished for PR 20011 at commit 931b2d2.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

squito

lgtm

squito · 2017-12-29T16:45:03Z

merged to master

vanzin commented Dec 19, 2017

View reviewed changes

Change approximate size heuristic to account for compression.

31c2755

Also remove the config to tweak the heuristic since it doesn't make a lot of sense anymore.

Merge branch 'master' into SPARK-20654

2d72dd1

squito reviewed Dec 22, 2017

View reviewed changes

Marcelo Vanzin added 3 commits December 22, 2017 13:27

Feedback.

c1822aa

Rename class.

4d58f7d

Track committed usage separately to make log message more accurate.

be3f130

Also fix a couple of bugs found in the process.

squito reviewed Dec 22, 2017

View reviewed changes

Add more logs.

931b2d2

squito approved these changes Dec 27, 2017

View reviewed changes

asfgit closed this in 8b49704 Dec 29, 2017

vanzin deleted the SPARK-20654 branch January 2, 2018 19:21

[SPARK-20654][core] Add config to limit disk usage of the history server. #20011

[SPARK-20654][core] Add config to limit disk usage of the history server. #20011

Uh oh!

Conversation

vanzin commented Dec 18, 2017

Uh oh!

vanzin commented Dec 18, 2017

Uh oh!

SparkQA commented Dec 19, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Dec 19, 2017

Uh oh!

SparkQA commented Dec 22, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Dec 23, 2017

Uh oh!

SparkQA commented Dec 23, 2017

Uh oh!

squito left a comment

Choose a reason for hiding this comment

Uh oh!

squito commented Dec 29, 2017

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants