json endpoint for one job

nemccarthy · Mar 19, 2015 · 97d411f · 97d411f
1 parent 0c96147
commit 97d411f
Show file tree

Hide file tree

Showing 5 changed files with 85 additions and 8 deletions.
diff --git a/core/src/main/scala/org/apache/spark/status/api/v1/AllJobsResource.scala b/core/src/main/scala/org/apache/spark/status/api/v1/AllJobsResource.scala
@@ -16,11 +16,13 @@
  */
 package org.apache.spark.status.api.v1
 
+import java.util
 import java.util.Date
 import javax.ws.rs._
 import javax.ws.rs.core.MediaType
 
 import org.apache.spark.JobExecutionStatus
+import org.apache.spark.ui.SparkUI
 import org.apache.spark.ui.jobs.JobProgressListener
 import org.apache.spark.ui.jobs.UIData.JobUIData
 
@@ -33,14 +35,9 @@ class AllJobsResource(uiRoot: UIRoot) {
     @QueryParam("status") statuses: java.util.List[JobExecutionStatus]
   ): Seq[JobData] = {
     uiRoot.withSparkUI(appId) { ui =>
-      val statusToJobs = ui.jobProgressListener.synchronized {
-        Seq(
-          JobExecutionStatus.RUNNING -> ui.jobProgressListener.activeJobs.values.toSeq,
-          JobExecutionStatus.SUCCEEDED -> ui.jobProgressListener.completedJobs.toSeq,
-          JobExecutionStatus.FAILED -> ui.jobProgressListener.failedJobs.reverse.toSeq
-        )
-      }
-      val adjStatuses: java.util.List[JobExecutionStatus] = {
+      val statusToJobs: Seq[(JobExecutionStatus, Seq[JobUIData])] =
+        AllJobsResource.getStatusToJobs(ui)
+      val adjStatuses: util.List[JobExecutionStatus] = {
         if (statuses.isEmpty) {
           java.util.Arrays.asList(JobExecutionStatus.values(): _*)
         }
@@ -61,6 +58,19 @@ class AllJobsResource(uiRoot: UIRoot) {
 }
 
 object AllJobsResource {
+
+  def getStatusToJobs(ui: SparkUI): Seq[(JobExecutionStatus, Seq[JobUIData])] = {
+    val statusToJobs = ui.jobProgressListener.synchronized {
+      Seq(
+        JobExecutionStatus.RUNNING -> ui.jobProgressListener.activeJobs.values.toSeq,
+        JobExecutionStatus.SUCCEEDED -> ui.jobProgressListener.completedJobs.toSeq,
+        JobExecutionStatus.FAILED -> ui.jobProgressListener.failedJobs.reverse.toSeq
+      )
+    }
+    statusToJobs
+  }
+
+
   def convertJobData(
     job: JobUIData,
     listener: JobProgressListener,

diff --git a/core/src/main/scala/org/apache/spark/status/api/v1/JsonRootResource.scala b/core/src/main/scala/org/apache/spark/status/api/v1/JsonRootResource.scala
@@ -46,6 +46,12 @@ class JsonRootResource extends UIRootFromServletContext {
     new AllJobsResource(uiRoot)
   }
 
+  @Path("applications/{appId}/jobs/{jobId: \\d+}")
+  def getJob(): OneJobResource = {
+    new OneJobResource(uiRoot)
+  }
+
+
   @Path("applications/{appId}/executors")
   def getExecutors(): ExecutorListResource = {
     new ExecutorListResource(uiRoot)

diff --git a/core/src/main/scala/org/apache/spark/status/api/v1/OneJobResource.scala b/core/src/main/scala/org/apache/spark/status/api/v1/OneJobResource.scala
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.status.api.v1
+
+import javax.ws.rs.{PathParam, GET, Produces}
+import javax.ws.rs.core.MediaType
+
+import org.apache.spark.JobExecutionStatus
+import org.apache.spark.ui.jobs.UIData.JobUIData
+
+@Produces(Array(MediaType.APPLICATION_JSON))
+class OneJobResource(uiRoot: UIRoot) {
+
+  @GET
+  def jobsList(
+    @PathParam("appId") appId: String,
+    @PathParam("jobId") jobId: Int
+  ): JobData = {
+    uiRoot.withSparkUI(appId) { ui =>
+      val statusToJobs: Seq[(JobExecutionStatus, Seq[JobUIData])] =
+        AllJobsResource.getStatusToJobs(ui)
+      val jobOpt = statusToJobs.map {_._2} .flatten.find { jobInfo => jobInfo.jobId == jobId}
+      jobOpt.map { job =>
+        AllJobsResource.convertJobData(job, ui.jobProgressListener, false)
+      }.getOrElse {
+        throw new NotFoundException("unknown job: " + jobId)
+      }
+    }
+  }
+
+}
diff --git a/...ources/HistoryServerExpectations/applications/local-1422981780767/jobs/0/json_expectation b/...ources/HistoryServerExpectations/applications/local-1422981780767/jobs/0/json_expectation
@@ -0,0 +1,15 @@
+{
+  "jobId" : 0,
+  "name" : "count at <console>:15",
+  "stageIds" : [ 0 ],
+  "status" : "SUCCEEDED",
+  "numTasks" : 8,
+  "numActiveTasks" : 0,
+  "numCompletedTasks" : 8,
+  "numSkippedTasks" : 8,
+  "numFailedTasks" : 0,
+  "numActiveStages" : 0,
+  "numCompletedStages" : 1,
+  "numSkippedStages" : 0,
+  "numFailedStages" : 0
+}
diff --git a/core/src/test/scala/org/apache/spark/deploy/history/HistoryServerSuite.scala b/core/src/test/scala/org/apache/spark/deploy/history/HistoryServerSuite.scala
@@ -71,6 +71,7 @@ class HistoryServerSuite extends FunSuite with BeforeAndAfter with Matchers with
     "maxDate2 app list json" -> "applications?maxDate=2015-02-03T10:42:40.000CST",
     "one app json" -> "applications/local-1422981780767",
     "job list json" -> "applications/local-1422981780767/jobs",
+    "one job json" -> "applications/local-1422981780767/jobs/0",
     "succeeded job list json" -> "applications/local-1422981780767/jobs?status=succeeded",
     "succeeded&failed job list json" -> "applications/local-1422981780767/jobs?status=succeeded&status=failed",
     "executor list json" -> "applications/local-1422981780767/executors",