Merge pull request #1455 from hashicorp/f-job-summary

Job Summary - Part 2

api/jobs.go

@@ -159,6 +159,15 @@ func (j *Jobs) Plan(job *Job, diff bool, q *WriteOptions) (*JobPlanResponse, *Wr
 	return &resp, wm, nil
 }
 
+func (j *Jobs) Summary(jobID string, q *QueryOptions) (*JobSummary, *QueryMeta, error) {
+	var resp JobSummary
+	qm, err := j.client.query("/v1/job/"+jobID+"/summary", &resp, q)
+	if err != nil {
+		return nil, nil, err
+	}
+	return &resp, qm, nil
+}
+
 // periodicForceResponse is used to deserialize a force response
 type periodicForceResponse struct {
 	EvalID string
@@ -199,6 +208,27 @@ type Job struct {
 	JobModifyIndex    uint64
 }
 
+// JobSummary summarizes the state of the allocations of a job
+type JobSummary struct {
+	JobID   string
+	Summary map[string]TaskGroupSummary
+
+	// Raft Indexes
+	CreateIndex uint64
+	ModifyIndex uint64
+}
+
+// TaskGroup summarizes the state of all the allocations of a particular
+// TaskGroup
+type TaskGroupSummary struct {
+	Queued   int
+	Complete int
+	Failed   int
+	Running  int
+	Starting int
+	Lost     int
+}
+
 // JobListStub is used to return a subset of information about
 // jobs during list operations.
 type JobListStub struct {
@@ -209,6 +239,7 @@ type JobListStub struct {
 	Priority          int
 	Status            string
 	StatusDescription string
+	JobSummary        *JobSummary
 	CreateIndex       uint64
 	ModifyIndex       uint64
 	JobModifyIndex    uint64

api/jobs_test.go

@@ -488,6 +488,48 @@ func TestJobs_Plan(t *testing.T) {
 	}
 }
 
+func TestJobs_JobSummary(t *testing.T) {
+	c, s := makeClient(t, nil, nil)
+	defer s.Stop()
+	jobs := c.Jobs()
+
+	// Trying to retrieve a job summary before the job exists
+	// returns an error
+	_, _, err := jobs.Summary("job1", nil)
+	if err == nil || !strings.Contains(err.Error(), "not found") {
+		t.Fatalf("expected not found error, got: %#v", err)
+	}
+
+	// Register the job
+	job := testJob()
+	_, wm, err := jobs.Register(job, nil)
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+	assertWriteMeta(t, wm)
+
+	// Query the job summary again and ensure it exists
+	result, qm, err := jobs.Summary("job1", nil)
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+	assertQueryMeta(t, qm)
+
+	expectedJobSummary := JobSummary{
+		JobID: job.ID,
+		Summary: map[string]TaskGroupSummary{
+			job.TaskGroups[0].Name: {},
+		},
+		CreateIndex: result.CreateIndex,
+		ModifyIndex: result.ModifyIndex,
+	}
+
+	// Check that the result is what we expect
+	if !reflect.DeepEqual(&expectedJobSummary, result) {
+		t.Fatalf("expect: %#v, got: %#v", expectedJobSummary, result)
+	}
+}
+
 func TestJobs_NewBatchJob(t *testing.T) {
 	job := NewBatchJob("job1", "myjob", "region1", 5)
 	expect := &Job{

client/client_test.go

@@ -357,6 +357,9 @@ func TestClient_UpdateAllocStatus(t *testing.T) {
 	alloc.ClientStatus = originalStatus
 
 	state := s1.State()
+	if err := state.UpsertJobSummary(99, mock.JobSummary(alloc.JobID)); err != nil {
+		t.Fatal(err)
+	}
 	state.UpsertAllocs(100, []*structs.Allocation{alloc})
 
 	testutil.WaitForResult(func() (bool, error) {
@@ -394,6 +397,12 @@ func TestClient_WatchAllocs(t *testing.T) {
 	alloc2.NodeID = c1.Node().ID
 
 	state := s1.State()
+	if err := state.UpsertJobSummary(998, mock.JobSummary(alloc1.JobID)); err != nil {
+		t.Fatal(err)
+	}
+	if err := state.UpsertJobSummary(999, mock.JobSummary(alloc2.JobID)); err != nil {
+		t.Fatal(err)
+	}
 	err := state.UpsertAllocs(100,
 		[]*structs.Allocation{alloc1, alloc2})
 	if err != nil {
@@ -469,8 +478,10 @@ func TestClient_SaveRestoreState(t *testing.T) {
 	task.Config["args"] = []string{"10"}
 
 	state := s1.State()
-	err := state.UpsertAllocs(100, []*structs.Allocation{alloc1})
-	if err != nil {
+	if err := state.UpsertJobSummary(99, mock.JobSummary(alloc1.JobID)); err != nil {
+		t.Fatal(err)
+	}
+	if err := state.UpsertAllocs(100, []*structs.Allocation{alloc1}); err != nil {
 		t.Fatalf("err: %v", err)
 	}
 
@@ -485,8 +496,7 @@ func TestClient_SaveRestoreState(t *testing.T) {
 	})
 
 	// Shutdown the client, saves state
-	err = c1.Shutdown()
-	if err != nil {
+	if err := c1.Shutdown(); err != nil {
 		t.Fatalf("err: %v", err)
 	}
 

command/agent/alloc_endpoint_test.go

@@ -16,6 +16,8 @@ func TestHTTP_AllocsList(t *testing.T) {
 		state := s.Agent.server.State()
 		alloc1 := mock.Alloc()
 		alloc2 := mock.Alloc()
+		state.UpsertJobSummary(998, mock.JobSummary(alloc1.JobID))
+		state.UpsertJobSummary(999, mock.JobSummary(alloc2.JobID))
 		err := state.UpsertAllocs(1000,
 			[]*structs.Allocation{alloc1, alloc2})
 		if err != nil {
@@ -58,13 +60,21 @@ func TestHTTP_AllocsPrefixList(t *testing.T) {
 	httpTest(t, nil, func(s *TestServer) {
 		// Directly manipulate the state
 		state := s.Agent.server.State()
+
 		alloc1 := mock.Alloc()
 		alloc1.ID = "aaaaaaaa-e8f7-fd38-c855-ab94ceb89706"
 		alloc2 := mock.Alloc()
 		alloc2.ID = "aaabbbbb-e8f7-fd38-c855-ab94ceb89706"
-		err := state.UpsertAllocs(1000,
-			[]*structs.Allocation{alloc1, alloc2})
-		if err != nil {
+		summary1 := mock.JobSummary(alloc1.JobID)
+		summary2 := mock.JobSummary(alloc2.JobID)
+		if err := state.UpsertJobSummary(998, summary1); err != nil {
+			t.Fatal(err)
+		}
+		if err := state.UpsertJobSummary(999, summary2); err != nil {
+			t.Fatal(err)
+		}
+		if err := state.UpsertAllocs(1000,
+			[]*structs.Allocation{alloc1, alloc2}); err != nil {
 			t.Fatalf("err: %v", err)
 		}
 
@@ -110,6 +120,9 @@ func TestHTTP_AllocQuery(t *testing.T) {
 		// Directly manipulate the state
 		state := s.Agent.server.State()
 		alloc := mock.Alloc()
+		if err := state.UpsertJobSummary(999, mock.JobSummary(alloc.JobID)); err != nil {
+			t.Fatal(err)
+		}
 		err := state.UpsertAllocs(1000,
 			[]*structs.Allocation{alloc})
 		if err != nil {

command/agent/eval_endpoint_test.go

@@ -111,6 +111,8 @@ func TestHTTP_EvalAllocations(t *testing.T) {
 		alloc1 := mock.Alloc()
 		alloc2 := mock.Alloc()
 		alloc2.EvalID = alloc1.EvalID
+		state.UpsertJobSummary(998, mock.JobSummary(alloc1.JobID))
+		state.UpsertJobSummary(999, mock.JobSummary(alloc2.JobID))
 		err := state.UpsertAllocs(1000,
 			[]*structs.Allocation{alloc1, alloc2})
 		if err != nil {

command/agent/job_endpoint.go

@@ -54,6 +54,9 @@ func (s *HTTPServer) JobSpecificRequest(resp http.ResponseWriter, req *http.Requ
 	case strings.HasSuffix(path, "/plan"):
 		jobName := strings.TrimSuffix(path, "/plan")
 		return s.jobPlan(resp, req, jobName)
+	case strings.HasSuffix(path, "/summary"):
+		jobName := strings.TrimSuffix(path, "/summary")
+		return s.jobSummaryRequest(resp, req, jobName)
 	default:
 		return s.jobCRUD(resp, req, path)
 	}
@@ -241,3 +244,24 @@ func (s *HTTPServer) jobDelete(resp http.ResponseWriter, req *http.Request,
 	setIndex(resp, out.Index)
 	return out, nil
 }
+
+func (s *HTTPServer) jobSummaryRequest(resp http.ResponseWriter, req *http.Request, name string) (interface{}, error) {
+	args := structs.JobSummaryRequest{
+		JobID: name,
+	}
+	if s.parse(resp, req, &args.Region, &args.QueryOptions) {
+		return nil, nil
+	}
+
+	var out structs.JobSummaryResponse
+	if err := s.agent.RPC("Job.Summary", &args, &out); err != nil {
+		return nil, err
+	}
+
+	setMeta(resp, &out.QueryMeta)
+	if out.JobSummary == nil {
+		return nil, CodedError(404, "job not found")
+	}
+	setIndex(resp, out.Index)
+	return out.JobSummary, nil
+}

command/agent/node_endpoint_test.go

@@ -129,6 +129,9 @@ func TestHTTP_NodeForceEval(t *testing.T) {
 		state := s.Agent.server.State()
 		alloc1 := mock.Alloc()
 		alloc1.NodeID = node.ID
+		if err := state.UpsertJobSummary(999, mock.JobSummary(alloc1.JobID)); err != nil {
+			t.Fatal(err)
+		}
 		err := state.UpsertAllocs(1000, []*structs.Allocation{alloc1})
 		if err != nil {
 			t.Fatalf("err: %v", err)
@@ -177,6 +180,9 @@ func TestHTTP_NodeAllocations(t *testing.T) {
 		state := s.Agent.server.State()
 		alloc1 := mock.Alloc()
 		alloc1.NodeID = node.ID
+		if err := state.UpsertJobSummary(999, mock.JobSummary(alloc1.JobID)); err != nil {
+			t.Fatal(err)
+		}
 		err := state.UpsertAllocs(1000, []*structs.Allocation{alloc1})
 		if err != nil {
 			t.Fatalf("err: %v", err)
@@ -231,6 +237,9 @@ func TestHTTP_NodeDrain(t *testing.T) {
 		state := s.Agent.server.State()
 		alloc1 := mock.Alloc()
 		alloc1.NodeID = node.ID
+		if err := state.UpsertJobSummary(999, mock.JobSummary(alloc1.JobID)); err != nil {
+			t.Fatal(err)
+		}
 		err := state.UpsertAllocs(1000, []*structs.Allocation{alloc1})
 		if err != nil {
 			t.Fatalf("err: %v", err)

command/status.go

@@ -246,6 +246,29 @@ func (c *StatusCommand) outputJobInfo(client *api.Client, job *api.Job) error {
 		return fmt.Errorf("Error querying job evaluations: %s", err)
 	}
 
+	// Query the summary
+	summary, _, err := client.Jobs().Summary(job.ID, nil)
+	if err != nil {
+		return fmt.Errorf("Error querying job summary: %s", err)
+	}
+
+	// Format the summary
+	c.Ui.Output(c.Colorize().Color("\n[bold]Summary[reset]"))
+	if summary != nil {
+		summaries := make([]string, len(summary.Summary)+1)
+		summaries[0] = "Task Group|Queued|Starting|Running|Failed|Complete|Lost"
+		idx := 1
+		for tg, tgs := range summary.Summary {
+			summaries[idx] = fmt.Sprintf("%s|%d|%d|%d|%d|%d|%d",
+				tg, tgs.Queued, tgs.Starting,
+				tgs.Running, tgs.Failed,
+				tgs.Complete, tgs.Lost,
+			)
+			idx += 1
+		}
+		c.Ui.Output(formatList(summaries))
+	}
+
 	// Determine latest evaluation with failures whose follow up hasn't
 	// completed, this is done while formatting
 	var latestFailedPlacement *api.Evaluation

command/status_test.go

@@ -63,6 +63,9 @@ func TestStatusCommand_Run(t *testing.T) {
 	if !strings.Contains(out, "Allocations") {
 		t.Fatalf("should dump allocations")
 	}
+	if !strings.Contains(out, "Summary") {
+		t.Fatalf("should dump summary")
+	}
 	ui.OutputWriter.Reset()
 
 	// Query a single job showing evals

nomad/alloc_endpoint_test.go

@@ -19,9 +19,13 @@ func TestAllocEndpoint_List(t *testing.T) {
 
 	// Create the register request
 	alloc := mock.Alloc()
+	summary := mock.JobSummary(alloc.JobID)
 	state := s1.fsm.State()
-	err := state.UpsertAllocs(1000, []*structs.Allocation{alloc})
-	if err != nil {
+
+	if err := state.UpsertJobSummary(999, summary); err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	if err := state.UpsertAllocs(1000, []*structs.Allocation{alloc}); err != nil {
 		t.Fatalf("err: %v", err)
 	}
 
@@ -75,6 +79,10 @@ func TestAllocEndpoint_List_Blocking(t *testing.T) {
 	// Create the alloc
 	alloc := mock.Alloc()
 
+	summary := mock.JobSummary(alloc.JobID)
+	if err := state.UpsertJobSummary(1, summary); err != nil {
+		t.Fatalf("err: %v", err)
+	}
 	// Upsert alloc triggers watches
 	time.AfterFunc(100*time.Millisecond, func() {
 		if err := state.UpsertAllocs(2, []*structs.Allocation{alloc}); err != nil {
@@ -109,12 +117,13 @@ func TestAllocEndpoint_List_Blocking(t *testing.T) {
 	alloc2.ID = alloc.ID
 	alloc2.ClientStatus = structs.AllocClientStatusRunning
 	time.AfterFunc(100*time.Millisecond, func() {
-		if err := state.UpdateAllocsFromClient(3, []*structs.Allocation{alloc2}); err != nil {
+		state.UpsertJobSummary(3, mock.JobSummary(alloc2.JobID))
+		if err := state.UpdateAllocsFromClient(4, []*structs.Allocation{alloc2}); err != nil {
 			t.Fatalf("err: %v", err)
 		}
 	})
 
-	req.MinQueryIndex = 2
+	req.MinQueryIndex = 3
 	start = time.Now()
 	var resp2 structs.AllocListResponse
 	if err := msgpackrpc.CallWithCodec(codec, "Alloc.List", req, &resp2); err != nil {
@@ -124,8 +133,8 @@ func TestAllocEndpoint_List_Blocking(t *testing.T) {
 	if elapsed := time.Since(start); elapsed < 100*time.Millisecond {
 		t.Fatalf("should block (returned in %s) %#v", elapsed, resp2)
 	}
-	if resp2.Index != 3 {
-		t.Fatalf("Bad index: %d %d", resp2.Index, 3)
+	if resp2.Index != 4 {
+		t.Fatalf("Bad index: %d %d", resp2.Index, 4)
 	}
 	if len(resp2.Allocations) != 1 || resp.Allocations[0].ID != alloc.ID ||
 		resp2.Allocations[0].ClientStatus != structs.AllocClientStatusRunning {
@@ -142,6 +151,7 @@ func TestAllocEndpoint_GetAlloc(t *testing.T) {
 	// Create the register request
 	alloc := mock.Alloc()
 	state := s1.fsm.State()
+	state.UpsertJobSummary(999, mock.JobSummary(alloc.JobID))
 	err := state.UpsertAllocs(1000, []*structs.Allocation{alloc})
 	if err != nil {
 		t.Fatalf("err: %v", err)
@@ -178,6 +188,7 @@ func TestAllocEndpoint_GetAlloc_Blocking(t *testing.T) {
 
 	// First create an unrelated alloc
 	time.AfterFunc(100*time.Millisecond, func() {
+		state.UpsertJobSummary(99, mock.JobSummary(alloc1.JobID))
 		err := state.UpsertAllocs(100, []*structs.Allocation{alloc1})
 		if err != nil {
 			t.Fatalf("err: %v", err)
@@ -186,6 +197,7 @@ func TestAllocEndpoint_GetAlloc_Blocking(t *testing.T) {
 
 	// Create the alloc we are watching later
 	time.AfterFunc(200*time.Millisecond, func() {
+		state.UpsertJobSummary(999, mock.JobSummary(alloc2.JobID))
 		err := state.UpsertAllocs(200, []*structs.Allocation{alloc2})
 		if err != nil {
 			t.Fatalf("err: %v", err)
@@ -227,6 +239,8 @@ func TestAllocEndpoint_GetAllocs(t *testing.T) {
 	alloc := mock.Alloc()
 	alloc2 := mock.Alloc()
 	state := s1.fsm.State()
+	state.UpsertJobSummary(998, mock.JobSummary(alloc.JobID))
+	state.UpsertJobSummary(999, mock.JobSummary(alloc2.JobID))
 	err := state.UpsertAllocs(1000, []*structs.Allocation{alloc, alloc2})
 	if err != nil {
 		t.Fatalf("err: %v", err)