scheduler: add rpcs for actions on job and invocation.

scheduler/appengine/apiservers/scheduler.go

Index: scheduler/appengine/apiservers/scheduler.go
diff --git a/scheduler/appengine/apiservers/scheduler.go b/scheduler/appengine/apiservers/scheduler.go
index 3de1d4e5aed5995af4b3f0770ecc83cce455d774..5bfa0ae8d61740da84a8f1719dae1fbb357d212d 100644
--- a/scheduler/appengine/apiservers/scheduler.go
+++ b/scheduler/appengine/apiservers/scheduler.go
@@ -9,9 +9,12 @@ import (
 	"github.com/luci/luci-go/scheduler/appengine/catalog"
 	"github.com/luci/luci-go/scheduler/appengine/engine"
 	"github.com/luci/luci-go/scheduler/appengine/presentation"
+	"github.com/luci/luci-go/server/auth"
 	"golang.org/x/net/context"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
+
+	"github.com/golang/protobuf/ptypes/empty"
 )
 
 // SchedulerServer implements scheduler.Scheduler API.
@@ -24,6 +27,11 @@ var _ scheduler.SchedulerServer = (*SchedulerServer)(nil)
 
 // GetJobs fetches all jobs satisfying JobsRequest and visibility ACLs.
 func (s SchedulerServer) GetJobs(ctx context.Context, in *scheduler.JobsRequest) (*scheduler.JobsReply, error) {
+	if in.GetCursor() != "" {
+		// Paging in GetJobs isn't implemented until we have enough jobs to care.
+		// Until then, not empty cursor implies no more jobs to return.
+		return &scheduler.JobsReply{Jobs: []*scheduler.Job{}, NextCursor: ""}, nil
+	}
 	var ejobs []*engine.Job
 	var err error
 	if in.GetProject() == "" {
@@ -42,19 +50,21 @@ func (s SchedulerServer) GetJobs(ctx context.Context, in *scheduler.JobsRequest)
 			return nil, grpc.Errorf(codes.Internal, "failed to get traits: %s", err)
 		}
 		jobs[i] = &scheduler.Job{
-			Name:     ej.GetJobName(),
-			Project:  ej.ProjectID,
+			JobRef: &scheduler.JobRef{
+				Project: ej.ProjectID,
+				Job:     ej.GetJobName(),
+			},
 			Schedule: ej.Schedule,
 			State: &scheduler.JobState{
 				UiStatus: string(presentation.GetPublicStateKind(ej, traits)),
 			},
 		}
 	}
-	return &scheduler.JobsReply{Jobs: jobs}, nil
+	return &scheduler.JobsReply{Jobs: jobs, NextCursor: ""}, nil
 }
 
 func (s SchedulerServer) GetInvocations(ctx context.Context, in *scheduler.InvocationsRequest) (*scheduler.InvocationsReply, error) {
-	ejob, err := s.Engine.GetJob(ctx, in.GetProject()+"/"+in.GetJob())
+	ejob, err := s.Engine.GetJob(ctx, getJobId(in.GetJobRef()))
 	if err != nil {
 		return nil, grpc.Errorf(codes.Internal, "datastore error: %s", err)
 	}
@@ -74,9 +84,13 @@ func (s SchedulerServer) GetInvocations(ctx context.Context, in *scheduler.Invoc
 	invs := make([]*scheduler.Invocation, len(einvs))
 	for i, einv := range einvs {
 		invs[i] = &scheduler.Invocation{
-			Id:             einv.ID,
-			Job:            ejob.GetJobName(),
-			Project:        ejob.ProjectID,
+			InvocationRef: &scheduler.InvocationRef{
+				JobRef: &scheduler.JobRef{
+					Project: ejob.ProjectID,
+					Job:     ejob.GetJobName(),
+				},
+				InvocationId: einv.ID,
+			},
 			StartedTs:      einv.Started.UnixNano() / 1000,
 			TriggeredBy:    string(einv.TriggeredBy),
 			Status:         string(einv.Status),
@@ -90,3 +104,52 @@ func (s SchedulerServer) GetInvocations(ctx context.Context, in *scheduler.Invoc
 	}
 	return &scheduler.InvocationsReply{Invocations: invs, NextCursor: cursor}, nil
 }
+
+//// Actions.
+
+func (s SchedulerServer) PauseJob(ctx context.Context, in *scheduler.JobRef) (*empty.Empty, error) {
+	return runAction(ctx, in, func() error {
+		return s.Engine.PauseJob(ctx, getJobId(in), auth.CurrentIdentity(ctx))
+	})
+}
+
+func (s SchedulerServer) ResumeJob(ctx context.Context, in *scheduler.JobRef) (*empty.Empty, error) {
+	return runAction(ctx, in, func() error {
+		return s.Engine.ResumeJob(ctx, getJobId(in), auth.CurrentIdentity(ctx))
+	})
+}
+
+func (s SchedulerServer) AbortJob(ctx context.Context, in *scheduler.JobRef) (*empty.Empty, error) {
+	return runAction(ctx, in, func() error {
+		return s.Engine.AbortJob(ctx, getJobId(in), auth.CurrentIdentity(ctx))
+	})
+}
+
+func (s SchedulerServer) AbortInvocation(ctx context.Context, in *scheduler.InvocationRef) (
+	*empty.Empty, error) {
+	return runAction(ctx, in.GetJobRef(), func() error {
+		return s.Engine.AbortInvocation(ctx, getJobId(in.GetJobRef()), in.GetInvocationId(), auth.CurrentIdentity(ctx))
+	})
+}
+
+//// Private helpers.
+
+func runAction(ctx context.Context, jobRef *scheduler.JobRef, action func() error) (*empty.Empty, error) {
+	if !presentation.IsJobOwner(ctx, jobRef.GetProject(), jobRef.GetJob()) {
+		return nil, grpc.Errorf(codes.PermissionDenied, "No permission to execute action")
+	}
+	switch err := action(); {
+	case err == nil:
+		return &empty.Empty{}, nil
+	case err == engine.ErrNoSuchJob:
+		return nil, grpc.Errorf(codes.NotFound, "no such job")
+	case err == engine.ErrNoSuchInvocation:
+		return nil, grpc.Errorf(codes.NotFound, "no such invocation")
+	default:
+		return nil, grpc.Errorf(codes.Internal, "internal error: %s", err)
+	}
+}
+
+func getJobId(jobRef *scheduler.JobRef) string {
+	return jobRef.GetProject() + "/" + jobRef.GetJob()
+}