Scaling API changes

acl/policy.go

@@ -11,10 +11,11 @@ const (
 	// The following levels are the only valid values for the `policy = "read"` stanza.
 	// When policies are merged together, the most privilege is granted, except for deny
 	// which always takes precedence and supercedes.
-	PolicyDeny  = "deny"
-	PolicyRead  = "read"
-	PolicyList  = "list"
-	PolicyWrite = "write"
+	PolicyDeny       = "deny"
+	PolicyRead       = "read"
+	PolicyList       = "list"
+	PolicyWrite      = "write"
+	PolicyAutoscaler = "autoscaler"
 )
 
 const (
@@ -23,22 +24,26 @@ const (
 	// combined we take the union of all capabilities. If the deny capability is present, it
 	// takes precedence and overwrites all other capabilities.
 
-	NamespaceCapabilityDeny              = "deny"
-	NamespaceCapabilityListJobs          = "list-jobs"
-	NamespaceCapabilityReadJob           = "read-job"
-	NamespaceCapabilitySubmitJob         = "submit-job"
-	NamespaceCapabilityDispatchJob       = "dispatch-job"
-	NamespaceCapabilityReadLogs          = "read-logs"
-	NamespaceCapabilityReadFS            = "read-fs"
-	NamespaceCapabilityAllocExec         = "alloc-exec"
-	NamespaceCapabilityAllocNodeExec     = "alloc-node-exec"
-	NamespaceCapabilityAllocLifecycle    = "alloc-lifecycle"
-	NamespaceCapabilitySentinelOverride  = "sentinel-override"
-	NamespaceCapabilityCSIRegisterPlugin = "csi-register-plugin"
-	NamespaceCapabilityCSIWriteVolume    = "csi-write-volume"
-	NamespaceCapabilityCSIReadVolume     = "csi-read-volume"
-	NamespaceCapabilityCSIListVolume     = "csi-list-volume"
-	NamespaceCapabilityCSIMountVolume    = "csi-mount-volume"
+	NamespaceCapabilityDeny                = "deny"
+	NamespaceCapabilityListJobs            = "list-jobs"
+	NamespaceCapabilityReadJob             = "read-job"
+	NamespaceCapabilitySubmitJob           = "submit-job"
+	NamespaceCapabilityDispatchJob         = "dispatch-job"
+	NamespaceCapabilityReadLogs            = "read-logs"
+	NamespaceCapabilityReadFS              = "read-fs"
+	NamespaceCapabilityAllocExec           = "alloc-exec"
+	NamespaceCapabilityAllocNodeExec       = "alloc-node-exec"
+	NamespaceCapabilityAllocLifecycle      = "alloc-lifecycle"
+	NamespaceCapabilitySentinelOverride    = "sentinel-override"
+	NamespaceCapabilityCSIRegisterPlugin   = "csi-register-plugin"
+	NamespaceCapabilityCSIWriteVolume      = "csi-write-volume"
+	NamespaceCapabilityCSIReadVolume       = "csi-read-volume"
+	NamespaceCapabilityCSIListVolume       = "csi-list-volume"
+	NamespaceCapabilityCSIMountVolume      = "csi-mount-volume"
+	NamespaceCapabilityListScalingPolicies = "list-scaling-policies"
+	NamespaceCapabilityReadScalingPolicy   = "read-scaling-policy"
+	NamespaceCapabilityReadJobScaling      = "read-job-scaling"
+	NamespaceCapabilityScaleJob            = "scale-job"
 )
 
 var (
@@ -121,7 +126,7 @@ type PluginPolicy struct {
 // isPolicyValid makes sure the given string matches one of the valid policies.
 func isPolicyValid(policy string) bool {
 	switch policy {
-	case PolicyDeny, PolicyRead, PolicyWrite:
+	case PolicyDeny, PolicyRead, PolicyWrite, PolicyAutoscaler:
 		return true
 	default:
 		return false
@@ -144,7 +149,8 @@ func isNamespaceCapabilityValid(cap string) bool {
 		NamespaceCapabilitySubmitJob, NamespaceCapabilityDispatchJob, NamespaceCapabilityReadLogs,
 		NamespaceCapabilityReadFS, NamespaceCapabilityAllocLifecycle,
 		NamespaceCapabilityAllocExec, NamespaceCapabilityAllocNodeExec,
-		NamespaceCapabilityCSIReadVolume, NamespaceCapabilityCSIWriteVolume, NamespaceCapabilityCSIListVolume, NamespaceCapabilityCSIMountVolume, NamespaceCapabilityCSIRegisterPlugin:
+		NamespaceCapabilityCSIReadVolume, NamespaceCapabilityCSIWriteVolume, NamespaceCapabilityCSIListVolume, NamespaceCapabilityCSIMountVolume, NamespaceCapabilityCSIRegisterPlugin,
+		NamespaceCapabilityListScalingPolicies, NamespaceCapabilityReadScalingPolicy, NamespaceCapabilityReadJobScaling, NamespaceCapabilityScaleJob:
 		return true
 	// Separate the enterprise-only capabilities
 	case NamespaceCapabilitySentinelOverride:
@@ -162,9 +168,13 @@ func expandNamespacePolicy(policy string) []string {
 		NamespaceCapabilityReadJob,
 		NamespaceCapabilityCSIListVolume,
 		NamespaceCapabilityCSIReadVolume,
+		NamespaceCapabilityReadJobScaling,
+		NamespaceCapabilityListScalingPolicies,
+		NamespaceCapabilityReadScalingPolicy,
 	}
 
 	write := append(read, []string{
+		NamespaceCapabilityScaleJob,
 		NamespaceCapabilitySubmitJob,
 		NamespaceCapabilityDispatchJob,
 		NamespaceCapabilityReadLogs,
@@ -182,6 +192,13 @@ func expandNamespacePolicy(policy string) []string {
 		return read
 	case PolicyWrite:
 		return write
+	case PolicyAutoscaler:
+		return []string{
+			NamespaceCapabilityListScalingPolicies,
+			NamespaceCapabilityReadScalingPolicy,
+			NamespaceCapabilityReadJobScaling,
+			NamespaceCapabilityScaleJob,
+		}
 	default:
 		return nil
 	}

acl/policy_test.go

@@ -32,6 +32,9 @@ func TestParse(t *testing.T) {
 							NamespaceCapabilityReadJob,
 							NamespaceCapabilityCSIListVolume,
 							NamespaceCapabilityCSIReadVolume,
+							NamespaceCapabilityReadJobScaling,
+							NamespaceCapabilityListScalingPolicies,
+							NamespaceCapabilityReadScalingPolicy,
 						},
 					},
 				},
@@ -48,6 +51,9 @@ func TestParse(t *testing.T) {
 			namespace "secret" {
 				capabilities = ["deny", "read-logs"]
 			}
+			namespace "autoscaler" {
+				policy = "autoscaler"
+			}
 			agent {
 				policy = "read"
 			}
@@ -75,6 +81,9 @@ func TestParse(t *testing.T) {
 							NamespaceCapabilityReadJob,
 							NamespaceCapabilityCSIListVolume,
 							NamespaceCapabilityCSIReadVolume,
+							NamespaceCapabilityReadJobScaling,
+							NamespaceCapabilityListScalingPolicies,
+							NamespaceCapabilityReadScalingPolicy,
 						},
 					},
 					{
@@ -85,6 +94,10 @@ func TestParse(t *testing.T) {
 							NamespaceCapabilityReadJob,
 							NamespaceCapabilityCSIListVolume,
 							NamespaceCapabilityCSIReadVolume,
+							NamespaceCapabilityReadJobScaling,
+							NamespaceCapabilityListScalingPolicies,
+							NamespaceCapabilityReadScalingPolicy,
+							NamespaceCapabilityScaleJob,
 							NamespaceCapabilitySubmitJob,
 							NamespaceCapabilityDispatchJob,
 							NamespaceCapabilityReadLogs,
@@ -102,6 +115,16 @@ func TestParse(t *testing.T) {
 							NamespaceCapabilityReadLogs,
 						},
 					},
+					{
+						Name:   "autoscaler",
+						Policy: PolicyAutoscaler,
+						Capabilities: []string{
+							NamespaceCapabilityListScalingPolicies,
+							NamespaceCapabilityReadScalingPolicy,
+							NamespaceCapabilityReadJobScaling,
+							NamespaceCapabilityScaleJob,
+						},
+					},
 				},
 				Agent: &AgentPolicy{
 					Policy: PolicyRead,

api/jobs.go

@@ -153,6 +153,43 @@ func (j *Jobs) Info(jobID string, q *QueryOptions) (*Job, *QueryMeta, error) {
 	return &resp, qm, nil
 }
 
+// Scale is used to retrieve information about a particular
+// job given its unique ID.
+func (j *Jobs) Scale(jobID, group string, count *int,
+	reason, error *string, meta map[string]interface{}, q *WriteOptions) (*JobRegisterResponse, *WriteMeta, error) {
+	var count64 *int64
+	if count != nil {
+		count64 = int64ToPtr(int64(*count))
+	}
+	req := &ScalingRequest{
+		Count: count64,
+		Target: map[string]string{
+			"Job":   jobID,
+			"Group": group,
+		},
+		Error:  error,
+		Reason: reason,
+		Meta:   meta,
+	}
+	var resp JobRegisterResponse
+	qm, err := j.client.write(fmt.Sprintf("/v1/job/%s/scale", url.PathEscape(jobID)), req, &resp, q)
+	if err != nil {
+		return nil, nil, err
+	}
+	return &resp, qm, nil
+}
+
+// ScaleStatus is used to retrieve information about a particular
+// job given its unique ID.
+func (j *Jobs) ScaleStatus(jobID string, q *QueryOptions) (*JobScaleStatusResponse, *QueryMeta, error) {
+	var resp JobScaleStatusResponse
+	qm, err := j.client.query(fmt.Sprintf("/v1/job/%s/scale", url.PathEscape(jobID)), &resp, q)
+	if err != nil {
+		return nil, nil, err
+	}
+	return &resp, qm, nil
+}
+
 // Versions is used to retrieve all versions of a particular job given its
 // unique ID.
 func (j *Jobs) Versions(jobID string, diffs bool, q *QueryOptions) ([]*Job, []*JobDiff, *QueryMeta, error) {

api/jobs_test.go

@@ -909,6 +909,46 @@ func TestJobs_Info(t *testing.T) {
 	}
 }
 
+func TestJobs_ScaleInvalidAction(t *testing.T) {
+	t.Parallel()
+	require := require.New(t)
+
+	c, s := makeClient(t, nil, nil)
+	defer s.Stop()
+	jobs := c.Jobs()
+
+	// Check if invalid inputs fail
+	tests := []struct {
+		jobID string
+		group string
+		value int
+		want  string
+	}{
+		{"", "", 1, "404"},
+		{"i-dont-exist", "", 1, "400"},
+		{"", "i-dont-exist", 1, "404"},
+		{"i-dont-exist", "me-neither", 1, "404"},
+	}
+	for _, test := range tests {
+		_, _, err := jobs.Scale(test.jobID, test.group, &test.value, stringToPtr("reason"), nil, nil, nil)
+		require.Errorf(err, "expected jobs.Scale(%s, %s) to fail", test.jobID, test.group)
+		require.Containsf(err.Error(), test.want, "jobs.Scale(%s, %s) error doesn't contain %s, got: %s", test.jobID, test.group, test.want, err)
+	}
+
+	// Register test job
+	job := testJob()
+	job.ID = stringToPtr("TestJobs_Scale")
+	_, wm, err := jobs.Register(job, nil)
+	require.NoError(err)
+	assertWriteMeta(t, wm)
+
+	// Perform a scaling action with bad group name, verify error
+	_, _, err = jobs.Scale(*job.ID, "incorrect-group-name", intToPtr(2),
+		stringToPtr("because"), nil, nil, nil)
+	require.Error(err)
+	require.Contains(err.Error(), "does not exist")
+}
+
 func TestJobs_Versions(t *testing.T) {
 	t.Parallel()
 	c, s := makeClient(t, nil, nil)
@@ -1533,3 +1573,82 @@ func TestJobs_AddSpread(t *testing.T) {
 		t.Fatalf("expect: %#v, got: %#v", expect, job.Spreads)
 	}
 }
+
+// TestJobs_ScaleAction tests the scale target for task group count
+func TestJobs_ScaleAction(t *testing.T) {
+	t.Parallel()
+	require := require.New(t)
+
+	c, s := makeClient(t, nil, nil)
+	defer s.Stop()
+	jobs := c.Jobs()
+
+	id := "job-id/with\\troublesome:characters\n?&字\000"
+	job := testJobWithScalingPolicy()
+	job.ID = &id
+	groupName := *job.TaskGroups[0].Name
+	groupCount := *job.TaskGroups[0].Count
+
+	// Trying to scale against a target before it exists returns an error
+	_, _, err := jobs.Scale(id, "missing", intToPtr(groupCount+1), stringToPtr("this won't work"), nil, nil, nil)
+	require.Error(err)
+	require.Contains(err.Error(), "not found")
+
+	// Register the job
+	_, wm, err := jobs.Register(job, nil)
+	require.NoError(err)
+	assertWriteMeta(t, wm)
+
+	// Perform scaling action
+	newCount := groupCount + 1
+	resp1, wm, err := jobs.Scale(id, groupName,
+		intToPtr(newCount), stringToPtr("need more instances"), nil, nil, nil)
+
+	require.NoError(err)
+	require.NotNil(resp1)
+	require.NotEmpty(resp1.EvalID)
+	assertWriteMeta(t, wm)
+
+	// Query the job again
+	resp, _, err := jobs.Info(*job.ID, nil)
+	require.NoError(err)
+	require.Equal(*resp.TaskGroups[0].Count, newCount)
+
+	// TODO: check if reason is stored
+}
+
+// TestJobs_ScaleStatus tests the /scale status endpoint for task group count
+func TestJobs_ScaleStatus(t *testing.T) {
+	t.Parallel()
+
+	require := require.New(t)
+
+	c, s := makeClient(t, nil, nil)
+	defer s.Stop()
+	jobs := c.Jobs()
+
+	// Trying to retrieve a status before it exists returns an error
+	id := "job-id/with\\troublesome:characters\n?&字\000"
+	_, _, err := jobs.ScaleStatus(id, nil)
+	require.Error(err)
+	require.Contains(err.Error(), "not found")
+
+	// Register the job
+	job := testJobWithScalingPolicy()
+	job.ID = &id
+	groupName := *job.TaskGroups[0].Name
+	groupCount := *job.TaskGroups[0].Count
+	_, wm, err := jobs.Register(job, nil)
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+	assertWriteMeta(t, wm)
+
+	// Query the scaling endpoint and verify success
+	result, qm, err := jobs.ScaleStatus(id, nil)
+	require.NoError(err)
+	assertQueryMeta(t, qm)
+
+	// Check that the result is what we expect
+	require.Equal(groupCount, result.TaskGroups[groupName].Desired)
+}