[SPARK-7336][HistoryServer] Fix bug that applications status uncorrect on JobHistory UI. #5886
Conversation
ArcherShao
changed the title
|
Test build #31755 has finished for PR 5886 at commit
|
|
On inspection, I don't agree with the scenario you suggest. You say that if a last modification time is T, and a check happens at T + t/2, then it won't notice a file update at T + t/4. But the logic in this case finds files modified >= T. Why would this not update? |
|
I'm not convinced either. Please write a test that fails with the current code and passes with your changes. Also, storing another huge map in |
|
@srowen Check all log files one time will cost some time, say 't', if the first file(app1) check at time T, and the last file(appN) check at time T + t, the last modification time of app1 is T + t / 4, and the last modification time of appN is T + t / 2.Then app1 will ignored at this check, and 'lastModifiedTime' will update to T + t / 2, app1 will ignored at next checks. |
|
I still don't understand. The "last modification time" is not based on how long it took to parse logs. It's based on the modification time of the newest log found. If you scan files at time But feel free to prove me wrong by writing a test for this. |
|
Scan files will cost time, I mean start scan at time T, and finish at time T + t. If some update happen to the first file between T and T + t, the bug will repreduce. |
|
I think this may be the problem scenario:
|
| time >= lastModifiedTime | ||
| val fileName = entry.getPath.getName | ||
| val oldLastModifiedTime = lastModifiedTimes.getOrElse(fileName, -1L) | ||
| time > oldLastModifiedTime |
There was a problem hiding this comment.
I think technically this has to be >= since you might have an update that occurred in the same millisecond. This is very unlikely. But is there harm in reprocessing logs in this code? does it have to guarantee a log isn't read twice?
There was a problem hiding this comment.
If the app is in process, it doesn't matter if we don't reprocess the log file. After the app completed, its' file name will change, and it will get the oldLastModifiedTime = -1, so time > oldLastModifiedTime will be true, and the log file will reprocess.
|
Test build #31875 has finished for PR 5886 at commit
|
| @@ -319,6 +318,15 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock) | |||
|
|
|||
| applications = appsToRetain | |||
|
|
|||
| val modifiedTimesToRetain = new mutable.LinkedHashMap[String, Long]() | |||
There was a problem hiding this comment.
What about simply deleting the entries that are too old? then the member can be val and doesn't need to be @volatile
This shouldn't technically happen since the scan started at time 100. The listing is done in one call, so it's very unlikely such a situation would exist. Still it might be possible. But in any case, this needs a test. |
|
@srowen I guess the part that I don't follow in your example is: how can a scan see a file with mod time = 102 but not see a file with mod time = 101 in the same listing call? Unless HDFS is really screwed up (unlikely), or the user is manually messing with the files' times, this shouldn't happen. |
|
@vanzin , there is time interval from getting the first file's modification time to the last file's. Assume there are 3 files: F1, F2, F3. And before scanning, their modification times are TF1=100, TF2=101, TF3=102 respectively. Then we continue to load F3 mode time, and at time T6=108, we finished loading F3 mode time. At this point, So for the next round, we would not pick up F1 even it has been modified at time T4=105. |
| val fileName = entry.getPath.getName | ||
| val oldLastModifiedTime = lastModifiedTimes.getOrElse(fileName, -1L) | ||
| lastModifiedTimes += (fileName -> time) | ||
| time > oldLastModifiedTime |
There was a problem hiding this comment.
We can not use > to check whether the file is modified. We must use >=. Since there might be modification to the file immediately before and after calling getModificationTime, the result of which is the modified time stay the same. We need use >= to check the time along with checking the file size changing. For some discussions, please refer to SPARK-7189
That doesn't exist. All mod times are retrieved in the same call. There's a single call to "listStatus" which returns all available log files - and log files are now files, and not directories as they used to be. If that were still the case then you'd be correct. So you should never, in the same call to listStatus, get "f2 with mod time 102" and not get "f1 with mod time 101". |
|
And BTW I'm still waiting for that test that shows an issue exists here. |
|
Hi @vanzin , you are correct that a single call to |
|
@vanzin yes the question is really what can happen inside
This could be my ignorance but is this metadata update atomic in the name node? meaning, as far as HDFS is concerned, the file could not have been modified at 101 since it was busy with ... but this boxes the issue into a really small corner. What happens in this case? I realize that some log file doesn't get processed until a bit later then but does the subsequent processing then go wrong? If the application state isn't corrupted or wrong afterwards, I think this isn't worth addressing. |
|
Ok, I buy that scenario. But as I mentioned before, the current solution is not very good: it increases the memory usage of the history server too much. If you have My suggestion: before doing a That may exacerbate the issue raised in SPARK-7189, though.
That's a good question. Because if the mod time changes, then the file is being written to, and its mod time will eventually change again. But I guess the same race can occur if the file is closed / renamed during an app shutdown, while the HS is doing the listing? |
|
@vanzin Thanks for your suggestion. Shoud the empty file's filename be a fixed value or a random value every time creat it ? |
Random is probably better, potentially prefixed by |
|
Test build #32180 has finished for PR 5886 at commit
|
| @@ -184,15 +183,14 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock) | |||
| */ | |||
| private[history] def checkForLogs(): Unit = { | |||
| try { | |||
| val newLastScanTime = getNewLastScanTime() | |||
There was a problem hiding this comment.
So, doesn't this have the opposite 'problem', that a log file will get scanned twice even if it hasn't changed in some cases?
If lastScanTime is 90, and newLastScanTime is 100, and a file is modified (once) at 101 (just after newLastScanTime is established) then it will be read twice. I'm just double-checking that this is fine if it happens only once in a while.
Touching a file seems a little icky but I understand the logic. I can't think of something better that doesn't involve listing the dir again, processing files again frequently, or taking arbitrary guesses about how long the listing takes.
This is worth it in the sense that the cost of missing an update in this very rare case is high? like a correctness issue? you'd miss a bit of history forever?
There was a problem hiding this comment.
That's what I mentioned that my suggestion would make SPARK-7189 worse. There are ways to fix it, such as those discussed in the bug, that don't require too much extra memory in the HS - basically keep track of all log files that have lastModified > newLastScanTime and, if they haven't changed in the next poll, don't re-parse them. That's a lot less state to keep track of than the previous approach in this PR - in fact most of the time there will be nothing to keep track of.
|
Yeah, you have a good point that there's not really a way to 100% solve similar very-rare scenarios, like a file rename or deletion. This isn't an issue that affects app correctness right? just collecting logs? which is important, but I'm wondering if this approach would only open up other rare problems: logs collection fails entirely because touching a file fails, etc. Is it worth even trying to address this? it could be documented in comments at least. |
It depends on what you're calling "app correctness" here. Isn't missing an updated / renamed log a correctness issue?
True, but that should only cause one poll to fail. It shouldn't cause the HS to go down. If the HS can't persistently write to HDFS, that means some configuration is wrong, since the history directory is expected to be world-writable.
I think it's worth it if it can be done without harming other parts of the HS (like requiring way more memory than before). I think the latest approach is close; if some code is added to address SPARK-7189, then it would be ready to go. The issue raise by this PR should be a really, really rare situation though. |
|
Test build #32359 has finished for PR 5886 at commit
|
|
Test build #32376 has finished for PR 5886 at commit
|
|
Test build #32456 has finished for PR 5886 at commit
|
|
@vanzin , the current implementation will make SPARK-7189 worse, what about introduce a hashMap to maintain the filename, modifiedTime of each file, file size, say mutable.HashMap[String, Long, Long], it can not only handle the modification case and also rename/delete cases. Since each file's modification time is maintained, this can both solve the problem of the race condition in this issue and also solve SPARK-7189? And it will only introduce extra memory with size of the hashMap size. |
That is pretty expensive for all files - hash maps use a ton of extra memory per entry. If you do that just for the files that have |
|
Test build #33317 has finished for PR 5886 at commit
|
|
Test build #33401 has finished for PR 5886 at commit
|
| val path = new Path(logDir, fileName) | ||
| val fos = fs.create(path) | ||
|
|
||
| val newLastScanTime = |
There was a problem hiding this comment.
You don't actually need to declare this variable.
|
The problem @srowen mentions exists, but I think in the end it's not really a problem; that will mostly happen when applications are actively writing to their logs, and in those cases we do want to reload the file to get new events. The current change looks ok to me (with one minor nit), but you'll need to update the code to match the current master. |
|
(Also, if there's a test that can be written - or an existing one modified - to verify the new behavior, it wouldn't hurt.) |
|
@ArcherShao can you rebase to master? |
ArcherShao
force-pushed
the
SPARK-7336
branch
from
68a9633
to
a4a876a
Compare
|
Test build #41909 has finished for PR 5886 at commit
|
| @@ -19,6 +19,7 @@ package org.apache.spark.deploy.history | |||
|
|
|||
| import java.io.{BufferedInputStream, FileNotFoundException, InputStream, IOException, OutputStream} | |||
| import java.util.concurrent.{ExecutorService, Executors, TimeUnit} | |||
| import java.util.UUID | |||
|
LGTM, I'll merge and fix the nit in the process. |
No description provided.