This is an automated email from the ASF dual-hosted git repository.
tgraves pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 70f4bbc [SPARK-28414][WEBUI] UI updates to show resource info in
Standalone
70f4bbc is described below
commit 70f4bbccc511c60266511c752a1fa37b20be3f8d
Author: wuyi <ngone_5...@163.com>
AuthorDate: Tue Aug 27 08:59:29 2019 -0500
[SPARK-28414][WEBUI] UI updates to show resource info in Standalone
## What changes were proposed in this pull request?
Since SPARK-27371 has supported GPU-aware resource scheduling in
Standalone, this PR adds resources info in Standalone UI.
## How was this patch tested?
Updated `JsonProtocolSuite` and tested manually.
Master page:
Worker page
Application page
Closes #25409 from Ngone51/SPARK-28414.
Authored-by: wuyi <ngone_5...@163.com>
Signed-off-by: Thomas Graves <tgra...@apache.org>
---
.../org/apache/spark/deploy/DeployMessage.scala | 4 +-
.../org/apache/spark/deploy/JsonProtocol.scala | 39 +++++++++++-
.../spark/deploy/StandaloneResourceUtils.scala | 74 ++++++++++++++++++++++
.../org/apache/spark/deploy/master/Master.scala | 4 +-
.../apache/spark/deploy/master/WorkerInfo.scala | 29 +++++++--
.../spark/deploy/master/ui/ApplicationPage.scala | 9 ++-
.../apache/spark/deploy/master/ui/MasterPage.scala | 41 ++++++++++--
.../org/apache/spark/deploy/worker/Worker.scala | 25 +++++++-
.../apache/spark/deploy/worker/ui/WorkerPage.scala | 28 +++++++-
.../apache/spark/resource/ResourceAllocator.scala | 1 -
.../spark/resource/ResourceInformation.scala | 2 +
.../org/apache/spark/deploy/DeployTestUtils.scala | 40 ++++++++++--
.../apache/spark/deploy/JsonProtocolSuite.scala | 37 +++++++++--
13 files changed, 298 insertions(+), 35 deletions(-)
diff --git a/core/src/main/scala/org/apache/spark/deploy/DeployMessage.scala
b/core/src/main/scala/org/apache/spark/deploy/DeployMessage.scala
index 3f1d1ae..fba371d 100644
--- a/core/src/main/scala/org/apache/spark/deploy/DeployMessage.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/DeployMessage.scala
@@ -238,7 +238,9 @@ private[deploy] object DeployMessages {
case class WorkerStateResponse(host: String, port: Int, workerId: String,
executors: List[ExecutorRunner], finishedExecutors: List[ExecutorRunner],
drivers: List[DriverRunner], finishedDrivers: List[DriverRunner],
masterUrl: String,
- cores: Int, memory: Int, coresUsed: Int, memoryUsed: Int, masterWebUiUrl:
String) {
+ cores: Int, memory: Int, coresUsed: Int, memoryUsed: Int, masterWebUiUrl:
String,
+ resources: Map[String, ResourceInformation] = Map.empty,
+ resourcesUsed: Map[String, ResourceInformation] = Map.empty) {
Utils.checkHost(host)
assert (port > 0)
diff --git a/core/src/main/scala/org/apache/spark/deploy/JsonProtocol.scala
b/core/src/main/scala/org/apache/spark/deploy/JsonProtocol.scala
index 7212696..6c3276c 100644
--- a/core/src/main/scala/org/apache/spark/deploy/JsonProtocol.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/JsonProtocol.scala
@@ -17,15 +17,29 @@
package org.apache.spark.deploy
-import org.json4s.JsonAST.JObject
+import org.json4s.JsonAST._
import org.json4s.JsonDSL._
import org.apache.spark.deploy.DeployMessages.{MasterStateResponse,
WorkerStateResponse}
import org.apache.spark.deploy.master._
import org.apache.spark.deploy.master.RecoveryState.MasterState
import org.apache.spark.deploy.worker.ExecutorRunner
+import org.apache.spark.resource.{ResourceInformation, ResourceRequirement}
private[deploy] object JsonProtocol {
+
+ private def writeResourcesInfo(info: Map[String, ResourceInformation]):
JObject = {
+ val jsonFields = info.map {
+ case (k, v) => JField(k, v.toJson)
+ }
+ JObject(jsonFields.toList)
+ }
+
+ private def writeResourceRequirement(req: ResourceRequirement): JObject = {
+ ("name" -> req.resourceName) ~
+ ("amount" -> req.amount)
+ }
+
/**
* Export the [[WorkerInfo]] to a Json object. A [[WorkerInfo]] consists of
the information of a
* worker.
@@ -41,6 +55,9 @@ private[deploy] object JsonProtocol {
* `memory` total memory of the worker
* `memoryused` allocated memory of the worker
* `memoryfree` free memory of the worker
+ * `resources` total resources of the worker
+ * `resourcesused` allocated resources of the worker
+ * `resourcesfree` free resources of the worker
* `state` state of the worker, see [[WorkerState]]
* `lastheartbeat` time in milliseconds that the latest heart beat
message from the
* worker is received
@@ -56,6 +73,9 @@ private[deploy] object JsonProtocol {
("memory" -> obj.memory) ~
("memoryused" -> obj.memoryUsed) ~
("memoryfree" -> obj.memoryFree) ~
+ ("resources" -> writeResourcesInfo(obj.resourcesInfo)) ~
+ ("resourcesused" -> writeResourcesInfo(obj.resourcesInfoUsed)) ~
+ ("resourcesfree" -> writeResourcesInfo(obj.resourcesInfoFree)) ~
("state" -> obj.state.toString) ~
("lastheartbeat" -> obj.lastHeartbeat)
}
@@ -71,6 +91,7 @@ private[deploy] object JsonProtocol {
* `cores` total cores granted to the application
* `user` name of the user who submitted the application
* `memoryperslave` minimal memory in MB required to each executor
+ * `resourcesperslave` minimal resources required to each executor
* `submitdate` time in Date that the application is submitted
* `state` state of the application, see [[ApplicationState]]
* `duration` time in milliseconds that the application has been
running
@@ -82,6 +103,8 @@ private[deploy] object JsonProtocol {
("cores" -> obj.coresGranted) ~
("user" -> obj.desc.user) ~
("memoryperslave" -> obj.desc.memoryPerExecutorMB) ~
+ ("resourcesperslave" -> obj.desc.resourceReqsPerExecutor
+ .toList.map(writeResourceRequirement)) ~
("submitdate" -> obj.submitDate.toString) ~
("state" -> obj.state.toString) ~
("duration" -> obj.duration)
@@ -95,6 +118,7 @@ private[deploy] object JsonProtocol {
* `name` the description of the application
* `cores` max cores that can be allocated to the application, 0
means unlimited
* `memoryperslave` minimal memory in MB required to each executor
+ * `resourcesperslave` minimal resources required to each executor
* `user` name of the user who submitted the application
* `command` the command string used to submit the application
*/
@@ -102,6 +126,7 @@ private[deploy] object JsonProtocol {
("name" -> obj.name) ~
("cores" -> obj.maxCores.getOrElse(0)) ~
("memoryperslave" -> obj.memoryPerExecutorMB) ~
+ ("resourcesperslave" ->
obj.resourceReqsPerExecutor.toList.map(writeResourceRequirement)) ~
("user" -> obj.user) ~
("command" -> obj.command.toString)
}
@@ -113,6 +138,7 @@ private[deploy] object JsonProtocol {
* @return a Json object containing the following fields:
* `id` an integer identifier of the executor
* `memory` memory in MB allocated to the executor
+ * `resources` resources allocated to the executor
* `appid` a string identifier of the application that the executor
is working on
* `appdesc` a Json object of the [[ApplicationDescription]] of the
application that the
* executor is working on
@@ -120,6 +146,7 @@ private[deploy] object JsonProtocol {
def writeExecutorRunner(obj: ExecutorRunner): JObject = {
("id" -> obj.execId) ~
("memory" -> obj.memory) ~
+ ("resources" -> writeResourcesInfo(obj.resources)) ~
("appid" -> obj.appId) ~
("appdesc" -> writeApplicationDescription(obj.appDesc))
}
@@ -134,6 +161,7 @@ private[deploy] object JsonProtocol {
* `state` state of the driver, see [[DriverState]]
* `cores` cores allocated to the driver
* `memory` memory in MB allocated to the driver
+ * `resources` resources allocated to the driver
* `submitdate` time in Date that the driver is created
* `worker` identifier of the worker that the driver is running on
* `mainclass` main class of the command string that started the
driver
@@ -144,6 +172,7 @@ private[deploy] object JsonProtocol {
("state" -> obj.state.toString) ~
("cores" -> obj.desc.cores) ~
("memory" -> obj.desc.mem) ~
+ ("resources" -> writeResourcesInfo(obj.resources)) ~
("submitdate" -> obj.submitDate.toString) ~
("worker" -> obj.worker.map(_.id).getOrElse("None")) ~
("mainclass" -> obj.desc.command.arguments(2))
@@ -162,6 +191,8 @@ private[deploy] object JsonProtocol {
* `coresused` cores used by the master
* `memory` total memory available of the master
* `memoryused` memory used by the master
+ * `resources` total resources available of the master
+ * `resourcesused` resources used by the master
* `activeapps` a list of Json objects of [[ApplicationInfo]] of the
active applications
* running on the master
* `completedapps` a list of Json objects of [[ApplicationInfo]] of
the applications
@@ -181,6 +212,8 @@ private[deploy] object JsonProtocol {
("coresused" -> aliveWorkers.map(_.coresUsed).sum) ~
("memory" -> aliveWorkers.map(_.memory).sum) ~
("memoryused" -> aliveWorkers.map(_.memoryUsed).sum) ~
+ ("resources" ->
aliveWorkers.map(_.resourcesInfo).toList.map(writeResourcesInfo)) ~
+ ("resourcesused" ->
aliveWorkers.map(_.resourcesInfoUsed).toList.map(writeResourcesInfo)) ~
("activeapps" -> obj.activeApps.toList.map(writeApplicationInfo)) ~
("completedapps" -> obj.completedApps.toList.map(writeApplicationInfo)) ~
("activedrivers" -> obj.activeDrivers.toList.map(writeDriverInfo)) ~
@@ -200,6 +233,8 @@ private[deploy] object JsonProtocol {
* `coreused` used cores of the worker
* `memory` total memory of the worker
* `memoryused` used memory of the worker
+ * `resources` total resources of the worker
+ * `resourcesused` used resources of the worker
* `executors` a list of Json objects of [[ExecutorRunner]] of the
executors running on
* the worker
* `finishedexecutors` a list of Json objects of [[ExecutorRunner]]
of the finished
@@ -213,6 +248,8 @@ private[deploy] object JsonProtocol {
("coresused" -> obj.coresUsed) ~
("memory" -> obj.memory) ~
("memoryused" -> obj.memoryUsed) ~
+ ("resources" -> writeResourcesInfo(obj.resources)) ~
+ ("resourcesused" -> writeResourcesInfo(obj.resourcesUsed)) ~
("executors" -> obj.executors.map(writeExecutorRunner)) ~
("finishedexecutors" -> obj.finishedExecutors.map(writeExecutorRunner))
}
diff --git
a/core/src/main/scala/org/apache/spark/deploy/StandaloneResourceUtils.scala
b/core/src/main/scala/org/apache/spark/deploy/StandaloneResourceUtils.scala
index b64a36f..d6f9618a 100644
--- a/core/src/main/scala/org/apache/spark/deploy/StandaloneResourceUtils.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/StandaloneResourceUtils.scala
@@ -43,6 +43,37 @@ private[spark] object StandaloneResourceUtils extends
Logging {
val RESOURCES_LOCK_FILE = "__allocated_resources__.lock"
/**
+ * A mutable resource information which provides more efficient modification
on addresses.
+ */
+ private[spark] case class MutableResourceInfo(name: String, addresses:
mutable.HashSet[String]) {
+
+ def + (other: MutableResourceInfo): this.type = {
+ assert(name == other.name, s"Inconsistent resource name, expected $name,
" +
+ s"but got ${other.name}")
+ other.addresses.foreach(this.addresses.add)
+ this
+ }
+
+ def + (other: ResourceInformation): this.type = {
+ assert(name == other.name, s"Inconsistent resource name, expected $name,
" +
+ s"but got ${other.name}")
+ other.addresses.foreach(this.addresses.add)
+ this
+ }
+
+ def - (other: ResourceInformation): this.type = {
+ assert(name == other.name, s"Inconsistent resource name, expected $name,
" +
+ s"but got ${other.name}")
+ other.addresses.foreach(this.addresses.remove)
+ this
+ }
+
+ def toResourceInformation: ResourceInformation = {
+ new ResourceInformation(name, addresses.toArray)
+ }
+ }
+
+ /**
* Resource allocation used in Standalone only, which tracks assignments with
* worker/driver(client only) pid.
*/
@@ -345,4 +376,47 @@ private[spark] object StandaloneResourceUtils extends
Logging {
def needCoordinate(conf: SparkConf): Boolean = {
conf.get(SPARK_RESOURCES_COORDINATE)
}
+
+ def toMutable(immutableResources: Map[String, ResourceInformation])
+ : Map[String, MutableResourceInfo] = {
+ immutableResources.map { case (rName, rInfo) =>
+ val mutableAddress = new mutable.HashSet[String]()
+ mutableAddress ++= rInfo.addresses
+ rName -> MutableResourceInfo(rInfo.name, mutableAddress)
+ }
+ }
+
+ // used for UI
+ def formatResourcesDetails(
+ usedInfo: Map[String, ResourceInformation],
+ freeInfo: Map[String, ResourceInformation]): String = {
+ usedInfo.map { case (rName, rInfo) =>
+ val used = rInfo.addresses.mkString("[", ", ", "]")
+ val free = freeInfo(rName).addresses.mkString("[", ", ", "]")
+ s"$rName: Free: $free / Used: $used"
+ }.mkString(", ")
+ }
+
+ // used for UI
+ def formatResourcesAddresses(resources: Map[String, ResourceInformation]):
String = {
+ resources.map { case (rName, rInfo) =>
+ s"$rName: ${rInfo.addresses.mkString("[", ", ", "]")}"
+ }.mkString(", ")
+ }
+
+ // used for UI
+ def formatResourcesUsed(
+ resourcesTotal: Map[String, ResourceInformation],
+ resourcesUsed: Map[String, ResourceInformation]): String = {
+ resourcesTotal.map { case (rName, rInfo) =>
+ val used = resourcesUsed(rName).addresses.length
+ val total = rInfo.addresses.length
+ s"$used / $total $rName"
+ }.mkString(", ")
+ }
+
+ // used for UI
+ def formatResourceRequirements(requirements: Seq[ResourceRequirement]):
String = {
+ requirements.map(req => s"${req.amount} ${req.resourceName}").mkString(",
")
+ }
}
diff --git a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala
b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala
index 6765519..5588dc8 100644
--- a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala
@@ -644,7 +644,7 @@ private[deploy] class Master(
val assignedResources = resourceReqsPerExecutor.map {
req => req.resourceName -> req.amount * assignedExecutorNum
}.toMap
- val resourcesFree = usableWorkers(pos).resourcesFree.map {
+ val resourcesFree = usableWorkers(pos).resourcesAmountFree.map {
case (rName, free) => rName -> (free -
assignedResources.getOrElse(rName, 0))
}
val enoughResources = ResourceUtils.resourcesMeetRequirements(
@@ -752,7 +752,7 @@ private[deploy] class Master(
val enoughMem = worker.memoryFree >= memoryReq
val enoughCores = worker.coresFree >= coresReq
val enoughResources = ResourceUtils.resourcesMeetRequirements(
- worker.resourcesFree, resourceRequirements)
+ worker.resourcesAmountFree, resourceRequirements)
enoughMem && enoughCores && enoughResources
}
diff --git
a/core/src/main/scala/org/apache/spark/deploy/master/WorkerInfo.scala
b/core/src/main/scala/org/apache/spark/deploy/master/WorkerInfo.scala
index d485db4..9509844 100644
--- a/core/src/main/scala/org/apache/spark/deploy/master/WorkerInfo.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/master/WorkerInfo.scala
@@ -18,7 +18,9 @@
package org.apache.spark.deploy.master
import scala.collection.mutable
+import scala.reflect.ClassTag
+import org.apache.spark.deploy.StandaloneResourceUtils.MutableResourceInfo
import org.apache.spark.resource.{ResourceAllocator, ResourceInformation,
ResourceRequirement}
import org.apache.spark.rpc.RpcEndpointRef
import org.apache.spark.util.Utils
@@ -26,10 +28,6 @@ import org.apache.spark.util.Utils
private[spark] case class WorkerResourceInfo(name: String, addresses:
Seq[String])
extends ResourceAllocator(name, addresses) {
- def toResourceInformation(): ResourceInformation = {
- new ResourceInformation(name, addresses.toArray)
- }
-
def acquire(amount: Int): ResourceInformation = {
val allocated = availableAddrs.take(amount)
acquire(allocated)
@@ -45,8 +43,7 @@ private[spark] class WorkerInfo(
val memory: Int,
val endpoint: RpcEndpointRef,
val webUiAddress: String,
- val resources: Map[String, WorkerResourceInfo],
- val pid: Int = 0)
+ val resources: Map[String, WorkerResourceInfo])
extends Serializable {
Utils.checkHost(host)
@@ -64,12 +61,30 @@ private[spark] class WorkerInfo(
def coresFree: Int = cores - coresUsed
def memoryFree: Int = memory - memoryUsed
- def resourcesFree: Map[String, Int] = {
+ def resourcesAmountFree: Map[String, Int] = {
resources.map { case (rName, rInfo) =>
rName -> rInfo.availableAddrs.length
}
}
+ def resourcesInfo: Map[String, ResourceInformation] = {
+ resources.map { case (rName, rInfo) =>
+ rName -> new ResourceInformation(rName, rInfo.addresses.toArray)
+ }
+ }
+
+ def resourcesInfoFree: Map[String, ResourceInformation] = {
+ resources.map { case (rName, rInfo) =>
+ rName -> new ResourceInformation(rName, rInfo.availableAddrs.toArray)
+ }
+ }
+
+ def resourcesInfoUsed: Map[String, ResourceInformation] = {
+ resources.map { case (rName, rInfo) =>
+ rName -> new ResourceInformation(rName, rInfo.assignedAddrs.toArray)
+ }
+ }
+
private def readObject(in: java.io.ObjectInputStream): Unit =
Utils.tryOrIOException {
in.defaultReadObject()
init()
diff --git
a/core/src/main/scala/org/apache/spark/deploy/master/ui/ApplicationPage.scala
b/core/src/main/scala/org/apache/spark/deploy/master/ui/ApplicationPage.scala
index bcd7a7e..c7e73bc 100644
---
a/core/src/main/scala/org/apache/spark/deploy/master/ui/ApplicationPage.scala
+++
b/core/src/main/scala/org/apache/spark/deploy/master/ui/ApplicationPage.scala
@@ -23,6 +23,7 @@ import scala.xml.Node
import org.apache.spark.deploy.DeployMessages.{MasterStateResponse,
RequestMasterState}
import org.apache.spark.deploy.ExecutorState
+import
org.apache.spark.deploy.StandaloneResourceUtils.{formatResourceRequirements,
formatResourcesAddresses}
import org.apache.spark.deploy.master.ExecutorDesc
import org.apache.spark.ui.{ToolTips, UIUtils, WebUIPage}
import org.apache.spark.util.Utils
@@ -42,7 +43,8 @@ private[ui] class ApplicationPage(parent: MasterWebUI)
extends WebUIPage("app")
return UIUtils.basicSparkPage(request, msg, "Not Found")
}
- val executorHeaders = Seq("ExecutorID", "Worker", "Cores", "Memory",
"State", "Logs")
+ val executorHeaders = Seq("ExecutorID", "Worker", "Cores", "Memory",
"Resources",
+ "State", "Logs")
val allExecutors = (app.executors.values ++
app.removedExecutors).toSet.toSeq
// This includes executors that are either still running or have exited
cleanly
val executors = allExecutors.filter { exec =>
@@ -83,6 +85,10 @@ private[ui] class ApplicationPage(parent: MasterWebUI)
extends WebUIPage("app")
<strong>Executor Memory:</strong>
{Utils.megabytesToString(app.desc.memoryPerExecutorMB)}
</li>
+ <li>
+ <strong>Executor Resources:</strong>
+ {formatResourceRequirements(app.desc.resourceReqsPerExecutor)}
+ </li>
<li><strong>Submit Date:</strong>
{UIUtils.formatDate(app.submitDate)}</li>
<li><strong>State:</strong> {app.state}</li>
{
@@ -139,6 +145,7 @@ private[ui] class ApplicationPage(parent: MasterWebUI)
extends WebUIPage("app")
</td>
<td>{executor.cores}</td>
<td>{executor.memory}</td>
+ <td>{formatResourcesAddresses(executor.resources)}</td>
<td>{executor.state}</td>
<td>
<a
href={s"$workerUrlRef/logPage?appId=${executor.application.id}&executorId=${executor.
diff --git
a/core/src/main/scala/org/apache/spark/deploy/master/ui/MasterPage.scala
b/core/src/main/scala/org/apache/spark/deploy/master/ui/MasterPage.scala
index 6701465..e8b6145 100644
--- a/core/src/main/scala/org/apache/spark/deploy/master/ui/MasterPage.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/master/ui/MasterPage.scala
@@ -25,7 +25,9 @@ import org.json4s.JValue
import org.apache.spark.deploy.DeployMessages.{KillDriverResponse,
MasterStateResponse, RequestKillDriver, RequestMasterState}
import org.apache.spark.deploy.JsonProtocol
+import org.apache.spark.deploy.StandaloneResourceUtils._
import org.apache.spark.deploy.master._
+import org.apache.spark.resource.ResourceInformation
import org.apache.spark.ui.{UIUtils, WebUIPage}
import org.apache.spark.util.Utils
@@ -67,24 +69,48 @@ private[ui] class MasterPage(parent: MasterWebUI) extends
WebUIPage("") {
}
}
+ private def formatWorkerResourcesDetails(worker: WorkerInfo): String = {
+ val usedInfo = worker.resourcesInfoUsed
+ val freeInfo = worker.resourcesInfoFree
+ formatResourcesDetails(usedInfo, freeInfo)
+ }
+
+ private def formatMasterResourcesInUse(aliveWorkers: Array[WorkerInfo]):
String = {
+ val totalInfo = aliveWorkers.map(_.resourcesInfo)
+ .map(resources => toMutable(resources))
+ .flatMap(_.toIterator)
+ .groupBy(_._1) // group by resource name
+ .map { case (rName, rInfoArr) =>
+ rName -> rInfoArr.map(_._2).reduce(_ + _)
+ }.map { case (k, v) => (k, v.toResourceInformation) }
+ val usedInfo = aliveWorkers.map(_.resourcesInfoUsed)
+ .map (resources => toMutable(resources))
+ .flatMap(_.toIterator)
+ .groupBy(_._1) // group by resource name
+ .map { case (rName, rInfoArr) =>
+ rName -> rInfoArr.map(_._2).reduce(_ + _)
+ }.map { case (k, v) => (k, v.toResourceInformation) }
+ formatResourcesUsed(totalInfo, usedInfo)
+ }
+
/** Index view listing applications and executors */
def render(request: HttpServletRequest): Seq[Node] = {
val state = getMasterState
- val workerHeaders = Seq("Worker Id", "Address", "State", "Cores", "Memory")
+ val workerHeaders = Seq("Worker Id", "Address", "State", "Cores",
"Memory", "Resources")
val workers = state.workers.sortBy(_.id)
val aliveWorkers = state.workers.filter(_.state == WorkerState.ALIVE)
val workerTable = UIUtils.listingTable(workerHeaders, workerRow, workers)
- val appHeaders = Seq("Application ID", "Name", "Cores", "Memory per
Executor", "Submitted Time",
- "User", "State", "Duration")
+ val appHeaders = Seq("Application ID", "Name", "Cores", "Memory per
Executor",
+ "Resources Per Executor", "Submitted Time", "User", "State", "Duration")
val activeApps = state.activeApps.sortBy(_.startTime).reverse
val activeAppsTable = UIUtils.listingTable(appHeaders, appRow, activeApps)
val completedApps = state.completedApps.sortBy(_.endTime).reverse
val completedAppsTable = UIUtils.listingTable(appHeaders, appRow,
completedApps)
val driverHeaders = Seq("Submission ID", "Submitted Time", "Worker",
"State", "Cores",
- "Memory", "Main Class")
+ "Memory", "Resources", "Main Class")
val activeDrivers = state.activeDrivers.sortBy(_.startTime).reverse
val activeDriversTable = UIUtils.listingTable(driverHeaders, driverRow,
activeDrivers)
val completedDrivers = state.completedDrivers.sortBy(_.startTime).reverse
@@ -113,6 +139,8 @@ private[ui] class MasterPage(parent: MasterWebUI) extends
WebUIPage("") {
<li><strong>Memory in use:</strong>
{Utils.megabytesToString(aliveWorkers.map(_.memory).sum)}
Total,
{Utils.megabytesToString(aliveWorkers.map(_.memoryUsed).sum)}
Used</li>
+ <li><strong>Resources in use:</strong>
+ {formatMasterResourcesInUse(aliveWorkers)}</li>
<li><strong>Applications:</strong>
{state.activeApps.length} <a href="#running-app">Running</a>,
{state.completedApps.length} <a
href="#completed-app">Completed</a> </li>
@@ -236,6 +264,7 @@ private[ui] class MasterPage(parent: MasterWebUI) extends
WebUIPage("") {
{Utils.megabytesToString(worker.memory)}
({Utils.megabytesToString(worker.memoryUsed)} Used)
</td>
+ <td>{formatWorkerResourcesDetails(worker)}</td>
</tr>
}
@@ -272,6 +301,9 @@ private[ui] class MasterPage(parent: MasterWebUI) extends
WebUIPage("") {
<td sorttable_customkey={app.desc.memoryPerExecutorMB.toString}>
{Utils.megabytesToString(app.desc.memoryPerExecutorMB)}
</td>
+ <td>
+ {formatResourceRequirements(app.desc.resourceReqsPerExecutor)}
+ </td>
<td>{UIUtils.formatDate(app.submitDate)}</td>
<td>{app.desc.user}</td>
<td>{app.state.toString}</td>
@@ -312,6 +344,7 @@ private[ui] class MasterPage(parent: MasterWebUI) extends
WebUIPage("") {
<td sorttable_customkey={driver.desc.mem.toString}>
{Utils.megabytesToString(driver.desc.mem.toLong)}
</td>
+ <td>{formatResourcesAddresses(driver.resources)}</td>
<td>{driver.desc.command.arguments(2)}</td>
</tr>
}
diff --git a/core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala
b/core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala
index 899593d..3731b6a 100755
--- a/core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala
@@ -34,7 +34,7 @@ import org.apache.spark.deploy.{Command, ExecutorDescription,
ExecutorState}
import org.apache.spark.deploy.DeployMessages._
import org.apache.spark.deploy.ExternalShuffleService
import org.apache.spark.deploy.StandaloneResourceUtils._
-import org.apache.spark.deploy.master.{DriverState, Master}
+import org.apache.spark.deploy.master.{DriverState, Master, WorkerResourceInfo}
import org.apache.spark.deploy.worker.ui.WorkerWebUI
import org.apache.spark.internal.{config, Logging}
import org.apache.spark.internal.config.Tests.IS_TESTING
@@ -185,6 +185,7 @@ private[deploy] class Worker(
var coresUsed = 0
var memoryUsed = 0
+ val resourcesUsed = new HashMap[String, MutableResourceInfo]()
def coresFree: Int = cores - coresUsed
def memoryFree: Int = memory - memoryUsed
@@ -242,6 +243,21 @@ private[deploy] class Worker(
System.exit(1)
}
}
+ resources.keys.foreach { rName =>
+ resourcesUsed(rName) = MutableResourceInfo(rName, new HashSet[String])
+ }
+ }
+
+ private def addResourcesUsed(deltaInfo: Map[String, ResourceInformation]):
Unit = {
+ deltaInfo.foreach { case (rName, rInfo) =>
+ resourcesUsed(rName) = resourcesUsed(rName) + rInfo
+ }
+ }
+
+ private def removeResourcesUsed(deltaInfo: Map[String,
ResourceInformation]): Unit = {
+ deltaInfo.foreach { case (rName, rInfo) =>
+ resourcesUsed(rName) = resourcesUsed(rName) - rInfo
+ }
}
/**
@@ -588,6 +604,7 @@ private[deploy] class Worker(
manager.start()
coresUsed += cores_
memoryUsed += memory_
+ addResourcesUsed(resources_)
} catch {
case e: Exception =>
logError(s"Failed to launch executor $appId/$execId for
${appDesc.name}.", e)
@@ -634,6 +651,7 @@ private[deploy] class Worker(
coresUsed += driverDesc.cores
memoryUsed += driverDesc.mem
+ addResourcesUsed(resources_)
case KillDriver(driverId) =>
logInfo(s"Asked to kill driver $driverId")
@@ -660,7 +678,8 @@ private[deploy] class Worker(
context.reply(WorkerStateResponse(host, port, workerId,
executors.values.toList,
finishedExecutors.values.toList, drivers.values.toList,
finishedDrivers.values.toList, activeMasterUrl, cores, memory,
- coresUsed, memoryUsed, activeMasterWebUiUrl))
+ coresUsed, memoryUsed, activeMasterWebUiUrl, resources,
+ resourcesUsed.toMap.map { case (k, v) => (k,
v.toResourceInformation)}))
}
override def onDisconnected(remoteAddress: RpcAddress): Unit = {
@@ -773,6 +792,7 @@ private[deploy] class Worker(
trimFinishedDriversIfNecessary()
memoryUsed -= driver.driverDesc.mem
coresUsed -= driver.driverDesc.cores
+ removeResourcesUsed(driver.resources)
}
private[worker] def handleExecutorStateChanged(executorStateChanged:
ExecutorStateChanged):
@@ -794,6 +814,7 @@ private[deploy] class Worker(
trimFinishedExecutorsIfNecessary()
coresUsed -= executor.cores
memoryUsed -= executor.memory
+ removeResourcesUsed(executor.resources)
if (CLEANUP_FILES_AFTER_EXECUTOR_EXIT) {
shuffleService.executorRemoved(executorStateChanged.execId.toString, appId)
diff --git
a/core/src/main/scala/org/apache/spark/deploy/worker/ui/WorkerPage.scala
b/core/src/main/scala/org/apache/spark/deploy/worker/ui/WorkerPage.scala
index aa4e28d..1a4ba06 100644
--- a/core/src/main/scala/org/apache/spark/deploy/worker/ui/WorkerPage.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/worker/ui/WorkerPage.scala
@@ -25,8 +25,10 @@ import org.json4s.JValue
import org.apache.spark.deploy.{ExecutorState, JsonProtocol}
import org.apache.spark.deploy.DeployMessages.{RequestWorkerState,
WorkerStateResponse}
+import
org.apache.spark.deploy.StandaloneResourceUtils.{formatResourcesAddresses,
formatResourcesDetails}
import org.apache.spark.deploy.master.DriverState
import org.apache.spark.deploy.worker.{DriverRunner, ExecutorRunner}
+import org.apache.spark.resource.ResourceInformation
import org.apache.spark.ui.{UIUtils, WebUIPage}
import org.apache.spark.util.Utils
@@ -38,10 +40,25 @@ private[ui] class WorkerPage(parent: WorkerWebUI) extends
WebUIPage("") {
JsonProtocol.writeWorkerState(workerState)
}
+ private def formatWorkerResourcesDetails(workerState: WorkerStateResponse):
String = {
+ val totalInfo = workerState.resources
+ val usedInfo = workerState.resourcesUsed
+ val freeInfo = totalInfo.map { case (rName, rInfo) =>
+ val freeAddresses = if (usedInfo.contains(rName)) {
+ rInfo.addresses.diff(usedInfo(rName).addresses)
+ } else {
+ rInfo.addresses
+ }
+ rName -> new ResourceInformation(rName, freeAddresses)
+ }
+ formatResourcesDetails(usedInfo, freeInfo)
+ }
+
def render(request: HttpServletRequest): Seq[Node] = {
val workerState =
workerEndpoint.askSync[WorkerStateResponse](RequestWorkerState)
- val executorHeaders = Seq("ExecutorID", "Cores", "State", "Memory", "Job
Details", "Logs")
+ val executorHeaders = Seq("ExecutorID", "State", "Cores", "Memory",
"Resources",
+ "Job Details", "Logs")
val runningExecutors = workerState.executors
val runningExecutorTable =
UIUtils.listingTable(executorHeaders, executorRow, runningExecutors)
@@ -49,7 +66,8 @@ private[ui] class WorkerPage(parent: WorkerWebUI) extends
WebUIPage("") {
val finishedExecutorTable =
UIUtils.listingTable(executorHeaders, executorRow, finishedExecutors)
- val driverHeaders = Seq("DriverID", "Main Class", "State", "Cores",
"Memory", "Logs", "Notes")
+ val driverHeaders = Seq("DriverID", "Main Class", "State", "Cores",
"Memory", "Resources",
+ "Logs", "Notes")
val runningDrivers = workerState.drivers.sortBy(_.driverId).reverse
val runningDriverTable = UIUtils.listingTable[DriverRunner](driverHeaders,
driverRow(workerState.workerId, _), runningDrivers)
@@ -71,6 +89,8 @@ private[ui] class WorkerPage(parent: WorkerWebUI) extends
WebUIPage("") {
<li><strong>Cores:</strong> {workerState.cores}
({workerState.coresUsed} Used)</li>
<li><strong>Memory:</strong>
{Utils.megabytesToString(workerState.memory)}
({Utils.megabytesToString(workerState.memoryUsed)} Used)</li>
+ <li><strong>Resources:</strong>
+ {formatWorkerResourcesDetails(workerState)}</li>
</ul>
<p><a href={workerState.masterWebUiUrl}>Back to Master</a></p>
</div>
@@ -147,11 +167,12 @@ private[ui] class WorkerPage(parent: WorkerWebUI) extends
WebUIPage("") {
<tr>
<td>{executor.execId}</td>
- <td>{executor.cores}</td>
<td>{executor.state}</td>
+ <td>{executor.cores}</td>
<td sorttable_customkey={executor.memory.toString}>
{Utils.megabytesToString(executor.memory)}
</td>
+ <td>{formatResourcesAddresses(executor.resources)}</td>
<td>
<ul class="unstyled">
<li><strong>ID:</strong> {executor.appId}</li>
@@ -189,6 +210,7 @@ private[ui] class WorkerPage(parent: WorkerWebUI) extends
WebUIPage("") {
<td sorttable_customkey={driver.driverDesc.mem.toString}>
{Utils.megabytesToString(driver.driverDesc.mem)}
</td>
+ <td>{formatResourcesAddresses(driver.resources)}</td>
<td>
<a
href={s"$workerUrlRef/logPage?driverId=${driver.driverId}&logType=stdout"}>stdout</a>
<a
href={s"$workerUrlRef/logPage?driverId=${driver.driverId}&logType=stderr"}>stderr</a>
diff --git
a/core/src/main/scala/org/apache/spark/resource/ResourceAllocator.scala
b/core/src/main/scala/org/apache/spark/resource/ResourceAllocator.scala
index 719f34db..6f9b9c0 100644
--- a/core/src/main/scala/org/apache/spark/resource/ResourceAllocator.scala
+++ b/core/src/main/scala/org/apache/spark/resource/ResourceAllocator.scala
@@ -45,7 +45,6 @@ class ResourceAllocator(name: String, addresses: Seq[String])
extends Serializab
/**
* Sequence of currently assigned resource addresses.
- * Exposed for testing only.
*/
private[spark] def assignedAddrs: Seq[String] = addressAvailabilityMap
.flatMap { case (addr, available) =>
diff --git
a/core/src/main/scala/org/apache/spark/resource/ResourceInformation.scala
b/core/src/main/scala/org/apache/spark/resource/ResourceInformation.scala
index 96aef74..8b8bd7b 100644
--- a/core/src/main/scala/org/apache/spark/resource/ResourceInformation.scala
+++ b/core/src/main/scala/org/apache/spark/resource/ResourceInformation.scala
@@ -52,6 +52,8 @@ class ResourceInformation(
}
override def hashCode(): Int = Seq(name, addresses.toSeq).hashCode()
+
+ def toJson(): JValue = ResourceInformationJson(name, addresses).toJValue
}
private[spark] object ResourceInformation {
diff --git a/core/src/test/scala/org/apache/spark/deploy/DeployTestUtils.scala
b/core/src/test/scala/org/apache/spark/deploy/DeployTestUtils.scala
index a2c4669..31f065e 100644
--- a/core/src/test/scala/org/apache/spark/deploy/DeployTestUtils.scala
+++ b/core/src/test/scala/org/apache/spark/deploy/DeployTestUtils.scala
@@ -20,8 +20,10 @@ package org.apache.spark.deploy
import java.io.File
import org.apache.spark.{SecurityManager, SparkConf}
-import org.apache.spark.deploy.master.{ApplicationInfo, DriverInfo, WorkerInfo}
+import org.apache.spark.deploy.master.{ApplicationInfo, DriverInfo,
WorkerInfo, WorkerResourceInfo}
import org.apache.spark.deploy.worker.{DriverRunner, ExecutorRunner}
+import org.apache.spark.resource.{ResourceInformation, ResourceRequirement}
+import org.apache.spark.resource.ResourceUtils.{FPGA, GPU}
private[deploy] object DeployTestUtils {
def createAppDesc(): ApplicationDescription = {
@@ -32,7 +34,8 @@ private[deploy] object DeployTestUtils {
def createAppInfo() : ApplicationInfo = {
val appDesc = createAppDesc()
val appInfo = new ApplicationInfo(JsonConstants.appInfoStartTime,
- "id", appDesc, JsonConstants.submitDate, null, Int.MaxValue)
+ "id", appDesc.copy(resourceReqsPerExecutor = createResourceRequirement),
+ JsonConstants.submitDate, null, Int.MaxValue)
appInfo.endTime = JsonConstants.currTimeInMillis
appInfo
}
@@ -45,17 +48,29 @@ private[deploy] object DeployTestUtils {
def createDriverDesc(): DriverDescription =
new DriverDescription("hdfs://some-dir/some.jar", 100, 3, false,
createDriverCommand())
- def createDriverInfo(): DriverInfo = new DriverInfo(3, "driver-3",
- createDriverDesc(), JsonConstants.submitDate)
+ def createDriverInfo(): DriverInfo = {
+ val dDesc = createDriverDesc().copy(resourceReqs =
createResourceRequirement)
+ val dInfo = new DriverInfo(3, "driver-3", dDesc, JsonConstants.submitDate)
+ dInfo.withResources(createResourceInformation)
+ dInfo
+ }
def createWorkerInfo(): WorkerInfo = {
+ val gpuResource = new WorkerResourceInfo(GPU, Seq("0", "1", "2"))
+ val fpgaResource = new WorkerResourceInfo(FPGA, Seq("3", "4", "5"))
+ val resources = Map(GPU -> gpuResource, FPGA -> fpgaResource)
val workerInfo = new WorkerInfo("id", "host", 8080, 4, 1234, null,
- "http://publicAddress:80";, Map.empty)
+ "http://publicAddress:80";, resources)
workerInfo.lastHeartbeat = JsonConstants.currTimeInMillis
workerInfo
}
- def createExecutorRunner(execId: Int): ExecutorRunner = {
+ def createExecutorRunner(execId: Int, withResources: Boolean = false):
ExecutorRunner = {
+ val resources = if (withResources) {
+ createResourceInformation
+ } else {
+ Map.empty[String, ResourceInformation]
+ }
new ExecutorRunner(
"appId",
execId,
@@ -73,7 +88,8 @@ private[deploy] object DeployTestUtils {
"spark://worker",
new SparkConf,
Seq("localDir"),
- ExecutorState.RUNNING)
+ ExecutorState.RUNNING,
+ resources)
}
def createDriverRunner(driverId: String): DriverRunner = {
@@ -88,4 +104,14 @@ private[deploy] object DeployTestUtils {
"spark://worker",
new SecurityManager(conf))
}
+
+ private def createResourceInformation: Map[String, ResourceInformation] = {
+ val gpuResource = new ResourceInformation(GPU, Array("0", "1", "2"))
+ val fpgaResource = new ResourceInformation(FPGA, Array("3", "4", "5"))
+ Map(GPU -> gpuResource, FPGA -> fpgaResource)
+ }
+
+ private def createResourceRequirement: Seq[ResourceRequirement] = {
+ Seq(ResourceRequirement("gpu", 3), ResourceRequirement("fpga", 3))
+ }
}
diff --git
a/core/src/test/scala/org/apache/spark/deploy/JsonProtocolSuite.scala
b/core/src/test/scala/org/apache/spark/deploy/JsonProtocolSuite.scala
index 1903130..ad402c0 100644
--- a/core/src/test/scala/org/apache/spark/deploy/JsonProtocolSuite.scala
+++ b/core/src/test/scala/org/apache/spark/deploy/JsonProtocolSuite.scala
@@ -51,7 +51,7 @@ class JsonProtocolSuite extends SparkFunSuite with
JsonTestUtils {
}
test("writeExecutorRunner") {
- val output = JsonProtocol.writeExecutorRunner(createExecutorRunner(123))
+ val output = JsonProtocol.writeExecutorRunner(createExecutorRunner(123,
true))
assertValidJson(output)
assertValidDataInJson(output,
JsonMethods.parse(JsonConstants.executorRunnerJsonStr))
}
@@ -78,8 +78,8 @@ class JsonProtocolSuite extends SparkFunSuite with
JsonTestUtils {
test("writeWorkerState") {
val executors = List[ExecutorRunner]()
- val finishedExecutors = List[ExecutorRunner](createExecutorRunner(123),
- createExecutorRunner(123))
+ val finishedExecutors = List[ExecutorRunner](createExecutorRunner(123,
true),
+ createExecutorRunner(123, true))
val drivers = List(createDriverRunner("driverId"))
val finishedDrivers = List(createDriverRunner("driverId"),
createDriverRunner("driverId"))
val stateResponse = new WorkerStateResponse("host", 8080, "workerId",
executors,
@@ -106,7 +106,10 @@ object JsonConstants {
"""
|{"id":"id","starttime":3,"name":"name",
|"cores":0,"user":"%s",
- |"memoryperslave":1234,"submitdate":"%s",
+ |"memoryperslave":1234,
+ |"resourcesperslave":[{"name":"gpu",
+ |"amount":3},{"name":"fpga","amount":3}],
+ |"submitdate":"%s",
|"state":"WAITING","duration":%d}
""".format(System.getProperty("user.name", "<unknown>"),
submitDate.toString, currTimeInMillis - appInfoStartTime).stripMargin
@@ -117,18 +120,28 @@ object JsonConstants {
|"webuiaddress":"http://publicAddress:80";,
|"cores":4,"coresused":0,"coresfree":4,
|"memory":1234,"memoryused":0,"memoryfree":1234,
+ |"resources":{"gpu":{"name":"gpu","addresses":
+ |["0","1","2"]},"fpga":{"name":"fpga","addresses"
+ |:["3","4","5"]}},"resourcesused":{"gpu":
+ |{"name":"gpu","addresses":[]},"fpga":
+ |{"name":"fpga","addresses":[]}},"resourcesfree":
+ |{"gpu":{"name":"gpu","addresses":["2","1","0"]},
+ |"fpga":{"name":"fpga","addresses":["5","4","3"]}},
|"state":"ALIVE","lastheartbeat":%d}
""".format(currTimeInMillis).stripMargin
val appDescJsonStr =
"""
- |{"name":"name","cores":4,"memoryperslave":1234,
+ |{"name":"name","cores":4,"memoryperslave":1234,"resourcesperslave":[],
|"user":"%s","command":"Command(mainClass,List(arg1,
arg2),Map(),List(),List(),List())"}
""".format(System.getProperty("user.name", "<unknown>")).stripMargin
val executorRunnerJsonStr =
"""
- |{"id":123,"memory":1234,"appid":"appId",
+ |{"id":123,"memory":1234,"resources":
+ |{"gpu":{"name":"gpu","addresses":["0","1","2"]},
+ |"fpga":{"name":"fpga","addresses":["3","4","5"]}},
+ |"appid":"appId",
|"appdesc":%s}
""".format(appDescJsonStr).stripMargin
@@ -136,6 +149,9 @@ object JsonConstants {
"""
|{"id":"driver-3","starttime":"3",
|"state":"SUBMITTED","cores":3,"memory":100,
+ |"resources":{"gpu":{"name":"gpu","addresses":
+ |["0","1","2"]},"fpga":{"name":"fpga",
+ |"addresses":["3","4","5"]}},
|"submitdate":"%s","worker":"None",
|"mainclass":"mainClass"}
""".format(submitDate.toString).stripMargin
@@ -146,6 +162,14 @@ object JsonConstants {
|"workers":[%s,%s],
|"aliveworkers":2,
|"cores":8,"coresused":0,"memory":2468,"memoryused":0,
+ |"resources":[{"gpu":{"name":"gpu","addresses":
+ |["0","1","2"]},"fpga":{"name":"fpga","addresses":
+ |["3","4","5"]}},{"gpu":{"name":"gpu","addresses":
+ |["0","1","2"]},"fpga":{"name":"fpga","addresses":
+ |["3","4","5"]}}],"resourcesused":[{"gpu":{"name":
+ |"gpu","addresses":[]},"fpga":{"name":"fpga","addresses":[]}}
+ |,{"gpu":{"name":"gpu","addresses":[]},"fpga":
+ |{"name":"fpga","addresses":[]}}],
|"activeapps":[%s],"completedapps":[],
|"activedrivers":[%s],
|"completeddrivers":[%s],
@@ -158,6 +182,7 @@ object JsonConstants {
|{"id":"workerId","masterurl":"masterUrl",
|"masterwebuiurl":"masterWebUiUrl",
|"cores":4,"coresused":4,"memory":1234,"memoryused":1234,
+ |"resources":{},"resourcesused":{},
|"executors":[],
|"finishedexecutors":[%s,%s]}
""".format(executorRunnerJsonStr, executorRunnerJsonStr).stripMargin
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
