WorkerCoordination and GetSession API refactoring (#354)

* Possible api changes.

* Worker API update.

* Updating import alias name.

* Moving session down a level as a member of the GetSessionResponse.

* Marshal the auth flag into the GetSessionResponse proto instead of the session.

internal/servers/controller/handler.go

@@ -19,7 +19,9 @@ import (
 	"github.com/hashicorp/boundary/api"
 	"github.com/hashicorp/boundary/globals"
 	"github.com/hashicorp/boundary/internal/auth"
+	pb "github.com/hashicorp/boundary/internal/gen/controller/api/resources/sessions"
 	"github.com/hashicorp/boundary/internal/gen/controller/api/services"
+	wpbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services"
 	"github.com/hashicorp/boundary/internal/kms"
 	"github.com/hashicorp/boundary/internal/servers"
 	"github.com/hashicorp/boundary/internal/servers/controller/handlers/accounts"
@@ -342,7 +344,7 @@ func jobTestingHandler(c *Controller) http.Handler {
 			}
 		}
 
-		var workers []*services.WorkerInfo
+		var workers []*pb.WorkerInfo
 		repo, err := c.ServersRepoFn()
 		if err != nil {
 			errorResp(err)
@@ -354,7 +356,7 @@ func jobTestingHandler(c *Controller) http.Handler {
 			return
 		}
 		for _, v := range servers {
-			workers = append(workers, &services.WorkerInfo{Address: v.Address})
+			workers = append(workers, &pb.WorkerInfo{Address: v.Address})
 		}
 
 		wrapper, err := c.kms.GetWrapper(r.Context(), scope.Global.String(), kms.KeyPurposeSessions)
@@ -390,16 +392,18 @@ func jobTestingHandler(c *Controller) http.Handler {
 			return
 		}
 
-		ret := &services.ValidateSessionResponse{
-			Id:             jobId,
-			ScopeId:        scope.Global.String(),
-			UserId:         "u_1234567890",
-			Type:           "tcp",
-			Endpoint:       endpoint,
-			Certificate:    certBytes,
-			PrivateKey:     privKey,
-			WorkerInfo:     workers,
-			ExpirationTime: &timestamppb.Timestamp{Seconds: time.Now().Add(timeout).Unix()},
+		ret := &wpbs.GetSessionResponse{
+			Session: &pb.Session{
+				Id:             jobId,
+				ScopeId:        scope.Global.String(),
+				UserId:         "u_1234567890",
+				Type:           "tcp",
+				Endpoint:       endpoint,
+				Certificate:    certBytes,
+				PrivateKey:     privKey,
+				WorkerInfo:     workers,
+				ExpirationTime: &timestamppb.Timestamp{Seconds: time.Now().Add(timeout).Unix()},
+			},
 		}
 
 		marshaled, err := proto.Marshal(ret)
@@ -413,8 +417,8 @@ func jobTestingHandler(c *Controller) http.Handler {
 			return
 		}
 
-		ret.PrivateKey = nil
-		c.jobMap.Store(jobId, ret)
+		ret.Session.PrivateKey = nil
+		c.jobMap.Store(jobId, ret.GetSession())
 	})
 }
 

internal/servers/controller/handlers/workers/worker_service.go

@@ -5,7 +5,8 @@ import (
 	"sync"
 	"time"
 
-	pbs "github.com/hashicorp/boundary/internal/gen/controller/api/services"
+	pb "github.com/hashicorp/boundary/internal/gen/controller/api/resources/sessions"
+	pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services"
 	"github.com/hashicorp/boundary/internal/kms"
 	"github.com/hashicorp/boundary/internal/servers/controller/common"
 	"github.com/hashicorp/boundary/internal/sessions"
@@ -35,8 +36,11 @@ func NewWorkerServiceServer(logger hclog.Logger, repoFn common.ServersRepoFactor
 	}
 }
 
