[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user asfgit closed the pull request at: https://github.com/apache/spark/pull/20474 --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user squito commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r167917613
--- Diff:
core/src/main/scala/org/apache/spark/status/api/v1/OneApplicationResource.scala
---
@@ -51,6 +51,29 @@ private[v1] class AbstractApplicationResource extends
BaseAppResource {
@Path("executors")
def executorList(): Seq[ExecutorSummary] =
withUI(_.store.executorList(true))
+ @GET
+ @Path("executors/{executorId}/threads")
+ def threadDump(@PathParam("executorId") execId: String):
Array[ThreadStackTrace] = withUI { ui =>
+if (execId != SparkContext.DRIVER_IDENTIFIER &&
!execId.forall(Character.isDigit)) {
+ throw new BadParameterException(
+s"Invalid executorId: neither '${SparkContext.DRIVER_IDENTIFIER}'
nor number.")
+}
+
+val safeSparkContext = ui.sc.getOrElse {
+ throw new ServiceUnavailable("Thread dumps not available through the
history server.")
+}
+
+ui.store.asOption(ui.store.executorSummary(execId)) match {
+ case Some(executorSummary) if executorSummary.isActive =>
+ val safeThreadDump =
safeSparkContext.getExecutorThreadDump(execId).getOrElse {
+throw new NotFoundException("No thread dump is available.")
+ }
+ return safeThreadDump
--- End diff --
you don't need `return` here
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user squito commented on a diff in the pull request: https://github.com/apache/spark/pull/20474#discussion_r167917733 --- Diff: docs/monitoring.md --- @@ -347,6 +347,10 @@ can be identified by their `[attempt-id]`. In the API listed below, when running /applications/[app-id]/executors A list of all active executors for the given application. + + /applications/[app-id]/executors/[executor-id]/threads +Stack traces of all the threads running within the given active executor. --- End diff -- Add that this is not available via the history server. --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r166465338
--- Diff:
core/src/main/scala/org/apache/spark/status/api/v1/OneApplicationResource.scala
---
@@ -51,6 +51,30 @@ private[v1] class AbstractApplicationResource extends
BaseAppResource {
@Path("executors")
def executorList(): Seq[ExecutorSummary] =
withUI(_.store.executorList(true))
+ @GET
+ @Path("executors/{executorId}/threads")
+ def threadDump(@PathParam("executorId") executorId: String):
Array[ThreadStackTrace] =
+withUI { ui =>
+if (executorId != SparkContext.DRIVER_IDENTIFIER &&
!executorId.forall(Character.isDigit)) {
--- End diff --
This whole block now needs to be indented further, and a `{` added in the
declaration.
(Or, if you call the parameter `execId` instead, everything fits in one
line...)
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r166409901
--- Diff: core/src/main/scala/org/apache/spark/ui/SparkUI.scala ---
@@ -17,7 +17,7 @@
package org.apache.spark.ui
-import java.util.{Date, List => JList, ServiceLoader}
+import java.util.Date
--- End diff --
No code seems to be really changing in this file, so I'd avoid touching it.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r166409000
--- Diff:
core/src/main/scala/org/apache/spark/status/api/v1/OneApplicationResource.scala
---
@@ -51,6 +51,31 @@ private[v1] class AbstractApplicationResource extends
BaseAppResource {
@Path("executors")
def executorList(): Seq[ExecutorSummary] =
withUI(_.store.executorList(true))
+ @GET
+ @Path("executors/{executorId}/threads")
+ def threadDump(@PathParam("executorId") executorId: String): Response =
withUI { ui =>
+def isAllDigits(x: String) = x.forall(Character.isDigit)
--- End diff --
You only call this in one place, so I'd just inline it. The code is just as
self-explanatory as the method name.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r166409259
--- Diff:
core/src/main/scala/org/apache/spark/status/api/v1/OneApplicationResource.scala
---
@@ -51,6 +51,31 @@ private[v1] class AbstractApplicationResource extends
BaseAppResource {
@Path("executors")
def executorList(): Seq[ExecutorSummary] =
withUI(_.store.executorList(true))
+ @GET
+ @Path("executors/{executorId}/threads")
+ def threadDump(@PathParam("executorId") executorId: String): Response =
withUI { ui =>
--- End diff --
You can change the return type to `Array[ThreadStackTrace]` now (and adjust
the only path that actually returns something).
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r166409473
--- Diff:
core/src/main/scala/org/apache/spark/status/api/v1/OneApplicationResource.scala
---
@@ -51,6 +51,31 @@ private[v1] class AbstractApplicationResource extends
BaseAppResource {
@Path("executors")
def executorList(): Seq[ExecutorSummary] =
withUI(_.store.executorList(true))
+ @GET
+ @Path("executors/{executorId}/threads")
+ def threadDump(@PathParam("executorId") executorId: String): Response =
withUI { ui =>
+def isAllDigits(x: String) = x.forall(Character.isDigit)
+
+if (executorId != SparkContext.DRIVER_IDENTIFIER &&
!isAllDigits(executorId)) {
+ throw new BadParameterException(
+s"Invalid executorId: neither '${SparkContext.DRIVER_IDENTIFIER}'
nor number.")
+}
+
+val safeSparkContext = ui.sc.getOrElse {
+ throw new ServiceUnavailable("Thread dumps not available through the
history server.")
+}
+
+ui.store.asOption(ui.store.executorSummary(executorId)) match {
+ case Some(executorSummary) if executorSummary.isActive =>
+ val safeThreadDump =
safeSparkContext.getExecutorThreadDump(executorId).getOrElse {
+throw new NotFoundException("No thread dump is available.")
+ }
+ return Response.ok(safeThreadDump).build()
+ case Some(_) => throw new BadParameterException("Executor is already
dead.")
--- End diff --
nit: "Executor is not active."
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user attilapiros commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r166285595
--- Diff:
core/src/main/scala/org/apache/spark/status/api/v1/OneApplicationResource.scala
---
@@ -51,6 +51,52 @@ private[v1] class AbstractApplicationResource extends
BaseAppResource {
@Path("executors")
def executorList(): Seq[ExecutorSummary] =
withUI(_.store.executorList(true))
+ @GET
+ @Path("executors/{executorId}/threads")
+ def threadDump(@PathParam("executorId") executorId: String): Response =
withUI { ui =>
--- End diff --
Review comments are applied then retested and test results are updated.
Moreover I have removed the condition with UIUtils.stripXSS as the existing
check (whether the executorId is 'driver' or number) seams to me enough.
Some if conditions on options are transformed to a val initialised with a
getOrElse as the option value is safe to be used elsewhere.
As I have seen the JAX RS is at 2.0.1, so I did not touch the mapping of
exceptions to JSON.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r166089402
--- Diff:
core/src/main/scala/org/apache/spark/status/api/v1/OneApplicationResource.scala
---
@@ -51,6 +51,52 @@ private[v1] class AbstractApplicationResource extends
BaseAppResource {
@Path("executors")
def executorList(): Seq[ExecutorSummary] =
withUI(_.store.executorList(true))
+ @GET
+ @Path("executors/{executorId}/threads")
+ def threadDump(@PathParam("executorId") executorId: String): Response =
withUI { ui =>
--- End diff --
If you throw exceptions in the code below instead, you can have a more
specialized return type here, which I slightly prefer.
Not all errors currently have exceptions (e.g. 503 is missing), but it's
easy to add them.
(I also don't remember exactly what version of JAX-RS is being used by
Spark, but if it's 2.1, that version has all these exceptions already, so some
cleanup could be done here. 2.0 doesn't have them though.)
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r166087219
--- Diff:
core/src/main/scala/org/apache/spark/status/api/v1/OneApplicationResource.scala
---
@@ -51,6 +51,52 @@ private[v1] class AbstractApplicationResource extends
BaseAppResource {
@Path("executors")
def executorList(): Seq[ExecutorSummary] =
withUI(_.store.executorList(true))
+ @GET
+ @Path("executors/{executorId}/threads")
+ def threadDump(@PathParam("executorId") executorId: String): Response =
withUI { ui =>
+val safeExecutorId =
+ Option(UIUtils.stripXSS(executorId)).map { executorId =>
+UIUtils.decodeURLParameter(executorId)
+ }.getOrElse {
+throw new IllegalArgumentException(s"Missing executorId parameter")
+ }
+
+def isAllDigits(x: String) = x.forall(Character.isDigit)
+
+if (executorId != SparkContext.DRIVER_IDENTIFIER &&
!isAllDigits(executorId)) {
+ Response.serverError()
+.entity(s"Invalid executorId: neither
'${SparkContext.DRIVER_IDENTIFIER}' nor number.")
+.status(Response.Status.BAD_REQUEST)
+.build()
+} else {
+ ui.store.asOption(ui.store.executorSummary(executorId)).map {
executorSummary =>
+if (executorSummary.isActive) {
+ ui.sc match {
+case Some(sc) => sc.getExecutorThreadDump(safeExecutorId)
--- End diff --
nit: move case statement to next line when you have a multi-line statement.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r166088802
--- Diff:
core/src/main/scala/org/apache/spark/status/api/v1/OneApplicationResource.scala
---
@@ -51,6 +51,52 @@ private[v1] class AbstractApplicationResource extends
BaseAppResource {
@Path("executors")
def executorList(): Seq[ExecutorSummary] =
withUI(_.store.executorList(true))
+ @GET
+ @Path("executors/{executorId}/threads")
+ def threadDump(@PathParam("executorId") executorId: String): Response =
withUI { ui =>
+val safeExecutorId =
+ Option(UIUtils.stripXSS(executorId)).map { executorId =>
+UIUtils.decodeURLParameter(executorId)
+ }.getOrElse {
+throw new IllegalArgumentException(s"Missing executorId parameter")
+ }
+
+def isAllDigits(x: String) = x.forall(Character.isDigit)
+
+if (executorId != SparkContext.DRIVER_IDENTIFIER &&
!isAllDigits(executorId)) {
+ Response.serverError()
+.entity(s"Invalid executorId: neither
'${SparkContext.DRIVER_IDENTIFIER}' nor number.")
+.status(Response.Status.BAD_REQUEST)
+.build()
+} else {
+ ui.store.asOption(ui.store.executorSummary(executorId)).map {
executorSummary =>
+if (executorSummary.isActive) {
+ ui.sc match {
+case Some(sc) => sc.getExecutorThreadDump(safeExecutorId)
+ .map(Response.ok(_).build())
+ .getOrElse(Response.serverError()
--- End diff --
Throw `NotFoundException` instead.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r166088682
--- Diff:
core/src/main/scala/org/apache/spark/status/api/v1/OneApplicationResource.scala
---
@@ -51,6 +51,52 @@ private[v1] class AbstractApplicationResource extends
BaseAppResource {
@Path("executors")
def executorList(): Seq[ExecutorSummary] =
withUI(_.store.executorList(true))
+ @GET
+ @Path("executors/{executorId}/threads")
+ def threadDump(@PathParam("executorId") executorId: String): Response =
withUI { ui =>
+val safeExecutorId =
+ Option(UIUtils.stripXSS(executorId)).map { executorId =>
+UIUtils.decodeURLParameter(executorId)
+ }.getOrElse {
+throw new IllegalArgumentException(s"Missing executorId parameter")
+ }
+
+def isAllDigits(x: String) = x.forall(Character.isDigit)
+
+if (executorId != SparkContext.DRIVER_IDENTIFIER &&
!isAllDigits(executorId)) {
+ Response.serverError()
--- End diff --
Throw `BadParameterException` instead.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r166089445
--- Diff:
core/src/main/scala/org/apache/spark/status/api/v1/ApiRootResource.scala ---
@@ -93,6 +93,7 @@ private[spark] trait UIRoot {
.status(Response.Status.SERVICE_UNAVAILABLE)
.build()
}
+
--- End diff --
No need to change this file.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r166089845
--- Diff: core/src/main/scala/org/apache/spark/util/Utils.scala ---
@@ -2168,7 +2169,23 @@ private[spark] object Utils extends Logging {
// We need to filter out null values here because dumpAllThreads() may
return null array
// elements for threads that are dead / don't exist.
val threadInfos =
ManagementFactory.getThreadMXBean.dumpAllThreads(true, true).filter(_ != null)
-threadInfos.sortBy(_.getThreadId).map(threadInfoToThreadStackTrace)
+threadInfos.sortWith {
+ case (threadTrace1, threadTrace2) =>
--- End diff --
nit: fits in previous line
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r166088600
--- Diff:
core/src/main/scala/org/apache/spark/status/api/v1/OneApplicationResource.scala
---
@@ -51,6 +51,52 @@ private[v1] class AbstractApplicationResource extends
BaseAppResource {
@Path("executors")
def executorList(): Seq[ExecutorSummary] =
withUI(_.store.executorList(true))
+ @GET
+ @Path("executors/{executorId}/threads")
+ def threadDump(@PathParam("executorId") executorId: String): Response =
withUI { ui =>
+val safeExecutorId =
+ Option(UIUtils.stripXSS(executorId)).map { executorId =>
+UIUtils.decodeURLParameter(executorId)
+ }.getOrElse {
+throw new IllegalArgumentException(s"Missing executorId parameter")
+ }
+
+def isAllDigits(x: String) = x.forall(Character.isDigit)
+
+if (executorId != SparkContext.DRIVER_IDENTIFIER &&
!isAllDigits(executorId)) {
--- End diff --
So this kind of code is where I prefer the "return early" approach to avoid
nesting. It makes code easier to follow IMO.
```
if (executorId != SparkContext.DRIVER_IDENTIFIER &&
!isAllDigits(executorId)) {
return // or throw
}
if (ui.sc.isEmpty) {
return // or throw
}
ui.store.asOption(...) match {
case Some(e) if e.isActive => ...
case Some(e) => ...
case _ => ...
}
```
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user attilapiros commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r165741532
--- Diff:
core/src/main/scala/org/apache/spark/status/api/v1/OneApplicationResource.scala
---
@@ -51,6 +52,43 @@ private[v1] class AbstractApplicationResource extends
BaseAppResource {
@Path("executors")
def executorList(): Seq[ExecutorSummary] =
withUI(_.store.executorList(true))
+ @GET
+ @Path("executors/{executorId}/threads")
+ def threadDump(@PathParam("executorId") executorId: String): Response =
withUI { ui =>
+uiRoot.executorThreadDumpsNotAvailableError().getOrElse {
+ val safeExecutorId =
+Option(UIUtils.stripXSS(executorId)).map { executorId =>
+ UIUtils.decodeURLParameter(executorId)
+}.getOrElse {
+ throw new IllegalArgumentException(s"Missing executorId
parameter")
+}
+
+ def isAllDigits(x: String) = x.forall(Character.isDigit)
+
+ if (executorId != SparkContext.DRIVER_IDENTIFIER &&
!isAllDigits(executorId)) {
+Response.serverError()
+ .entity(s"Invalid executorId: neither
'${SparkContext.DRIVER_IDENTIFIER}' nor number.")
+ .status(Response.Status.BAD_REQUEST)
+ .build()
+ } else {
+if
(ui.store.asOption(ui.store.executorSummary(executorId)).exists(!_.isActive)) {
--- End diff --
Yeah, I see your point.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r165741132
--- Diff:
core/src/main/scala/org/apache/spark/status/api/v1/OneApplicationResource.scala
---
@@ -51,6 +52,43 @@ private[v1] class AbstractApplicationResource extends
BaseAppResource {
@Path("executors")
def executorList(): Seq[ExecutorSummary] =
withUI(_.store.executorList(true))
+ @GET
+ @Path("executors/{executorId}/threads")
+ def threadDump(@PathParam("executorId") executorId: String): Response =
withUI { ui =>
+uiRoot.executorThreadDumpsNotAvailableError().getOrElse {
+ val safeExecutorId =
+Option(UIUtils.stripXSS(executorId)).map { executorId =>
+ UIUtils.decodeURLParameter(executorId)
+}.getOrElse {
+ throw new IllegalArgumentException(s"Missing executorId
parameter")
+}
+
+ def isAllDigits(x: String) = x.forall(Character.isDigit)
+
+ if (executorId != SparkContext.DRIVER_IDENTIFIER &&
!isAllDigits(executorId)) {
+Response.serverError()
+ .entity(s"Invalid executorId: neither
'${SparkContext.DRIVER_IDENTIFIER}' nor number.")
+ .status(Response.Status.BAD_REQUEST)
+ .build()
+ } else {
+if
(ui.store.asOption(ui.store.executorSummary(executorId)).exists(!_.isActive)) {
--- End diff --
Those two error messages say "No stack traces are available", while the
actual error I'm pointing out is that the executor may not exist at all. The
HTTP error code is the same, it's just that the message is misleading.
e.g. if you have 5 executors and ask for the stack trace from executor
1000. Yeah, technically, "No stack traces are available" is correct, it's just
that we can do better.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user attilapiros commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r165739588
--- Diff:
core/src/main/scala/org/apache/spark/status/api/v1/OneApplicationResource.scala
---
@@ -51,6 +52,43 @@ private[v1] class AbstractApplicationResource extends
BaseAppResource {
@Path("executors")
def executorList(): Seq[ExecutorSummary] =
withUI(_.store.executorList(true))
+ @GET
+ @Path("executors/{executorId}/threads")
+ def threadDump(@PathParam("executorId") executorId: String): Response =
withUI { ui =>
+uiRoot.executorThreadDumpsNotAvailableError().getOrElse {
+ val safeExecutorId =
+Option(UIUtils.stripXSS(executorId)).map { executorId =>
+ UIUtils.decodeURLParameter(executorId)
+}.getOrElse {
+ throw new IllegalArgumentException(s"Missing executorId
parameter")
+}
+
+ def isAllDigits(x: String) = x.forall(Character.isDigit)
+
+ if (executorId != SparkContext.DRIVER_IDENTIFIER &&
!isAllDigits(executorId)) {
+Response.serverError()
+ .entity(s"Invalid executorId: neither
'${SparkContext.DRIVER_IDENTIFIER}' nor number.")
+ .status(Response.Status.BAD_REQUEST)
+ .build()
+ } else {
+if
(ui.store.asOption(ui.store.executorSummary(executorId)).exists(!_.isActive)) {
--- End diff --
This is exactly the current behaviour as for false it goes into:
```scala
} else {
ui.sc.flatMap(_.getExecutorThreadDump(safeExecutorId))
.map(Response.ok(_).build())
.getOrElse(
Response.serverError()
.entity("No stack traces are available.")
.status(Response.Status.NOT_FOUND)
.build())
}
```
Then further to:
```scala
.getOrElse(
Response.serverError()
.entity("No stack traces are available.")
.status(Response.Status.NOT_FOUND)
.build())
```
See the test screenshot for "Not existing (well formatted) executor ID".
Or did I misunderstood your requirement?
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r165733278
--- Diff:
core/src/main/scala/org/apache/spark/status/api/v1/OneApplicationResource.scala
---
@@ -51,6 +52,43 @@ private[v1] class AbstractApplicationResource extends
BaseAppResource {
@Path("executors")
def executorList(): Seq[ExecutorSummary] =
withUI(_.store.executorList(true))
+ @GET
+ @Path("executors/{executorId}/threads")
+ def threadDump(@PathParam("executorId") executorId: String): Response =
withUI { ui =>
+uiRoot.executorThreadDumpsNotAvailableError().getOrElse {
+ val safeExecutorId =
+Option(UIUtils.stripXSS(executorId)).map { executorId =>
+ UIUtils.decodeURLParameter(executorId)
+}.getOrElse {
+ throw new IllegalArgumentException(s"Missing executorId
parameter")
+}
+
+ def isAllDigits(x: String) = x.forall(Character.isDigit)
+
+ if (executorId != SparkContext.DRIVER_IDENTIFIER &&
!isAllDigits(executorId)) {
+Response.serverError()
+ .entity(s"Invalid executorId: neither
'${SparkContext.DRIVER_IDENTIFIER}' nor number.")
+ .status(Response.Status.BAD_REQUEST)
+ .build()
+ } else {
+if
(ui.store.asOption(ui.store.executorSummary(executorId)).exists(!_.isActive)) {
--- End diff --
Since that method doesn't throw an exception, perhaps it's not worth it to
try to fix this race nor make the log change.
Also I think the logic here is a little off. If the executor doesn't exist
at all in the store, this condition will not trigger.
```
scala> case class Exec(active: Boolean)
scala> Option[Exec](null).exists(!_.active)
res3: Boolean = false
```
In that situation you should probably return a 404 error (= throw
NotFoundException).
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r165731119
--- Diff:
core/src/main/scala/org/apache/spark/status/api/v1/ApiRootResource.scala ---
@@ -83,6 +83,8 @@ private[spark] trait UIRoot {
def getApplicationInfoList: Iterator[ApplicationInfo]
def getApplicationInfo(appId: String): Option[ApplicationInfo]
+ def executorThreadDumpsNotAvailableError(): Option[Response] = None
--- End diff --
In fact, now that you pointed out the method to get the thread dump is in
`SparkContext`, maybe you don't even need this at all. In the existing resource
code, you can just return the error if `ui.sc` is not defined (e.g. "Thread
dumps are only available in live applications.")
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r165730556
--- Diff:
core/src/main/scala/org/apache/spark/status/api/v1/ApiRootResource.scala ---
@@ -83,6 +83,8 @@ private[spark] trait UIRoot {
def getApplicationInfoList: Iterator[ApplicationInfo]
def getApplicationInfo(appId: String): Option[ApplicationInfo]
+ def executorThreadDumpsNotAvailableError(): Option[Response] = None
--- End diff --
Sure, the method you'd add could have the `SparkUI` as a parameter.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user attilapiros commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r165720298
--- Diff:
core/src/main/scala/org/apache/spark/status/api/v1/OneApplicationResource.scala
---
@@ -51,6 +52,43 @@ private[v1] class AbstractApplicationResource extends
BaseAppResource {
@Path("executors")
def executorList(): Seq[ExecutorSummary] =
withUI(_.store.executorList(true))
+ @GET
+ @Path("executors/{executorId}/threads")
+ def threadDump(@PathParam("executorId") executorId: String): Response =
withUI { ui =>
+uiRoot.executorThreadDumpsNotAvailableError().getOrElse {
+ val safeExecutorId =
+Option(UIUtils.stripXSS(executorId)).map { executorId =>
+ UIUtils.decodeURLParameter(executorId)
+}.getOrElse {
+ throw new IllegalArgumentException(s"Missing executorId
parameter")
+}
+
+ def isAllDigits(x: String) = x.forall(Character.isDigit)
+
+ if (executorId != SparkContext.DRIVER_IDENTIFIER &&
!isAllDigits(executorId)) {
+Response.serverError()
+ .entity(s"Invalid executorId: neither
'${SparkContext.DRIVER_IDENTIFIER}' nor number.")
+ .status(Response.Status.BAD_REQUEST)
+ .build()
+ } else {
+if
(ui.store.asOption(ui.store.executorSummary(executorId)).exists(!_.isActive)) {
--- End diff --
In that case I would suggest to consider changing the log level in
org.apache.spark.SparkContext#getExecutorThreadDump
``` scala
logError(s"Exception getting thread dump from executor
$executorId", e)
```
Maybe to info. What is your opinion?
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user attilapiros commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r165719025
--- Diff:
core/src/main/scala/org/apache/spark/status/api/v1/OneApplicationResource.scala
---
@@ -51,6 +52,43 @@ private[v1] class AbstractApplicationResource extends
BaseAppResource {
@Path("executors")
def executorList(): Seq[ExecutorSummary] =
withUI(_.store.executorList(true))
+ @GET
+ @Path("executors/{executorId}/threads")
+ def threadDump(@PathParam("executorId") executorId: String): Response =
withUI { ui =>
+uiRoot.executorThreadDumpsNotAvailableError().getOrElse {
+ val safeExecutorId =
+Option(UIUtils.stripXSS(executorId)).map { executorId =>
+ UIUtils.decodeURLParameter(executorId)
+}.getOrElse {
+ throw new IllegalArgumentException(s"Missing executorId
parameter")
+}
+
+ def isAllDigits(x: String) = x.forall(Character.isDigit)
+
+ if (executorId != SparkContext.DRIVER_IDENTIFIER &&
!isAllDigits(executorId)) {
+Response.serverError()
+ .entity(s"Invalid executorId: neither
'${SparkContext.DRIVER_IDENTIFIER}' nor number.")
+ .status(Response.Status.BAD_REQUEST)
+ .build()
+ } else {
+if
(ui.store.asOption(ui.store.executorSummary(executorId)).exists(!_.isActive)) {
+ Response.serverError()
+.entity("Executor is already dead.")
+.status(Response.Status.BAD_REQUEST)
+.build()
+} else {
+ ui.sc.flatMap(_.getExecutorThreadDump(safeExecutorId))
--- End diff --
Thanks then I will move ThreadStackTrace to api.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user attilapiros commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r165717844
--- Diff:
core/src/main/scala/org/apache/spark/status/api/v1/ApiRootResource.scala ---
@@ -83,6 +83,8 @@ private[spark] trait UIRoot {
def getApplicationInfoList: Iterator[ApplicationInfo]
def getApplicationInfo(appId: String): Option[ApplicationInfo]
+ def executorThreadDumpsNotAvailableError(): Option[Response] = None
--- End diff --
UIRoot has no access to AppStatusStore and SparkContext can I pass this
parameters along with the executorId to the UIRoot?
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r165711425
--- Diff:
core/src/main/scala/org/apache/spark/status/api/v1/OneApplicationResource.scala
---
@@ -51,6 +52,43 @@ private[v1] class AbstractApplicationResource extends
BaseAppResource {
@Path("executors")
def executorList(): Seq[ExecutorSummary] =
withUI(_.store.executorList(true))
+ @GET
+ @Path("executors/{executorId}/threads")
+ def threadDump(@PathParam("executorId") executorId: String): Response =
withUI { ui =>
+uiRoot.executorThreadDumpsNotAvailableError().getOrElse {
+ val safeExecutorId =
+Option(UIUtils.stripXSS(executorId)).map { executorId =>
+ UIUtils.decodeURLParameter(executorId)
+}.getOrElse {
+ throw new IllegalArgumentException(s"Missing executorId
parameter")
+}
+
+ def isAllDigits(x: String) = x.forall(Character.isDigit)
+
+ if (executorId != SparkContext.DRIVER_IDENTIFIER &&
!isAllDigits(executorId)) {
+Response.serverError()
+ .entity(s"Invalid executorId: neither
'${SparkContext.DRIVER_IDENTIFIER}' nor number.")
+ .status(Response.Status.BAD_REQUEST)
+ .build()
+ } else {
+if
(ui.store.asOption(ui.store.executorSummary(executorId)).exists(!_.isActive)) {
+ Response.serverError()
+.entity("Executor is already dead.")
+.status(Response.Status.BAD_REQUEST)
+.build()
+} else {
+ ui.sc.flatMap(_.getExecutorThreadDump(safeExecutorId))
--- End diff --
This is still exposing an internal type through the public REST API.
You need to create an API type for it and translate between the two, so
that the type returned here is public and checked by mima. Or you could even
just move that type into the api package and make it public.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r165711666
--- Diff:
core/src/main/scala/org/apache/spark/status/api/v1/OneApplicationResource.scala
---
@@ -51,6 +52,43 @@ private[v1] class AbstractApplicationResource extends
BaseAppResource {
@Path("executors")
def executorList(): Seq[ExecutorSummary] =
withUI(_.store.executorList(true))
+ @GET
+ @Path("executors/{executorId}/threads")
+ def threadDump(@PathParam("executorId") executorId: String): Response =
withUI { ui =>
+uiRoot.executorThreadDumpsNotAvailableError().getOrElse {
+ val safeExecutorId =
+Option(UIUtils.stripXSS(executorId)).map { executorId =>
+ UIUtils.decodeURLParameter(executorId)
+}.getOrElse {
+ throw new IllegalArgumentException(s"Missing executorId
parameter")
+}
+
+ def isAllDigits(x: String) = x.forall(Character.isDigit)
+
+ if (executorId != SparkContext.DRIVER_IDENTIFIER &&
!isAllDigits(executorId)) {
+Response.serverError()
+ .entity(s"Invalid executorId: neither
'${SparkContext.DRIVER_IDENTIFIER}' nor number.")
+ .status(Response.Status.BAD_REQUEST)
+ .build()
+ } else {
+if
(ui.store.asOption(ui.store.executorSummary(executorId)).exists(!_.isActive)) {
--- End diff --
There is a minor race here in which the active state can change in between
here and when you get the thread dump below. Can you handle it in the call
below instead (e.g. if it throws an exception when the executor is stopped)?
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r165712466
--- Diff:
core/src/main/scala/org/apache/spark/status/api/v1/ApiRootResource.scala ---
@@ -83,6 +83,8 @@ private[spark] trait UIRoot {
def getApplicationInfoList: Iterator[ApplicationInfo]
def getApplicationInfo(appId: String): Option[ApplicationInfo]
+ def executorThreadDumpsNotAvailableError(): Option[Response] = None
--- End diff --
This feels a little odd. You could instead have a `def threadDump(...):
Response` method and have the implementation of it be in each `UIRoot`
implementation, instead of in the resource class. I think I prefer that a
little bit more over this.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user attilapiros commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r165604504
--- Diff:
core/src/main/scala/org/apache/spark/status/api/v1/OneApplicationResource.scala
---
@@ -51,6 +52,46 @@ private[v1] class AbstractApplicationResource extends
BaseAppResource {
@Path("executors")
def executorList(): Seq[ExecutorSummary] =
withUI(_.store.executorList(true))
+ @GET
+ @Path("executors/{executorId}/threads")
+ def threadDump(@PathParam("executorId") executorId: String): Response =
withUI { ui =>
+uiRoot.executorThreadDumpsNotAvailableError().getOrElse {
+ val safeExecutorId =
+Option(UIUtils.stripXSS(executorId)).map { executorId =>
+ UIUtils.decodeURLParameter(executorId)
+}.getOrElse {
+ throw new IllegalArgumentException(s"Missing executorId
parameter")
+}
+
+ def isAllDigits(x: String) = x.forall(Character.isDigit)
+
+ if (executorId != SparkContext.DRIVER_IDENTIFIER &&
!isAllDigits(executorId)) {
--- End diff --
I don't think so. As I see the app-id is also not case insensitive at the
URL api/v1/applications/{app-id}.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user jerryshao commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r165548039
--- Diff:
core/src/main/scala/org/apache/spark/status/api/v1/OneApplicationResource.scala
---
@@ -51,6 +52,46 @@ private[v1] class AbstractApplicationResource extends
BaseAppResource {
@Path("executors")
def executorList(): Seq[ExecutorSummary] =
withUI(_.store.executorList(true))
+ @GET
+ @Path("executors/{executorId}/threads")
+ def threadDump(@PathParam("executorId") executorId: String): Response =
withUI { ui =>
+uiRoot.executorThreadDumpsNotAvailableError().getOrElse {
+ val safeExecutorId =
+Option(UIUtils.stripXSS(executorId)).map { executorId =>
+ UIUtils.decodeURLParameter(executorId)
+}.getOrElse {
+ throw new IllegalArgumentException(s"Missing executorId
parameter")
+}
+
+ def isAllDigits(x: String) = x.forall(Character.isDigit)
+
+ if (executorId != SparkContext.DRIVER_IDENTIFIER &&
!isAllDigits(executorId)) {
+Response.serverError()
+ .entity(s"Invalid executorId: neither
'${SparkContext.DRIVER_IDENTIFIER}' nor number.")
+ .status(Response.Status.BAD_REQUEST)
+ .build()
+ } else {
+if
(ui.store.asOption(ui.store.executorSummary(executorId)).exists(!_.isActive)) {
+ Response.serverError()
+.entity("Executor is already dead.")
+.status(Response.Status.BAD_REQUEST)
+.build()
+} else {
+ val maybeStackTraces = ui.sc.flatMap { sc =>
+sc.getExecutorThreadDump(safeExecutorId)
+ }
+ maybeStackTraces match {
+case Some(stackTraces) => Response.ok(stackTraces).build()
+case None => Response.serverError()
+ .entity("No stack traces are available.")
+ .status(Response.Status.NOT_FOUND)
+ .build()
--- End diff --
nit: I think the above code can be simplified as: `ui.sc.flatMap { xxx
}.map { xxx }.getOrElse(xxx)`.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user jerryshao commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r165547727
--- Diff:
core/src/main/scala/org/apache/spark/status/api/v1/OneApplicationResource.scala
---
@@ -51,6 +52,46 @@ private[v1] class AbstractApplicationResource extends
BaseAppResource {
@Path("executors")
def executorList(): Seq[ExecutorSummary] =
withUI(_.store.executorList(true))
+ @GET
+ @Path("executors/{executorId}/threads")
+ def threadDump(@PathParam("executorId") executorId: String): Response =
withUI { ui =>
+uiRoot.executorThreadDumpsNotAvailableError().getOrElse {
+ val safeExecutorId =
+Option(UIUtils.stripXSS(executorId)).map { executorId =>
+ UIUtils.decodeURLParameter(executorId)
+}.getOrElse {
+ throw new IllegalArgumentException(s"Missing executorId
parameter")
+}
+
+ def isAllDigits(x: String) = x.forall(Character.isDigit)
+
+ if (executorId != SparkContext.DRIVER_IDENTIFIER &&
!isAllDigits(executorId)) {
--- End diff --
Shall we change the "executorId" to low case before doing the comparison?
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user attilapiros commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r165510071
--- Diff:
core/src/main/scala/org/apache/spark/status/api/v1/OneApplicationResource.scala
---
@@ -51,6 +52,21 @@ private[v1] class AbstractApplicationResource extends
BaseAppResource {
@Path("executors")
def executorList(): Seq[ExecutorSummary] =
withUI(_.store.executorList(true))
+ @GET
+ @Path("executors/{executorId}/threads")
+ def threadDump(@PathParam("executorId") executorId: String):
+ Option[Array[ThreadStackTrace]] = withUI { ui =>
+val safeExecutorId =
+ Option(UIUtils.stripXSS(executorId)).map { executorId =>
+UIUtils.decodeURLParameter(executorId)
+ }.getOrElse {
+throw new IllegalArgumentException(s"Missing executorId parameter")
+ }
+ui.sc.flatMap { sc =>
--- End diff --
fixed by:
Response.serverError()
.entity("Thread dumps not available through the history server.")
.status(Response.Status.SERVICE_UNAVAILABLE)
.build()
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user attilapiros commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r165509830
--- Diff:
core/src/main/scala/org/apache/spark/status/api/v1/OneApplicationResource.scala
---
@@ -51,6 +52,21 @@ private[v1] class AbstractApplicationResource extends
BaseAppResource {
@Path("executors")
def executorList(): Seq[ExecutorSummary] =
withUI(_.store.executorList(true))
+ @GET
+ @Path("executors/{executorId}/threads")
+ def threadDump(@PathParam("executorId") executorId: String):
+ Option[Array[ThreadStackTrace]] = withUI { ui =>
--- End diff --
fixed, now it a response which able to store the error message too
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r165504748
--- Diff:
core/src/main/scala/org/apache/spark/status/api/v1/OneApplicationResource.scala
---
@@ -51,6 +52,21 @@ private[v1] class AbstractApplicationResource extends
BaseAppResource {
@Path("executors")
def executorList(): Seq[ExecutorSummary] =
withUI(_.store.executorList(true))
+ @GET
+ @Path("executors/{executorId}/threads")
+ def threadDump(@PathParam("executorId") executorId: String):
+ Option[Array[ThreadStackTrace]] = withUI { ui =>
--- End diff --
Not a big fan of exposing a private class (`ThreadStackTrace`) in a public
API. (The api here is public even if the scala types aren't.)
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r165504275
--- Diff:
core/src/main/scala/org/apache/spark/status/api/v1/OneApplicationResource.scala
---
@@ -51,6 +52,21 @@ private[v1] class AbstractApplicationResource extends
BaseAppResource {
@Path("executors")
def executorList(): Seq[ExecutorSummary] =
withUI(_.store.executorList(true))
+ @GET
+ @Path("executors/{executorId}/threads")
+ def threadDump(@PathParam("executorId") executorId: String):
+ Option[Array[ThreadStackTrace]] = withUI { ui =>
+val safeExecutorId =
+ Option(UIUtils.stripXSS(executorId)).map { executorId =>
+UIUtils.decodeURLParameter(executorId)
+ }.getOrElse {
+throw new IllegalArgumentException(s"Missing executorId parameter")
+ }
+ui.sc.flatMap { sc =>
--- End diff --
Should an error be thrown if this is called in a replayed app?
GONE/410 sounds like the most appropriate.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user gaborgsomogyi commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r165411432
--- Diff: core/src/main/scala/org/apache/spark/util/Utils.scala ---
@@ -2168,7 +2168,17 @@ private[spark] object Utils extends Logging {
// We need to filter out null values here because dumpAllThreads() may
return null array
// elements for threads that are dead / don't exist.
val threadInfos =
ManagementFactory.getThreadMXBean.dumpAllThreads(true, true).filter(_ != null)
-threadInfos.sortBy(_.getThreadId).map(threadInfoToThreadStackTrace)
--- End diff --
Keeping old behaviour is a good idea this case.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user attilapiros commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r165407125
--- Diff:
core/src/main/scala/org/apache/spark/status/api/v1/OneApplicationResource.scala
---
@@ -51,6 +52,21 @@ private[v1] class AbstractApplicationResource extends
BaseAppResource {
@Path("executors")
def executorList(): Seq[ExecutorSummary] =
withUI(_.store.executorList(true))
+ @GET
+ @Path("executors/{executorId}/threads")
+ def threadDump(@PathParam("executorId") executorId: String):
+ Option[Array[ThreadStackTrace]] = withUI { ui =>
+val safeExecutorId =
+ Option(UIUtils.stripXSS(executorId)).map { executorId =>
+UIUtils.decodeURLParameter(executorId)
+ }.getOrElse {
+throw new IllegalArgumentException(s"Missing executorId parameter")
+ }
+ui.sc.flatMap { sc =>
+ sc.getExecutorThreadDump(safeExecutorId)
--- End diff --
Thanks. Yes, currently it gives back null. You are right an error message
definitely is needed here. Moreover I can check for totally bogus executorId
(neither "driver" nor parsable to integer), finally I can access
ExecutorSummary to decide whether the executor is active. The history server
check is also easy.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user attilapiros commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r165387234
--- Diff: core/src/main/scala/org/apache/spark/util/Utils.scala ---
@@ -2168,7 +2168,17 @@ private[spark] object Utils extends Logging {
// We need to filter out null values here because dumpAllThreads() may
return null array
// elements for threads that are dead / don't exist.
val threadInfos =
ManagementFactory.getThreadMXBean.dumpAllThreads(true, true).filter(_ != null)
-threadInfos.sortBy(_.getThreadId).map(threadInfoToThreadStackTrace)
--- End diff --
Yes, it was. There is a new sortWith right after this line. Of course if
two threads has the same name then earlier the one with the smaller thread ID
was before (assuming sortWith is stable). I can add this extra conditions if it
makes sense, @squito what do you think (how often we have threads with the same
name)?
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user squito commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r165379694
--- Diff:
core/src/main/scala/org/apache/spark/status/api/v1/OneApplicationResource.scala
---
@@ -51,6 +52,21 @@ private[v1] class AbstractApplicationResource extends
BaseAppResource {
@Path("executors")
def executorList(): Seq[ExecutorSummary] =
withUI(_.store.executorList(true))
+ @GET
+ @Path("executors/{executorId}/threads")
+ def threadDump(@PathParam("executorId") executorId: String):
+ Option[Array[ThreadStackTrace]] = withUI { ui =>
+val safeExecutorId =
+ Option(UIUtils.stripXSS(executorId)).map { executorId =>
+UIUtils.decodeURLParameter(executorId)
+ }.getOrElse {
+throw new IllegalArgumentException(s"Missing executorId parameter")
+ }
+ui.sc.flatMap { sc =>
+ sc.getExecutorThreadDump(safeExecutorId)
--- End diff --
what happens if you give a bad executor Id? Looks like you'll just return
an empty response, but I think an error might be more appropriate?
In fact would be nice if the error distinguished between a totally bogus
executorId vs. an executorId which is dead vs. calling this on the history
server, where its totally unavailable. Here's an example of that kind of error
handling for the opposite case, where the endpoint is *only* available on the
history server:
https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/status/api/v1/ApiRootResource.scala?utf8=%E2%9C%93#L90-L95
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user squito commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r165377861
--- Diff:
core/src/main/scala/org/apache/spark/status/api/v1/OneApplicationResource.scala
---
@@ -51,6 +52,21 @@ private[v1] class AbstractApplicationResource extends
BaseAppResource {
@Path("executors")
def executorList(): Seq[ExecutorSummary] =
withUI(_.store.executorList(true))
+ @GET
+ @Path("executors/{executorId}/threads")
+ def threadDump(@PathParam("executorId") executorId: String):
+ Option[Array[ThreadStackTrace]] = withUI { ui =>
--- End diff --
if this doesn't fit on oneline, the style is to still put each param on its
own line
```scala
def threadDump(
@PathParam("executorId") executorId: String):
Option[Array[ThreadStackTrace]] = ...
```
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
Github user gaborgsomogyi commented on a diff in the pull request:
https://github.com/apache/spark/pull/20474#discussion_r165377643
--- Diff: core/src/main/scala/org/apache/spark/util/Utils.scala ---
@@ -2168,7 +2168,17 @@ private[spark] object Utils extends Logging {
// We need to filter out null values here because dumpAllThreads() may
return null array
// elements for threads that are dead / don't exist.
val threadInfos =
ManagementFactory.getThreadMXBean.dumpAllThreads(true, true).filter(_ != null)
-threadInfos.sortBy(_.getThreadId).map(threadInfoToThreadStackTrace)
--- End diff --
`sortBy(_.getThreadId)` just disappeared. Is it intentional?
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20474: [SPARK-23235][Core] Add executor Threaddump to ap...
GitHub user attilapiros opened a pull request:
https://github.com/apache/spark/pull/20474
[SPARK-23235][Core] Add executor Threaddump to api
## What changes were proposed in this pull request?
Extending api with the executor thread dump data.
For this new REST URL is introduced:
- GET
http://localhost:4040/api/v1/applications/{applicationId}/executors/{executorId}/threads
Example response:
``` javascript
[ {
"threadId" : 52,
"threadName" : "context-cleaner-periodic-gc",
"threadState" : "TIMED_WAITING",
"stackTrace" : "sun.misc.Unsafe.park(Native
Method)\njava.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)\njava.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)\njava.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1093)\njava.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)\njava.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1074)\njava.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1134)\njava.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)\njava.lang.Thread.run(Thread.java:748)",
"blockedByLock" :
"Lock(java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject@1385411893})",
"holdingLocks" : [ ]
}, {
"threadId" : 48,
"threadName" : "dag-scheduler-event-loop",
"threadState" : "WAITING",
"stackTrace" : "sun.misc.Unsafe.park(Native
Method)\njava.util.concurrent.locks.LockSupport.park(LockSupport.java:175)\njava.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2039)\njava.util.concurrent.LinkedBlockingDeque.takeFirst(LinkedBlockingDeque.java:492)\njava.util.concurrent.LinkedBlockingDeque.take(LinkedBlockingDeque.java:680)\norg.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:46)",
"blockedByLock" :
"Lock(java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject@1138053349})",
"holdingLocks" : [ ]
}, {
"threadId" : 17,
"threadName" : "dispatcher-event-loop-0",
"threadState" : "WAITING",
"stackTrace" : "sun.misc.Unsafe.park(Native
Method)\njava.util.concurrent.locks.LockSupport.park(LockSupport.java:175)\njava.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2039)\njava.util.concurrent.LinkedBlockingQueue.take(LinkedBlockingQueue.java:442)\norg.apache.spark.rpc.netty.Dispatcher$MessageLoop.run(Dispatcher.scala:215)\njava.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)\njava.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)\njava.lang.Thread.run(Thread.java:748)",
"blockedByLock" :
"Lock(java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject@1764626380})",
"holdingLocks" : [
"Lock(java.util.concurrent.ThreadPoolExecutor$Worker@832743930})" ]
}, {
"threadId" : 18,
"threadName" : "dispatcher-event-loop-1",
"threadState" : "WAITING",
"stackTrace" : "sun.misc.Unsafe.park(Native
Method)\njava.util.concurrent.locks.LockSupport.park(LockSupport.java:175)\njava.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2039)\njava.util.concurrent.LinkedBlockingQueue.take(LinkedBlockingQueue.java:442)\norg.apache.spark.rpc.netty.Dispatcher$MessageLoop.run(Dispatcher.scala:215)\njava.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)\njava.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)\njava.lang.Thread.run(Thread.java:748)",
"blockedByLock" :
"Lock(java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject@1764626380})",
"holdingLocks" : [
"Lock(java.util.concurrent.ThreadPoolExecutor$Worker@834153999})" ]
}, {
"threadId" : 19,
"threadName" : "dispatcher-event-loop-2",
"threadState" : "WAITING",
"stackTrace" : "sun.misc.Unsafe.park(Native
Method)\njava.util.concurrent.locks.LockSupport.park(LockSupport.java:175)\njava.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2039)\njava.util.concurrent.LinkedBlockingQueue.take(LinkedBlockingQueue.java:442)\norg.apache.spark.rpc.netty.Dispatcher$MessageLoop.run(Dispatcher.scala:215)\njava.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)\njava.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)\njava.lang.Thread.run(Thread.java:748)",
"blockedByLock" :
"Lock(java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject@1764626380})",
"holdingLocks" : [
"Lock(java.util.concurrent.ThreadPoolExecutor$Worker@664836465})" ]
}, {
"threadId
