Merge pull request #5702 from hashicorp/f-filter-by-create-index

Filter deployments by create index

api/jobs.go

@@ -179,9 +179,17 @@ func (j *Jobs) Allocations(jobID string, allAllocs bool, q *QueryOptions) ([]*Al
 
 // Deployments is used to query the deployments associated with the given job
 // ID.
-func (j *Jobs) Deployments(jobID string, q *QueryOptions) ([]*Deployment, *QueryMeta, error) {
+func (j *Jobs) Deployments(jobID string, all bool, q *QueryOptions) ([]*Deployment, *QueryMeta, error) {
 	var resp []*Deployment
-	qm, err := j.client.query("/v1/job/"+jobID+"/deployments", &resp, q)
+	u, err := url.Parse("/v1/job/" + jobID + "/deployments")
+	if err != nil {
+		return nil, nil, err
+	}
+
+	v := u.Query()
+	v.Add("all", strconv.FormatBool(all))
+	u.RawQuery = v.Encode()
+	qm, err := j.client.query(u.String(), &resp, q)
 	if err != nil {
 		return nil, nil, err
 	}

command/agent/job_endpoint.go

@@ -219,8 +219,8 @@ func (s *HTTPServer) jobAllocations(resp http.ResponseWriter, req *http.Request,
 	allAllocs, _ := strconv.ParseBool(req.URL.Query().Get("all"))
 
 	args := structs.JobSpecificRequest{
-		JobID:     jobName,
-		AllAllocs: allAllocs,
+		JobID: jobName,
+		All:   allAllocs,
 	}
 	if s.parse(resp, req, &args.Region, &args.QueryOptions) {
 		return nil, nil
@@ -270,8 +270,10 @@ func (s *HTTPServer) jobDeployments(resp http.ResponseWriter, req *http.Request,
 	if req.Method != "GET" {
 		return nil, CodedError(405, ErrInvalidMethod)
 	}
+	all, _ := strconv.ParseBool(req.URL.Query().Get("all"))
 	args := structs.JobSpecificRequest{
 		JobID: jobName,
+		All:   all,
 	}
 	if s.parse(resp, req, &args.Region, &args.QueryOptions) {
 		return nil, nil

command/agent/job_endpoint_test.go

@@ -797,6 +797,8 @@ func TestHTTP_JobDeployments(t *testing.T) {
 		state := s.Agent.server.State()
 		d := mock.Deployment()
 		d.JobID = j.ID
+		d.JobCreateIndex = resp.JobModifyIndex
+
 		assert.Nil(state.UpsertDeployment(1000, d), "UpsertDeployment")
 
 		// Make the HTTP request
@@ -839,6 +841,7 @@ func TestHTTP_JobDeployment(t *testing.T) {
 		state := s.Agent.server.State()
 		d := mock.Deployment()
 		d.JobID = j.ID
+		d.JobCreateIndex = resp.JobModifyIndex
 		assert.Nil(state.UpsertDeployment(1000, d), "UpsertDeployment")
 
 		// Make the HTTP request

command/job_deployments.go

@@ -35,6 +35,10 @@ Deployments Options:
 
   -verbose
     Display full information.
+
+  -all-allocs
+    Display all deployments matching the job ID, including those 
+    from an older instance of the job.
 `
 	return strings.TrimSpace(helpText)
 }
@@ -50,6 +54,7 @@ func (c *JobDeploymentsCommand) AutocompleteFlags() complete.Flags {
 			"-t":       complete.PredictAnything,
 			"-latest":  complete.PredictNothing,
 			"-verbose": complete.PredictNothing,
+			"-all":     complete.PredictNothing,
 		})
 }
 
@@ -71,13 +76,14 @@ func (c *JobDeploymentsCommand) AutocompleteArgs() complete.Predictor {
 func (c *JobDeploymentsCommand) Name() string { return "job deployments" }
 
 func (c *JobDeploymentsCommand) Run(args []string) int {
-	var json, latest, verbose bool
+	var json, latest, verbose, all bool
 	var tmpl string
 
 	flags := c.Meta.FlagSet(c.Name(), FlagSetClient)
 	flags.Usage = func() { c.Ui.Output(c.Help()) }
 	flags.BoolVar(&latest, "latest", false, "")
 	flags.BoolVar(&verbose, "verbose", false, "")
+	flags.BoolVar(&all, "all", false, "")
 	flags.BoolVar(&json, "json", false, "")
 	flags.StringVar(&tmpl, "t", "", "")
 
@@ -146,7 +152,7 @@ func (c *JobDeploymentsCommand) Run(args []string) int {
 		return 0
 	}
 
-	deploys, _, err := client.Jobs().Deployments(jobID, nil)
+	deploys, _, err := client.Jobs().Deployments(jobID, all, nil)
 	if err != nil {
 		c.Ui.Error(fmt.Sprintf("Error retrieving deployments: %s", err))
 		return 1

command/job_deployments_test.go

@@ -69,6 +69,7 @@ func TestJobDeploymentsCommand_Run(t *testing.T) {
 	// Inject a deployment
 	d := mock.Deployment()
 	d.JobID = job.ID
+	d.JobCreateIndex = job.CreateIndex
 	assert.Nil(state.UpsertDeployment(200, d))
 
 	// Should now display the deployment
@@ -112,6 +113,7 @@ func TestJobDeploymentsCommand_Run_Latest(t *testing.T) {
 	// Inject a deployment
 	d := mock.Deployment()
 	d.JobID = job.ID
+	d.JobCreateIndex = job.CreateIndex
 	assert.Nil(state.UpsertDeployment(200, d))
 
 	// Should now display the deployment

e2e/consul/consul.go

@@ -113,7 +113,7 @@ func (tc *ConsulE2ETest) TestCanaryInplaceUpgrades(f *framework.F) {
 	g := NewGomegaWithT(f.T())
 
 	g.Eventually(func() []string {
-		deploys, _, err := jobs.Deployments(jobId, nil)
+		deploys, _, err := jobs.Deployments(jobId, false, nil)
 		require.Nil(err)
 		healthyDeploys := make([]string, 0, len(deploys))
 		for _, d := range deploys {
@@ -135,7 +135,7 @@ func (tc *ConsulE2ETest) TestCanaryInplaceUpgrades(f *framework.F) {
 	// Eventually have a canary
 	var deploys []*api.Deployment
 	g.Eventually(func() []*api.Deployment {
-		deploys, _, err = jobs.Deployments(*job.ID, nil)
+		deploys, _, err = jobs.Deployments(*job.ID, false, nil)
 		require.Nil(err)
 		return deploys
 	}, 2*time.Second, 20*time.Millisecond).Should(HaveLen(2))

e2e/rescheduling/server_side_restarts_test.go

@@ -53,7 +53,7 @@ var _ = Describe("Server Side Restart Tests", func() {
 		// deploymentStatus is a helper function that returns deployment status of all deployments
 		// sorted by time
 		deploymentStatus = func() []string {
-			deploys, _, err := jobs.Deployments(*job.ID, nil)
+			deploys, _, err := jobs.Deployments(*job.ID, false, nil)
 			Expect(err).ShouldNot(HaveOccurred())
 			var ret []string
 			sort.Slice(deploys, func(i, j int) bool {

nomad/job_endpoint.go

@@ -951,7 +951,7 @@ func (j *Job) Allocations(args *structs.JobSpecificRequest,
 		queryMeta: &reply.QueryMeta,
 		run: func(ws memdb.WatchSet, state *state.StateStore) error {
 			// Capture the allocations
-			allocs, err := state.AllocsByJob(ws, args.RequestNamespace(), args.JobID, args.AllAllocs)
+			allocs, err := state.AllocsByJob(ws, args.RequestNamespace(), args.JobID, args.All)
 			if err != nil {
 				return err
 			}
@@ -1042,7 +1042,7 @@ func (j *Job) Deployments(args *structs.JobSpecificRequest,
 		queryMeta: &reply.QueryMeta,
 		run: func(ws memdb.WatchSet, state *state.StateStore) error {
 			// Capture the deployments
-			deploys, err := state.DeploymentsByJobID(ws, args.RequestNamespace(), args.JobID)
+			deploys, err := state.DeploymentsByJobID(ws, args.RequestNamespace(), args.JobID, args.All)
 			if err != nil {
 				return err
 			}
@@ -1084,7 +1084,7 @@ func (j *Job) LatestDeployment(args *structs.JobSpecificRequest,
 		queryMeta: &reply.QueryMeta,
 		run: func(ws memdb.WatchSet, state *state.StateStore) error {
 			// Capture the deployments
-			deploys, err := state.DeploymentsByJobID(ws, args.RequestNamespace(), args.JobID)
+			deploys, err := state.DeploymentsByJobID(ws, args.RequestNamespace(), args.JobID, args.All)
 			if err != nil {
 				return err
 			}

nomad/job_endpoint_test.go

@@ -3528,6 +3528,9 @@ func TestJobEndpoint_Deployments(t *testing.T) {
 	d1.JobID = j.ID
 	d2.JobID = j.ID
 	require.Nil(state.UpsertJob(1000, j), "UpsertJob")
+	d1.JobCreateIndex = j.CreateIndex
+	d2.JobCreateIndex = j.CreateIndex
+
 	require.Nil(state.UpsertDeployment(1001, d1), "UpsertDeployment")
 	require.Nil(state.UpsertDeployment(1002, d2), "UpsertDeployment")
 
@@ -3562,6 +3565,8 @@ func TestJobEndpoint_Deployments_ACL(t *testing.T) {
 	d1.JobID = j.ID
 	d2.JobID = j.ID
 	require.Nil(state.UpsertJob(1000, j), "UpsertJob")
+	d1.JobCreateIndex = j.CreateIndex
+	d2.JobCreateIndex = j.CreateIndex
 	require.Nil(state.UpsertDeployment(1001, d1), "UpsertDeployment")
 	require.Nil(state.UpsertDeployment(1002, d2), "UpsertDeployment")
 
@@ -3622,7 +3627,7 @@ func TestJobEndpoint_Deployments_Blocking(t *testing.T) {
 	d2 := mock.Deployment()
 	d2.JobID = j.ID
 	require.Nil(state.UpsertJob(50, j), "UpsertJob")
-
+	d2.JobCreateIndex = j.CreateIndex
 	// First upsert an unrelated eval
 	time.AfterFunc(100*time.Millisecond, func() {
 		require.Nil(state.UpsertDeployment(100, d1), "UpsertDeployment")
@@ -3671,6 +3676,8 @@ func TestJobEndpoint_LatestDeployment(t *testing.T) {
 	d2.CreateIndex = d1.CreateIndex + 100
 	d2.ModifyIndex = d2.CreateIndex + 100
 	require.Nil(state.UpsertJob(1000, j), "UpsertJob")
+	d1.JobCreateIndex = j.CreateIndex
+	d2.JobCreateIndex = j.CreateIndex
 	require.Nil(state.UpsertDeployment(1001, d1), "UpsertDeployment")
 	require.Nil(state.UpsertDeployment(1002, d2), "UpsertDeployment")
 
@@ -3708,6 +3715,8 @@ func TestJobEndpoint_LatestDeployment_ACL(t *testing.T) {
 	d2.CreateIndex = d1.CreateIndex + 100
 	d2.ModifyIndex = d2.CreateIndex + 100
 	require.Nil(state.UpsertJob(1000, j), "UpsertJob")
+	d1.JobCreateIndex = j.CreateIndex
+	d2.JobCreateIndex = j.CreateIndex
 	require.Nil(state.UpsertDeployment(1001, d1), "UpsertDeployment")
 	require.Nil(state.UpsertDeployment(1002, d2), "UpsertDeployment")
 
@@ -3771,6 +3780,7 @@ func TestJobEndpoint_LatestDeployment_Blocking(t *testing.T) {
 	d2 := mock.Deployment()
 	d2.JobID = j.ID
 	require.Nil(state.UpsertJob(50, j), "UpsertJob")
+	d2.JobCreateIndex = j.CreateIndex
 
 	// First upsert an unrelated eval
 	time.AfterFunc(100*time.Millisecond, func() {

nomad/state/state_store.go

@@ -481,14 +481,24 @@ func (s *StateStore) deploymentByIDImpl(ws memdb.WatchSet, deploymentID string,
 	return nil, nil
 }
 
-func (s *StateStore) DeploymentsByJobID(ws memdb.WatchSet, namespace, jobID string) ([]*structs.Deployment, error) {
+func (s *StateStore) DeploymentsByJobID(ws memdb.WatchSet, namespace, jobID string, all bool) ([]*structs.Deployment, error) {
 	txn := s.db.Txn(false)
 
 	// COMPAT 0.7: Upgrade old objects that do not have namespaces
 	if namespace == "" {
 		namespace = structs.DefaultNamespace
 	}
 
+	var job *structs.Job
+	// Read job from state store
+	_, existing, err := txn.FirstWatch("jobs", "id", namespace, jobID)
+	if err != nil {
+		return nil, fmt.Errorf("job lookup failed: %v", err)
+	}
+	if existing != nil {
+		job = existing.(*structs.Job)
+	}
+
 	// Get an iterator over the deployments
 	iter, err := txn.Get("deployment", "job", namespace, jobID)
 	if err != nil {
@@ -503,8 +513,14 @@ func (s *StateStore) DeploymentsByJobID(ws memdb.WatchSet, namespace, jobID stri
 		if raw == nil {
 			break
 		}
-
 		d := raw.(*structs.Deployment)
+
+		// If the allocation belongs to a job with the same ID but a different
+		// create index and we are not getting all the allocations whose Jobs
+		// matches the same Job ID then we skip it
+		if !all && job != nil && d.JobCreateIndex != job.CreateIndex {
+			continue
+		}
 		out = append(out, d)
 	}
 

nomad/state/state_store_test.go

@@ -494,7 +494,7 @@ func TestStateStore_UpsertDeployment(t *testing.T) {
 
 	// Create a watchset so we can test that upsert fires the watch
 	ws := memdb.NewWatchSet()
-	_, err := state.DeploymentsByJobID(ws, deployment.Namespace, deployment.ID)
+	_, err := state.DeploymentsByJobID(ws, deployment.Namespace, deployment.ID, true)
 	if err != nil {
 		t.Fatalf("bad: %v", err)
 	}
@@ -530,6 +530,43 @@ func TestStateStore_UpsertDeployment(t *testing.T) {
 	}
 }
 
+// Tests that deployments of older create index and same job id are not returned
+func TestStateStore_OldDeployment(t *testing.T) {
+	state := testStateStore(t)
+	job := mock.Job()
+	job.ID = "job1"
+	state.UpsertJob(1000, job)
+
+	deploy1 := mock.Deployment()
+	deploy1.JobID = job.ID
+	deploy1.JobCreateIndex = job.CreateIndex
+
+	deploy2 := mock.Deployment()
+	deploy2.JobID = job.ID
+	deploy2.JobCreateIndex = 11
+
+	require := require.New(t)
+
+	// Insert both deployments
+	err := state.UpsertDeployment(1001, deploy1)
+	require.Nil(err)
+
+	err = state.UpsertDeployment(1002, deploy2)
+	require.Nil(err)
+
+	ws := memdb.NewWatchSet()
+	// Should return both deployments
+	deploys, err := state.DeploymentsByJobID(ws, deploy1.Namespace, job.ID, true)
+	require.Nil(err)
+	require.Len(deploys, 2)
+
+	// Should only return deploy1
+	deploys, err = state.DeploymentsByJobID(ws, deploy1.Namespace, job.ID, false)
+	require.Nil(err)
+	require.Len(deploys, 1)
+	require.Equal(deploy1.ID, deploys[0].ID)
+}
+
 func TestStateStore_DeleteDeployment(t *testing.T) {
 	state := testStateStore(t)
 	d1 := mock.Deployment()

nomad/structs/structs.go

@@ -508,8 +508,8 @@ type EvalOptions struct {
 
 // JobSpecificRequest is used when we just need to specify a target job
 type JobSpecificRequest struct {
-	JobID     string
-	AllAllocs bool
+	JobID string
+	All   bool
 	QueryOptions
 }
 

website/source/api/jobs.html.md

@@ -966,6 +966,10 @@ The table below shows this endpoint's support for
 - `:job_id` `(string: <required>)` - Specifies the ID of the job (as specified in
   the job file during submission). This is specified as part of the path.
 
+- `all` `(bool: false)` - Specifies whether the list of deployments should
+  include deployments from a previously registered job with the same ID. This is
+  possible if the job is deregistered and reregistered.
+
 ### Sample Request
 
 ```text

website/source/docs/commands/job/deployments.html.md.erb

@@ -34,6 +34,9 @@ of a job to display the list of deployments for.
 
 * `-verbose`: Show full information.
 
+* `-all`: Display all deployments matching the job ID, even those from an
+   older instance of the job.
+
 ## Examples
 
 List the deployment for a particular job: