|
16 | 16 | */ |
17 | 17 | package org.apache.spark.status.api.v1 |
18 | 18 |
|
19 | | -import java.io.{BufferedInputStream, FileInputStream, OutputStream, File, InputStream} |
20 | | -import javax.ws.rs.ext.Provider |
| 19 | +import java.io.{BufferedInputStream, FileInputStream, OutputStream} |
21 | 20 | import javax.ws.rs.{GET, Produces} |
22 | | -import javax.ws.rs.core.{StreamingOutput, MultivaluedMap, MediaType} |
| 21 | +import javax.ws.rs.core.{Response, StreamingOutput, MediaType} |
23 | 22 |
|
24 | 23 | import org.apache.spark.deploy.history.HistoryServer |
25 | 24 | import org.apache.spark.util.Utils |
26 | 25 |
|
27 | | - |
| 26 | +@Produces(Array(MediaType.APPLICATION_OCTET_STREAM)) |
28 | 27 | private[v1] class EventLogDownloadResource(val uIRoot: UIRoot, val appId: String) { |
29 | 28 |
|
30 | | - private def getErrorOutput(err: String): StreamingOutput = { |
31 | | - new StreamingOutput { |
32 | | - override def write(outputStream: OutputStream): Unit = { |
33 | | - outputStream.write( |
34 | | - s"File download not available for application : $appId due to $err".getBytes("utf-8")) |
35 | | - } |
36 | | - } |
37 | | - } |
38 | | - |
39 | 29 | @GET |
40 | | - @Produces(Array(MediaType.APPLICATION_OCTET_STREAM)) |
41 | | - def getEventLogs(): StreamingOutput = { |
| 30 | + def getEventLogs(): Response = { |
42 | 31 | uIRoot match { |
43 | 32 | case hs: HistoryServer => |
44 | 33 | val dir = Utils.createTempDir() |
45 | 34 | Utils.chmod700(dir) |
46 | 35 | hs.copyEventLogsToDirectory(appId, dir) |
47 | 36 | dir.listFiles().headOption.foreach { file => |
48 | | - return new StreamingOutput { |
| 37 | + val stream = new StreamingOutput { |
49 | 38 | override def write(output: OutputStream): Unit = { |
50 | | - val inStream = new BufferedInputStream(new FileInputStream(file)) |
51 | | - val buffer = new Array[Byte](1024 * 1024) |
52 | | - var dataRemains = true |
53 | | - while (dataRemains) { |
54 | | - val read = inStream.read(buffer) |
55 | | - if (read > 0) { |
56 | | - output.write(buffer, 0, read) |
57 | | - } else { |
58 | | - dataRemains = false |
| 39 | + try { |
| 40 | + val inStream = new BufferedInputStream(new FileInputStream(file)) |
| 41 | + val buffer = new Array[Byte](1024 * 1024) |
| 42 | + var dataRemains = true |
| 43 | + while (dataRemains) { |
| 44 | + val read = inStream.read(buffer) |
| 45 | + if (read > 0) { |
| 46 | + output.write(buffer, 0, read) |
| 47 | + } else { |
| 48 | + dataRemains = false |
| 49 | + } |
59 | 50 | } |
| 51 | + output.flush() |
| 52 | + } finally { |
| 53 | + Utils.deleteRecursively(dir) |
60 | 54 | } |
61 | | - output.flush() |
62 | 55 | } |
63 | 56 | } |
| 57 | + return Response.ok(stream) |
| 58 | + .header("Content-Length", file.length().toString) |
| 59 | + .header("Content-Disposition", s"attachment; filename=${file.getName}") |
| 60 | + .header("Content-Type", MediaType.APPLICATION_OCTET_STREAM) |
| 61 | + .build() |
64 | 62 | } |
65 | | - getErrorOutput("No files in dir.") |
66 | | - case _ => getErrorOutput("hs not history server") |
| 63 | + Response.serverError() |
| 64 | + .entity(s"Event logs for $appId not found.") |
| 65 | + .status(Response.Status.NOT_FOUND) |
| 66 | + .build() |
| 67 | + case _ => |
| 68 | + Response.serverError() |
| 69 | + .entity("History Server is not running - cannot return event logs.") |
| 70 | + .status(Response.Status.SERVICE_UNAVAILABLE) |
| 71 | + .build() |
67 | 72 | } |
68 | 73 | } |
69 | 74 | } |
|
0 commit comments