[DCOS-44703] Implement TaskStatus Endpoint (#2796)

* fix location of debug package

* fix checkstyle

* WIP: adding working task info endpoint

* working taskStatus endpoint

* fixing checkstyle

* refactor structure

* use pojos for construction

* fix formatting

* use POJOs and add unit test

* fix checkstyle

* use string types

* use jarray not jobject

* remove jobject import

sdk/scheduler/build.gradle

@@ -121,6 +121,7 @@ ext {
 
 dependencies {
     compile "com.fasterxml.jackson.datatype:jackson-datatype-jdk8:${jacksonVer}"
+    compile "com.fasterxml.jackson.datatype:jackson-datatype-json-org:${jacksonVer}"
     compile "com.fasterxml.jackson.dataformat:jackson-dataformat-yaml:${jacksonVer}"
     compile "com.fasterxml.jackson.core:jackson-databind:${jacksonVer}"
     compile 'com.hubspot.jackson:jackson-datatype-protobuf:0.9.9-preJackson2.7-proto3'

sdk/scheduler/src/main/java/com/mesosphere/sdk/debug/TaskStatusesTracker.java

@@ -0,0 +1,232 @@
+package com.mesosphere.sdk.debug;
+
+import com.mesosphere.sdk.http.ResponseUtils;
+import com.mesosphere.sdk.scheduler.plan.Phase;
+import com.mesosphere.sdk.scheduler.plan.Plan;
+import com.mesosphere.sdk.scheduler.plan.PlanCoordinator;
+import com.mesosphere.sdk.scheduler.plan.PlanManager;
+import com.mesosphere.sdk.scheduler.plan.Step;
+import com.mesosphere.sdk.specification.TaskSpec;
+import com.mesosphere.sdk.state.StateStore;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.datatype.jdk8.Jdk8Module;
+import com.fasterxml.jackson.datatype.jsonorg.JsonOrgModule;
+import org.apache.mesos.Protos;
+import org.json.JSONArray;
+
+import javax.ws.rs.QueryParam;
+import javax.ws.rs.core.Response;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Optional;
+
+
+/**
+ * TaskStatusesTracker is the backend of TaskStatusesResource.
+ * It aggregates taskStatuses (if present) for plans, phases and steps.
+ */
+public class TaskStatusesTracker implements DebugEndpoint {
+
+  private final StateStore stateStore;
+
+  private final PlanCoordinator planCoordinator;
+
+
+  public TaskStatusesTracker(PlanCoordinator planCoordinator, StateStore stateStore) {
+    this.planCoordinator = planCoordinator;
+    this.stateStore = stateStore;
+  }
+
+  //CHECKSTYLE:OFF NestedForDepth
+  public List<PlanResponse> getTaskStatuses(String filterPlan,
+                                            String filterPhase,
+                                            String filterStep)
+  {
+    List<PlanResponse> planArray = new ArrayList<>();
+    for (PlanManager planManager : planCoordinator.getPlanManagers()) {
+      if (filterPlan != null && !planManager.getPlan().getName().equalsIgnoreCase(filterPlan)) {
+        continue;
+      }
+      Plan plan = planManager.getPlan();
+      PlanResponse planObject = new PlanResponse();
+      planObject.setName(plan.getName());
+      for (Phase phase : plan.getChildren()) {
+        if (filterPhase != null && !phase.getName().equalsIgnoreCase(filterPhase)) {
+          continue;
+        }
+        PhaseResponse phaseObject = new PhaseResponse();
+        phaseObject.setName(phase.getName());
+        for (Step step : phase.getChildren()) {
+          if (filterStep != null && !step.getName().equalsIgnoreCase(filterStep)) {
+            continue;
+          }
+          Collection<TaskSpec> tasksInStep = step.getPodInstanceRequirement()
+              .get().getPodInstance().getPod().getTasks();
+          TaskSpec taskSpec = tasksInStep.stream()
+              .filter(t -> step.getName().contains(t.getName()))
+              .findFirst()
+              .get();
+          TaskStatusResponse taskStatusObject = new TaskStatusResponse();
+          String taskInstanceName = TaskSpec.getInstanceName(
+              step.getPodInstanceRequirement().get().getPodInstance(),
+              taskSpec.getName()
+          );
+          Optional<Protos.TaskStatus> status = stateStore.fetchStatus(taskInstanceName);
+          taskStatusObject.setName(taskInstanceName);
+          status.ifPresent(s -> {
+            taskStatusObject.setTaskId(s.getTaskId().getValue());
+            taskStatusObject.setTaskStatus(s.getState().toString());
+          });
+          StepResponse stepObject = new StepResponse();
+          stepObject.setName(step.getName());
+          stepObject.addTaskStatus(taskStatusObject);
+          phaseObject.addStep(stepObject);
+        }
+        planObject.addPhase(phaseObject);
+      }
+      planArray.add(planObject);
+    }
+    return planArray;
+  }
+
+  //CHECKSTYLE:ON NestedForDepth
+
+  public Response getJson(@QueryParam("plan") String filterPlan,
+                          @QueryParam("phase") String filterPhase,
+                          @QueryParam("step") String filterStep,
+                          @QueryParam("sync") boolean requireSync)
+  {
+    ObjectMapper jsonMapper = new ObjectMapper();
+    jsonMapper.registerModule(new Jdk8Module());
+    jsonMapper.registerModule(new JsonOrgModule());
+
+    List<PlanResponse> serviceResponse = getTaskStatuses(filterPlan, filterPhase, filterStep);
+
+    JSONArray response = jsonMapper.convertValue(serviceResponse, JSONArray.class);
+
+    return ResponseUtils.jsonOkResponse(response);
+  }
+
+  /**
+   * Captures metadata for an individual taskStatus.
+   */
+  public static class TaskStatusResponse {
+
+    private String name;
+
+    private String taskState;
+
+    private String taskId;
+
+    public TaskStatusResponse() {
+      this.name = "";
+      this.taskState = "TASK_UNKNOWN";
+      this.taskId = "";
+    }
+
+    public void setName(String name) {
+      this.name = name;
+    }
+
+    public void setTaskStatus(String newTaskState) {
+      this.taskState = newTaskState;
+    }
+
+    public void setTaskId(String taskId) {
+      this.taskId = taskId;
+    }
+
+    public String getName() {
+      return this.name;
+    }
+
+    public String getTaskStatus() {
+      return this.taskState;
+    }
+
+    public String getTaskId() {
+      return this.taskId;
+    }
+  }
+
+  /**
+   * Captures metadata for an individual Step containing a set of TaskStatuses.
+   */
+  public static class StepResponse {
+
+    private String name = "";
+
+    private List<TaskStatusResponse> tasks = new ArrayList<>();
+
+
+    public void setName(String name) {
+      this.name = name;
+    }
+
+    public void addTaskStatus(TaskStatusResponse task) {
+      this.tasks.add(task);
+    }
+
+    public String getName() {
+      return this.name;
+    }
+
+    public List<TaskStatusResponse> getTaskStatus() {
+      return this.tasks;
+    }
+  }
+
+  /**
+   * Captures metadata for an individual Phase containing a set of Steps.
+   */
+  public static class PhaseResponse {
+
+    private String name = "";
+
+    private List<StepResponse> steps = new ArrayList<>();
+
+    public void addStep(StepResponse stepResponse) {
+      this.steps.add(stepResponse);
+    }
+
+    public void setName(String name) {
+      this.name = name;
+    }
+
+    public String getName() {
+      return this.name;
+    }
+
+    public List<StepResponse> getSteps() {
+      return this.steps;
+    }
+  }
+
+  /**
+   * Captures metadata for an individual Plan containing a set of Phases.
+   */
+  public static class PlanResponse {
+    private String name = "";
+
+    private List<PhaseResponse> phases = new ArrayList<>();
+
+    public void setName(String name) {
+      this.name = name;
+    }
+
+    public void addPhase(PhaseResponse phase) {
+      this.phases.add(phase);
+    }
+
+    public String getName() {
+      return this.name;
+    }
+
+    public List<PhaseResponse> getPhases() {
+      return this.phases;
+    }
+  }
+}

sdk/scheduler/src/main/java/com/mesosphere/sdk/http/endpoints/TaskStatusesResource.java

@@ -0,0 +1,38 @@
+package com.mesosphere.sdk.http.endpoints;
+
+import com.mesosphere.sdk.debug.TaskStatusesTracker;
+
+import javax.ws.rs.GET;
+import javax.ws.rs.Path;
+import javax.ws.rs.QueryParam;
+import javax.ws.rs.core.Response;
+
+/**
+ * A read-only API taskStatuses.
+ */
+
+@Path("/v1/debug")
+public class TaskStatusesResource {
+
+  private final TaskStatusesTracker statusesTracker;
+
+
+  public TaskStatusesResource(TaskStatusesTracker statusesTracker) {
+    this.statusesTracker = statusesTracker;
+  }
+
+  /**
+   * Renders the current set of TaskStatuses.
+   *
+   * @return JSON response of the TaskStatus Endpoint.
+   */
+  @GET
+  @Path("taskStatuses")
+  public Response getOfferOutcomes(@QueryParam("plan") String plan,
+                                   @QueryParam("phase") String phase,
+                                   @QueryParam("step") String step,
+                                   @QueryParam("sync") boolean sync)
+  {
+    return statusesTracker.getJson(plan, phase, step, sync);
+  }
+}

sdk/scheduler/src/main/java/com/mesosphere/sdk/scheduler/DefaultScheduler.java

@@ -1,5 +1,6 @@
 package com.mesosphere.sdk.scheduler;
 
+import com.mesosphere.sdk.debug.TaskStatusesTracker;
 import com.mesosphere.sdk.framework.TaskKiller;
 import com.mesosphere.sdk.http.endpoints.ArtifactResource;
 import com.mesosphere.sdk.http.endpoints.ConfigResource;
@@ -10,6 +11,7 @@
 import com.mesosphere.sdk.http.endpoints.PlansResource;
 import com.mesosphere.sdk.http.endpoints.PodResource;
 import com.mesosphere.sdk.http.endpoints.StateResource;
+import com.mesosphere.sdk.http.endpoints.TaskStatusesResource;
 import com.mesosphere.sdk.http.queries.ArtifactQueries;
 import com.mesosphere.sdk.http.types.EndpointProducer;
 import com.mesosphere.sdk.http.types.StringPropertyDeserializer;
@@ -99,6 +101,8 @@ public class DefaultScheduler extends AbstractScheduler {
 
   private final Optional<OfferOutcomeTracker> offerOutcomeTracker;
 
+  private final Optional<TaskStatusesTracker> statusesTracker;
+
   private final PlanScheduler planScheduler;
 
   /**
@@ -176,6 +180,7 @@ protected DefaultScheduler(
     // If the service is namespaced (i.e. part of a multi-service scheduler), disable the OfferOutcomeTracker to
     // reduce memory consumption.
     this.offerOutcomeTracker = namespace.isPresent() ? Optional.empty() : Optional.of(new OfferOutcomeTracker());
+    this.statusesTracker = Optional.of(new TaskStatusesTracker(getPlanCoordinator(), stateStore));
 
     this.planScheduler = new PlanScheduler(
         new OfferEvaluator(
@@ -211,6 +216,8 @@ public Collection<Object> getHTTPEndpoints() {
     resources.add(new PodResource(stateStore, configStore, serviceSpec.getName()));
     resources.add(new StateResource(frameworkStore, stateStore, new StringPropertyDeserializer()));
     offerOutcomeTracker.ifPresent(x -> resources.add(new DebugResource(x)));
+    statusesTracker.ifPresent(x -> resources.add(new TaskStatusesResource(x)));
+
     return resources;
   }