+var _ pbs.SessionServiceServer = &workerServiceServer{}
+var _ pbs.ServerCoordinationServiceServer = &workerServiceServer{}
+
 func (ws *workerServiceServer) Status(ctx context.Context, req *pbs.StatusRequest) (*pbs.StatusResponse, error) {
-	ws.logger.Trace("got status request from worker", "name", req.Worker.Name, "address", req.Worker.Address, "active_jobs", req.ActiveJobIds)
+	ws.logger.Trace("got status request from worker", "name", req.Worker.Name, "address", req.Worker.Address, "jobs", req.GetJobs())
 	ws.updateTimes.Store(req.Worker.Name, time.Now())
 	repo, err := ws.repoFn()
 	if err != nil {
@@ -53,41 +57,47 @@ func (ws *workerServiceServer) Status(ctx context.Context, req *pbs.StatusReques
 		Controllers: controllers,
 	}
 	ws.jobCancelMap.Range(func(key, value interface{}) bool {
-		ret.CancelJobIds = append(ret.CancelJobIds, key.(string))
+		ret.JobsRequests = append(ret.JobsRequests, &pbs.JobChangeRequest{
+			Job: &pbs.Job{
+				JobId: key.(string),
+				Type:  pbs.Job_JOBTYPE_SESSION,
+			},
+			RequestType: 0,
+		})
 		return true
 	})
-	for _, id := range ret.CancelJobIds {
-		ws.jobCancelMap.Delete(id)
+	for _, j := range ret.JobsRequests {
+		ws.jobCancelMap.Delete(j.GetJob().GetJobId())
 	}
 	return ret, nil
 }
 
-func (ws *workerServiceServer) ValidateSession(ctx context.Context, req *pbs.ValidateSessionRequest) (*pbs.ValidateSessionResponse, error) {
+func (ws *workerServiceServer) GetSession(ctx context.Context, req *pbs.GetSessionRequest) (*pbs.GetSessionResponse, error) {
 	ws.logger.Trace("got validate session request from worker", "job_id", req.GetId())
 
 	// Look up the job info
 	storedSessionInfo, loaded := ws.jobMap.LoadAndDelete(req.GetId())
 	if !loaded {
-		return &pbs.ValidateSessionResponse{}, status.Errorf(codes.PermissionDenied, "Unknown job ID: %v", req.GetId())
+		return nil, status.Errorf(codes.PermissionDenied, "Unknown job ID: %v", req.GetId())
 	}
-	sessionInfo := storedSessionInfo.(*pbs.ValidateSessionResponse)
+	sessionInfo := storedSessionInfo.(*pb.Session)
 
 	wrapper, err := ws.kms.GetWrapper(ctx, sessionInfo.ScopeId, kms.KeyPurposeSessions)
 	if err != nil {
-		return &pbs.ValidateSessionResponse{}, status.Errorf(codes.Internal, "Error getting sessions wrapper: %v", err)
+		return nil, status.Errorf(codes.Internal, "Error getting sessions wrapper: %v", err)
 	}
 
 	// Derive the private key, which should match. Deriving on both ends allows
 	// us to not store it in the DB.
 	_, privKey, err := sessions.DeriveED25519Key(wrapper, sessionInfo.GetUserId(), req.GetId())
 	if err != nil {
-		return &pbs.ValidateSessionResponse{}, status.Errorf(codes.Internal, "Error deriving session key: %v", err)
+		return nil, status.Errorf(codes.Internal, "Error deriving session key: %v", err)
 	}
 
 	if sessionInfo.ExpirationTime.GetSeconds() > 0 {
 		timeDiff := time.Until(sessionInfo.GetExpirationTime().AsTime())
 		if timeDiff < 0 {
-			return &pbs.ValidateSessionResponse{}, status.Errorf(codes.OutOfRange, "Session has already expired")
+			return nil, status.Errorf(codes.OutOfRange, "Session has already expired")
 		}
 		defer func() {
 			time.AfterFunc(timeDiff, func() {
@@ -97,5 +107,5 @@ func (ws *workerServiceServer) ValidateSession(ctx context.Context, req *pbs.Val
 	}
 
 	sessionInfo.PrivateKey = privKey
-	return sessionInfo, nil
+	return &pbs.GetSessionResponse{Session: sessionInfo}, nil
 }

internal/servers/controller/listeners.go

@@ -12,7 +12,7 @@ import (
 	"time"
 
 	"github.com/hashicorp/boundary/internal/cmd/base"
-	"github.com/hashicorp/boundary/internal/gen/controller/api/services"
+	pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services"
 	"github.com/hashicorp/boundary/internal/servers/controller/handlers/workers"
 	"github.com/hashicorp/go-alpnmux"
 	"github.com/hashicorp/go-multierror"
@@ -120,7 +120,9 @@ func (c *Controller) startListeners() error {
 			grpc.MaxRecvMsgSize(math.MaxInt32),
 			grpc.MaxSendMsgSize(math.MaxInt32),
 		)
-		services.RegisterWorkerServiceServer(workerServer, workers.NewWorkerServiceServer(c.logger.Named("worker-handler"), c.ServersRepoFn, c.workerStatusUpdateTimes, c.kms, c.jobMap))
+		workerService := workers.NewWorkerServiceServer(c.logger.Named("worker-handler"), c.ServersRepoFn, c.workerStatusUpdateTimes, c.kms, c.jobMap)
+		pbs.RegisterServerCoordinationServiceServer(workerServer, workerService)
+		pbs.RegisterSessionServiceServer(workerServer, workerService)
 
 		interceptor := newInterceptingListener(c, l)
 		ln.ALPNListener = interceptor