consul: restore consul token when reverting a job (#15996)

* consul: reset consul token on job during registration of a reversion

* e2e: add test for reverting a job with a consul service

* cl: fixup cl entry

.changelog/15996.txt

@@ -0,0 +1,3 @@
+```release-note:bug
+consul: Fixed a bug where consul token was not respected when reverting a job
+```

e2e/consul/input/service_reversion.nomad

@@ -0,0 +1,36 @@
+variable "service" {
+  type = string
+}
+
+job "service-reversion" {
+  datacenters = ["dc1"]
+  type        = "service"
+
+  constraint {
+    attribute = "${attr.kernel.name}"
+    value     = "linux"
+  }
+
+  group "sleep" {
+
+    service {
+      name = "${var.service}"
+    }
+
+    task "busybox" {
+      driver = "docker"
+
+      config {
+        image   = "busybox:1"
+        command = "sleep"
+        args    = ["infinity"]
+      }
+
+      resources {
+        cpu    = 16
+        memory = 32
+        disk   = 64
+      }
+    }
+  }
+}

e2e/consul/service_revert_test.go

@@ -0,0 +1,85 @@
+package consul
+
+import (
+	"context"
+	"testing"
+
+	"github.com/hashicorp/nomad/e2e/e2eutil"
+	"github.com/hashicorp/nomad/helper/uuid"
+	"github.com/hashicorp/nomad/nomad/structs"
+	"github.com/shoenig/test/must"
+)
+
+func TestConsul(t *testing.T) {
+	// todo: migrate the remaining consul tests
+
+	nomad := e2eutil.NomadClient(t)
+
+	e2eutil.WaitForLeader(t, nomad)
+	e2eutil.WaitForNodesReady(t, nomad, 1)
+
+	t.Run("testServiceReversion", testServiceReversion)
+}
+
+// testServiceReversion asserts we can
+// - submit a job with a service
+// - update that job and modify service
+// - revert the job, restoring the original service
+func testServiceReversion(t *testing.T) {
+	const jobFile = "./input/service_reversion.nomad"
+	jobID := "service-reversion-" + uuid.Short()
+	jobIDs := []string{jobID}
+
+	// Defer a cleanup function to remove the job. This will trigger if the
+	// test fails, unless the cancel function is called.
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	defer e2eutil.CleanupJobsAndGCWithContext(t, ctx, &jobIDs)
+
+	// initial register of job, with service="one"
+	vars := []string{"-var", "service=one"}
+	err := e2eutil.RegisterWithArgs(jobID, jobFile, vars...)
+	must.NoError(t, err)
+
+	// wait for job to be running
+	err = e2eutil.WaitForAllocStatusExpected(jobID, "", []string{structs.AllocClientStatusRunning})
+	must.NoError(t, err)
+
+	// get our consul client
+	consulClient := e2eutil.ConsulClient(t)
+
+	assertService := func(name string, count int) {
+		services, _, consulErr := consulClient.Catalog().Service(name, "", nil)
+		must.NoError(t, consulErr)
+		must.Len(t, count, services, must.Sprintf("expected %d instances of %s, got %d", count, name, len(services)))
+	}
+
+	// query services, assert 1 instance of "one"
+	assertService("one", 1)
+	assertService("two", 0)
+
+	// second register of job, with service="two"
+	vars = []string{"-var", "service=two"}
+	err = e2eutil.RegisterWithArgs(jobID, jobFile, vars...)
+	must.NoError(t, err)
+
+	// wait for job to be running
+	err = e2eutil.WaitForAllocStatusExpected(jobID, "", []string{structs.AllocClientStatusRunning})
+	must.NoError(t, err)
+
+	// query services, assert 0 instance of "one" (replaced), 1 of "two"
+	assertService("one", 0)
+	assertService("two", 1)
+
+	// now revert our job back to version 0
+	err = e2eutil.Revert(jobID, jobFile, 0)
+	must.NoError(t, err)
+
+	// wait for job to be running
+	err = e2eutil.WaitForAllocStatusExpected(jobID, "", []string{structs.AllocClientStatusRunning})
+	must.NoError(t, err)
+
+	// query services, assert 1 instance of "one" (reverted), 1 of "two" (removed)
+	assertService("one", 1)
+	assertService("two", 0)
+}

e2e/e2eutil/job.go

@@ -7,6 +7,7 @@ import (
 	"io/ioutil"
 	"os/exec"
 	"regexp"
+	"strconv"
 	"strings"
 	"testing"
 	"time"
@@ -19,7 +20,7 @@ import (
 func Register(jobID, jobFilePath string) error {
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
 	defer cancel()
-	return register(jobID, jobFilePath, exec.CommandContext(ctx, "nomad", "job", "run", "-detach", "-"))
+	return execCmd(jobID, jobFilePath, exec.CommandContext(ctx, "nomad", "job", "run", "-detach", "-"))
 }
 
 // RegisterWithArgs registers a jobspec from a file but with a unique ID. The
@@ -34,11 +35,18 @@ func RegisterWithArgs(jobID, jobFilePath string, args ...string) error {
 	baseArgs = append(baseArgs, "-")
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
 	defer cancel()
+	return execCmd(jobID, jobFilePath, exec.CommandContext(ctx, "nomad", baseArgs...))
+}
 
-	return register(jobID, jobFilePath, exec.CommandContext(ctx, "nomad", baseArgs...))
+// Revert reverts the job to the given version.
+func Revert(jobID, jobFilePath string, version int) error {
+	args := []string{"job", "revert", "-detach", jobID, strconv.Itoa(version)}
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
+	defer cancel()
+	return execCmd(jobID, jobFilePath, exec.CommandContext(ctx, "nomad", args...))
 }
 
-func register(jobID, jobFilePath string, cmd *exec.Cmd) error {
+func execCmd(jobID, jobFilePath string, cmd *exec.Cmd) error {
 	stdin, err := cmd.StdinPipe()
 	if err != nil {
 		return fmt.Errorf("could not open stdin?: %w", err)
@@ -49,14 +57,16 @@ func register(jobID, jobFilePath string, cmd *exec.Cmd) error {
 		return fmt.Errorf("could not open job file: %w", err)
 	}
 
-	// hack off the first line to replace with our unique ID
+	// hack off the job block to replace with our unique ID
 	var re = regexp.MustCompile(`(?m)^job ".*" \{`)
 	jobspec := re.ReplaceAllString(string(content),
 		fmt.Sprintf("job \"%s\" {", jobID))
 
 	go func() {
-		defer stdin.Close()
-		io.WriteString(stdin, jobspec)
+		defer func() {
+			_ = stdin.Close()
+		}()
+		_, _ = io.WriteString(stdin, jobspec)
 	}()
 
 	out, err := cmd.CombinedOutput()

nomad/job_endpoint.go

@@ -643,8 +643,8 @@ func (j *Job) Revert(args *structs.JobRevertRequest, reply *structs.JobRegisterR
 
 	// Build the register request
 	revJob := jobV.Copy()
-	// Use Vault Token from revert request to perform registration of reverted job.
-	revJob.VaultToken = args.VaultToken
+	revJob.VaultToken = args.VaultToken   // use vault token from revert to perform (re)registration
+	revJob.ConsulToken = args.ConsulToken // use consul token from revert to perform (re)registration
 	reg := &structs.JobRegisterRequest{
 		Job:          revJob,
 		WriteRequest: args.WriteRequest,