refact(api): move worker handlers

internal/daemon/controller/handlers/targets/tcp/target_service_test.go

@@ -16,11 +16,11 @@ import (
 	"github.com/hashicorp/boundary/internal/authtoken"
 	"github.com/hashicorp/boundary/internal/credential"
 	"github.com/hashicorp/boundary/internal/credential/vault"
+	cluster "github.com/hashicorp/boundary/internal/daemon/cluster/handlers"
 	"github.com/hashicorp/boundary/internal/daemon/controller/auth"
 	"github.com/hashicorp/boundary/internal/daemon/controller/handlers"
 	"github.com/hashicorp/boundary/internal/daemon/controller/handlers/credentiallibraries"
 	"github.com/hashicorp/boundary/internal/daemon/controller/handlers/targets"
-	"github.com/hashicorp/boundary/internal/daemon/controller/handlers/workers"
 	"github.com/hashicorp/boundary/internal/db"
 	pbs "github.com/hashicorp/boundary/internal/gen/controller/api/services"
 	authpb "github.com/hashicorp/boundary/internal/gen/controller/auth"
@@ -2787,7 +2787,7 @@ func TestAuthorizeSession(t *testing.T) {
 			require.NoError(t, err)
 
 			// Tell our DB that there is a worker ready to serve the data
-			workerService := workers.NewWorkerServiceServer(serversRepoFn, sessionRepoFn, connectionRepoFn, &sync.Map{}, kms)
+			workerService := cluster.NewWorkerServiceServer(serversRepoFn, sessionRepoFn, connectionRepoFn, &sync.Map{}, kms)
 			_, err = workerService.Status(ctx, &spbs.StatusRequest{
 				Worker: &spb.Server{
 					PrivateId: "w_1234567890",
@@ -3075,7 +3075,7 @@ func TestAuthorizeSessionTypedCredentials(t *testing.T) {
 			require.NoError(t, err)
 
 			// Tell our DB that there is a worker ready to serve the data
-			workerService := workers.NewWorkerServiceServer(serversRepoFn, sessionRepoFn, connectionRepoFn, &sync.Map{}, kms)
+			workerService := cluster.NewWorkerServiceServer(serversRepoFn, sessionRepoFn, connectionRepoFn, &sync.Map{}, kms)
 			_, err = workerService.Status(ctx, &spbs.StatusRequest{
 				Worker: &spb.Server{
 					PrivateId: "w_1234567890",
@@ -3206,7 +3206,7 @@ func TestAuthorizeSession_Errors(t *testing.T) {
 	store := vault.TestCredentialStore(t, conn, wrapper, proj.GetPublicId(), v.Addr, tok, sec.Auth.Accessor)
 
 	workerExists := func(tar target.Target) (version uint32) {
-		workerService := workers.NewWorkerServiceServer(serversRepoFn, sessionRepoFn, connectionRepoFn, &sync.Map{}, kms)
+		workerService := cluster.NewWorkerServiceServer(serversRepoFn, sessionRepoFn, connectionRepoFn, &sync.Map{}, kms)
 		_, err := workerService.Status(context.Background(), &spbs.StatusRequest{
 			Worker: &spb.Server{
 				PrivateId: "w_1234567890",

internal/daemon/controller/handlers/workers/multihop_service.go

@@ -1,46 +0,0 @@
-package workers
-
-import (
-	"context"
-	"fmt"
-
-	pbs "github.com/hashicorp/nodeenrollment/multihop"
-	"github.com/hashicorp/nodeenrollment/nodeauth"
-	"github.com/hashicorp/nodeenrollment/registration"
-	"github.com/hashicorp/nodeenrollment/tls"
-	"github.com/hashicorp/nodeenrollment/types"
-)
-
-type multihopServiceServer struct {
-	pbs.UnimplementedMultihopServiceServer
-
-	currentParams nodeauth.CurrentParameterFactory
-}
-
-func NewMultihopServiceServer(
-	currentParams nodeauth.CurrentParameterFactory,
-) *multihopServiceServer {
-	return &multihopServiceServer{
-		currentParams: currentParams,
-	}
-}
-
-var _ pbs.MultihopServiceServer = (*multihopServiceServer)(nil)
-
-func (m *multihopServiceServer) FetchNodeCredentials(ctx context.Context, req *types.FetchNodeCredentialsRequest) (*types.FetchNodeCredentialsResponse, error) {
-	const op = "workers.(multihopServiceServer).FetchNodeCredentials"
-	_, storage, opt, err := m.currentParams()
-	if err != nil {
-		return nil, fmt.Errorf("%s: error getting current parameters: %w", op, err)
-	}
-	return registration.FetchNodeCredentials(ctx, storage, req, opt...)
-}
-
-func (m *multihopServiceServer) GenerateServerCertificates(ctx context.Context, req *types.GenerateServerCertificatesRequest) (*types.GenerateServerCertificatesResponse, error) {
-	const op = "workers.(multihopServiceServer).GenerateServerCertificates"
-	_, storage, opt, err := m.currentParams()
-	if err != nil {
-		return nil, fmt.Errorf("%s: error getting current parameters: %w", op, err)
-	}
-	return tls.GenerateServerCertificates(ctx, storage, req, opt...)
-}

internal/daemon/controller/handlers/workers/worker_proxy_service.go

@@ -1,59 +0,0 @@
-package workers
-
-import (
-	"context"
-	"sync/atomic"
-
-	pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services"
-)
-
-type workerProxyServiceServer struct {
-	pbs.UnimplementedServerCoordinationServiceServer
-	pbs.UnimplementedSessionServiceServer
-
-	scsClient *atomic.Value
-	ssClient  *atomic.Value
-}
-
-func NewWorkerProxyServiceServer(
-	scsClient *atomic.Value,
-	ssClient *atomic.Value,
-) *workerProxyServiceServer {
-	return &workerProxyServiceServer{
-		scsClient: scsClient,
-		ssClient:  ssClient,
-	}
-}
-
-var (
-	_ pbs.ServerCoordinationServiceServer = &workerProxyServiceServer{}
-	_ pbs.SessionServiceServer            = &workerProxyServiceServer{}
-)
-
-func (ws *workerProxyServiceServer) Status(ctx context.Context, req *pbs.StatusRequest) (*pbs.StatusResponse, error) {
-	return ws.scsClient.Load().(pbs.ServerCoordinationServiceClient).Status(ctx, req)
-}
-
-func (ws *workerProxyServiceServer) LookupSession(ctx context.Context, req *pbs.LookupSessionRequest) (*pbs.LookupSessionResponse, error) {
-	return ws.ssClient.Load().(pbs.SessionServiceClient).LookupSession(ctx, req)
-}
-
-func (ws *workerProxyServiceServer) CancelSession(ctx context.Context, req *pbs.CancelSessionRequest) (*pbs.CancelSessionResponse, error) {
-	return ws.ssClient.Load().(pbs.SessionServiceClient).CancelSession(ctx, req)
-}
-
-func (ws *workerProxyServiceServer) ActivateSession(ctx context.Context, req *pbs.ActivateSessionRequest) (*pbs.ActivateSessionResponse, error) {
-	return ws.ssClient.Load().(pbs.SessionServiceClient).ActivateSession(ctx, req)
-}
-
-func (ws *workerProxyServiceServer) AuthorizeConnection(ctx context.Context, req *pbs.AuthorizeConnectionRequest) (*pbs.AuthorizeConnectionResponse, error) {
-	return ws.ssClient.Load().(pbs.SessionServiceClient).AuthorizeConnection(ctx, req)
-}
-
-func (ws *workerProxyServiceServer) ConnectConnection(ctx context.Context, req *pbs.ConnectConnectionRequest) (*pbs.ConnectConnectionResponse, error) {
-	return ws.ssClient.Load().(pbs.SessionServiceClient).ConnectConnection(ctx, req)
-}
-
-func (ws *workerProxyServiceServer) CloseConnection(ctx context.Context, req *pbs.CloseConnectionRequest) (*pbs.CloseConnectionResponse, error) {
-	return ws.ssClient.Load().(pbs.SessionServiceClient).CloseConnection(ctx, req)
-}

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

@@ -1,456 +0,0 @@
-package workers
-
-import (
-	"context"
-	"errors"
-	"fmt"
-	"sync"
-	"time"
-
-	"github.com/hashicorp/boundary/internal/daemon/controller/common"
-	"github.com/hashicorp/boundary/internal/daemon/controller/handlers"
-	pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services"
-	"github.com/hashicorp/boundary/internal/kms"
-	"github.com/hashicorp/boundary/internal/observability/event"
-	"github.com/hashicorp/boundary/internal/servers"
-	"github.com/hashicorp/boundary/internal/session"
-	"github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/targets"
-	"github.com/hashicorp/go-bexpr"
-	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/status"
-	"google.golang.org/protobuf/proto"
-)
-
-type workerServiceServer struct {
-	pbs.UnimplementedServerCoordinationServiceServer
-	pbs.UnimplementedSessionServiceServer
-
-	serversRepoFn    common.ServersRepoFactory
-	sessionRepoFn    common.SessionRepoFactory
-	connectionRepoFn common.ConnectionRepoFactory
-	updateTimes      *sync.Map
-	kms              *kms.Kms
-}
-
-func NewWorkerServiceServer(
-	serversRepoFn common.ServersRepoFactory,
-	sessionRepoFn common.SessionRepoFactory,
-	connectionRepoFn common.ConnectionRepoFactory,
-	updateTimes *sync.Map,
-	kms *kms.Kms,
-) *workerServiceServer {
-	return &workerServiceServer{
-		serversRepoFn:    serversRepoFn,
-		sessionRepoFn:    sessionRepoFn,
-		connectionRepoFn: connectionRepoFn,
-		updateTimes:      updateTimes,
-		kms:              kms,
-	}
-}
-
-var (
-	_ pbs.SessionServiceServer            = &workerServiceServer{}
-	_ pbs.ServerCoordinationServiceServer = &workerServiceServer{}
-)
-
-func (ws *workerServiceServer) Status(ctx context.Context, req *pbs.StatusRequest) (*pbs.StatusResponse, error) {
-	const op = "workers.(workerServiceServer).Status"
-	// TODO: on the worker, if we get errors back from this repeatedly, do we
-	// terminate all sessions since we can't know if they were canceled?
-	ws.updateTimes.Store(req.Worker.PrivateId, time.Now())
-	serverRepo, err := ws.serversRepoFn()
-	if err != nil {
-		event.WriteError(ctx, op, err, event.WithInfoMsg("error getting servers repo"))
-		return &pbs.StatusResponse{}, status.Errorf(codes.Internal, "Error acquiring repo to store worker status: %v", err)
-	}
-	sessRepo, err := ws.sessionRepoFn()
-	connectionRepo, err := ws.connectionRepoFn()
-	if err != nil {
-		event.WriteError(ctx, op, err, event.WithInfoMsg("error getting sessions repo"))
-		return &pbs.StatusResponse{}, status.Errorf(codes.Internal, "Error acquiring repo to query session status: %v", err)
-	}
-
-	reqServer := req.GetWorker()
-	// Convert API tags to storage tags
-	workerTags := make([]*servers.Tag, 0, len(reqServer.Tags))
-	for k, v := range reqServer.Tags {
-		for _, val := range v.GetValues() {
-			workerTags = append(workerTags, &servers.Tag{
-				Key:   k,
-				Value: val,
-			})
-		}
-	}
-
-	wConf := servers.NewWorkerStatus(reqServer.PrivateId,
-		// TODO: Change the name to its own field to make this explicit.
-		servers.WithName(reqServer.PrivateId),
-		servers.WithAddress(reqServer.Address),
-		servers.WithWorkerTags(workerTags...))
-	controllers, _, err := serverRepo.UpsertWorkerStatus(ctx, wConf, servers.WithUpdateTags(req.GetUpdateTags()))
-	if err != nil {
-		event.WriteError(ctx, op, err, event.WithInfoMsg("error storing worker status"))
-		return &pbs.StatusResponse{}, status.Errorf(codes.Internal, "Error storing worker status: %v", err)
-	}
-
-	responseControllers := []*servers.Server{}
-	for _, c := range controllers {
-		thisController := &servers.Server{
-			PrivateId:  c.PrivateId,
-			Address:    c.Address,
-			CreateTime: c.CreateTime,
-			UpdateTime: c.UpdateTime,
-		}
-		responseControllers = append(responseControllers, thisController)
-	}
-	ret := &pbs.StatusResponse{
-		Controllers: responseControllers,
-	}
-
-	stateReport := make([]session.StateReport, 0, len(req.GetJobs()))
-
-	for _, jobStatus := range req.GetJobs() {
-		switch jobStatus.Job.GetType() {
-		case pbs.JOBTYPE_JOBTYPE_SESSION:
-			si := jobStatus.GetJob().GetSessionInfo()
-			if si == nil {
-				return nil, status.Error(codes.Internal, "Error getting session info at status time")
-			}
-
-			switch si.Status {
-			case pbs.SESSIONSTATUS_SESSIONSTATUS_CANCELING,
-				pbs.SESSIONSTATUS_SESSIONSTATUS_TERMINATED:
-				// No need to see about canceling anything
-				continue
-			}
-
-			sr := session.StateReport{
-				SessionId:     si.GetSessionId(),
-				ConnectionIds: make([]string, 0, len(si.GetConnections())),
-			}
-			for _, conn := range si.GetConnections() {
-				switch conn.Status {
-				case pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_AUTHORIZED,
-					pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_CONNECTED:
-					sr.ConnectionIds = append(sr.ConnectionIds, conn.GetConnectionId())
-				}
-			}
-			stateReport = append(stateReport, sr)
-		}
-	}
-
-	notActive, err := session.WorkerStatusReport(ctx, sessRepo, connectionRepo, req.Worker.PrivateId, stateReport)
-	if err != nil {
-		return nil, status.Errorf(codes.Internal, "Error comparing state of sessions for worker: %s: %v", req.Worker.PrivateId, err)
-	}
-	for _, na := range notActive {
-		var connChanges []*pbs.Connection
-		for _, connId := range na.ConnectionIds {
-			connChanges = append(connChanges, &pbs.Connection{
-				ConnectionId: connId,
-				Status:       session.StatusClosed.ProtoVal(),
-			})
-		}
-		ret.JobsRequests = append(ret.JobsRequests, &pbs.JobChangeRequest{
-			Job: &pbs.Job{
-				Type: pbs.JOBTYPE_JOBTYPE_SESSION,
-				JobInfo: &pbs.Job_SessionInfo{
-					SessionInfo: &pbs.SessionJobInfo{
-						SessionId:   na.SessionId,
-						Status:      na.Status.ProtoVal(),
-						Connections: connChanges,
-					},
-				},
-			},
-			RequestType: pbs.CHANGETYPE_CHANGETYPE_UPDATE_STATE,
-		})
-	}
-
-	return ret, nil
-}
-
-func (ws *workerServiceServer) LookupSession(ctx context.Context, req *pbs.LookupSessionRequest) (*pbs.LookupSessionResponse, error) {
-	const op = "workers.(workerServiceServer).LookupSession"
-
-	sessRepo, err := ws.sessionRepoFn()
-	if err != nil {
-		return nil, status.Errorf(codes.Internal, "Error getting session repo: %v", err)
-	}
-
-	sessionInfo, authzSummary, err := sessRepo.LookupSession(ctx, req.GetSessionId())
-	if err != nil {
-		return nil, status.Errorf(codes.Internal, "Error looking up session: %v", err)
-	}
-	if sessionInfo == nil {
-		return nil, status.Error(codes.PermissionDenied, "Unknown session ID.")
-	}
-	if len(sessionInfo.States) == 0 {
-		return nil, status.Error(codes.Internal, "Empty session states during lookup.")
-	}
-
-	if sessionInfo.WorkerFilter != "" {
-		if req.ServerId == "" {
-			event.WriteError(ctx, op, errors.New("worker filter enabled for session but got no server ID from worker"))
-			return &pbs.LookupSessionResponse{}, status.Errorf(codes.Internal, "Did not receive server ID when looking up session but filtering is enabled: %v", err)
-		}
-		serversRepo, err := ws.serversRepoFn()
-		if err != nil {
-			event.WriteError(ctx, op, err, event.WithInfoMsg("error getting servers repo"))
-			return &pbs.LookupSessionResponse{}, status.Errorf(codes.Internal, "Error acquiring server repo when looking up session: %v", err)
-		}
-		workerTags, err := serversRepo.ListTagsForWorkers(ctx, []string{req.ServerId})
-		if err != nil {
-			event.WriteError(ctx, op, err, event.WithInfoMsg("error looking up tags for server", "server_id", req.ServerId))
-			return &pbs.LookupSessionResponse{}, status.Errorf(codes.Internal, "Error looking up tags for server: %v", err)
-		}
-		// Build the map for filtering.
-		tagMap := make(map[string][]string)
-		for _, tags := range workerTags {
-			for _, tag := range tags {
-				tagMap[tag.Key] = append(tagMap[tag.Key], tag.Value)
-				// We don't need to reinsert after the fact because maps are
-				// reference types, so we don't need to re-insert into tagMap
-			}
-		}
-
-		// Create the evaluator
-		eval, err := bexpr.CreateEvaluator(sessionInfo.WorkerFilter)
-		if err != nil {
-			event.WriteError(ctx, op, err, event.WithInfoMsg("error creating worker filter evaluator", "server_id", req.ServerId))
-			return &pbs.LookupSessionResponse{}, status.Errorf(codes.Internal, "Error creating worker filter evaluator: %v", err)
-		}
-		filterInput := map[string]interface{}{
-			"name": req.ServerId,
-			"tags": tagMap,
-		}
-		ok, err := eval.Evaluate(filterInput)
-		if err != nil {
-			return &pbs.LookupSessionResponse{}, status.Errorf(codes.Internal,
-				fmt.Sprintf("Worker filter expression evaluation resulted in error: %s", err))
-		}
-		if !ok {
-			return nil, handlers.ApiErrorWithCodeAndMessage(
-				codes.FailedPrecondition,
-				"Worker filter expression precludes this worker from serving this session")
-		}
-	}
-
-	creds, err := sessRepo.ListSessionCredentials(ctx, sessionInfo.ScopeId, sessionInfo.PublicId)
-	if err != nil {
-		return &pbs.LookupSessionResponse{}, status.Errorf(codes.Internal,
-			fmt.Sprintf("Error retrieving session credentials: %s", err))
-	}
-	var workerCreds []*pbs.Credential
-	for _, c := range creds {
-		m := &pbs.Credential{}
-		err = proto.Unmarshal(c, m)
-		if err != nil {
-			return &pbs.LookupSessionResponse{}, status.Errorf(codes.Internal,
-				fmt.Sprintf("Error unmarshaling credentials: %s", err))
-		}
-		workerCreds = append(workerCreds, m)
-	}
-
-	resp := &pbs.LookupSessionResponse{
-		Authorization: &targets.SessionAuthorizationData{
-			SessionId:   sessionInfo.GetPublicId(),
-			Certificate: sessionInfo.Certificate,
-		},
-		Status:          sessionInfo.States[0].Status.ProtoVal(),
-		Version:         sessionInfo.Version,
-		TofuToken:       string(sessionInfo.TofuToken),
-		Endpoint:        sessionInfo.Endpoint,
-		Expiration:      sessionInfo.ExpirationTime.Timestamp,
-		ConnectionLimit: sessionInfo.ConnectionLimit,
-		ConnectionsLeft: authzSummary.ConnectionLimit,
-		HostId:          sessionInfo.HostId,
-		HostSetId:       sessionInfo.HostSetId,
-		TargetId:        sessionInfo.TargetId,
-		UserId:          sessionInfo.UserId,
-		Credentials:     workerCreds,
-	}
-	if resp.ConnectionsLeft != -1 {
-		resp.ConnectionsLeft -= int32(authzSummary.CurrentConnectionCount)
-	}
-
-	wrapper, err := ws.kms.GetWrapper(ctx, sessionInfo.ScopeId, kms.KeyPurposeSessions, kms.WithKeyId(sessionInfo.KeyId))
-	if err != nil {
-		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.
-	_, resp.Authorization.PrivateKey, err = session.DeriveED25519Key(ctx, wrapper, sessionInfo.UserId, sessionInfo.GetPublicId())
-	if err != nil {
-		return nil, status.Errorf(codes.Internal, "Error deriving session key: %v", err)
-	}
-
-	return resp, nil
-}
-
-func (ws *workerServiceServer) CancelSession(ctx context.Context, req *pbs.CancelSessionRequest) (*pbs.CancelSessionResponse, error) {
-	const op = "workers.(workerServiceServer).CancelSession"
-
-	sessRepo, err := ws.sessionRepoFn()
-	if err != nil {
-		return nil, status.Errorf(codes.Internal, "error getting session repo: %v", err)
-	}
-
-	ses, _, err := sessRepo.LookupSession(ctx, req.GetSessionId())
-	if err != nil {
-		return nil, err
-	}
-
-	ses, err = sessRepo.CancelSession(ctx, req.GetSessionId(), ses.Version)
-	if err != nil {
-		return nil, err
-	}
-
-	return &pbs.CancelSessionResponse{
-		Status: ses.States[0].Status.ProtoVal(),
-	}, nil
-}
-
-func (ws *workerServiceServer) ActivateSession(ctx context.Context, req *pbs.ActivateSessionRequest) (*pbs.ActivateSessionResponse, error) {
-	const op = "workers.(workerServiceServer).ActivateSession"
-
-	sessRepo, err := ws.sessionRepoFn()
-	if err != nil {
-		return nil, status.Errorf(codes.Internal, "error getting session repo: %v", err)
-	}
-
-	sessionInfo, sessionStates, err := sessRepo.ActivateSession(ctx, req.GetSessionId(), req.GetVersion(), []byte(req.GetTofuToken()))
-	if err != nil {
-		return nil, status.Errorf(codes.Internal, "error looking up session: %v", err)
-	}
-	if sessionInfo == nil {
-		return nil, status.Error(codes.PermissionDenied, "Unknown session ID.")
-	}
-	if len(sessionStates) == 0 {
-		return nil, status.Error(codes.Internal, "Invalid session state in activate response.")
-	}
-
-	return &pbs.ActivateSessionResponse{
-		Status: sessionStates[0].Status.ProtoVal(),
-	}, nil
-}
-
-func (ws *workerServiceServer) AuthorizeConnection(ctx context.Context, req *pbs.AuthorizeConnectionRequest) (*pbs.AuthorizeConnectionResponse, error) {
-	const op = "workers.(workerServiceServer).AuthorizeConnection"
-	connectionRepo, err := ws.connectionRepoFn()
-	if err != nil {
-		return nil, status.Errorf(codes.Internal, "error getting session repo: %v", err)
-	}
-
-	sessionRepo, err := ws.sessionRepoFn()
-	if err != nil {
-		return nil, status.Errorf(codes.Internal, "error getting session repo: %v", err)
-	}
-
-	connectionInfo, connStates, authzSummary, err := session.AuthorizeConnection(ctx, sessionRepo, connectionRepo, req.GetSessionId(), req.GetWorkerId())
-	if err != nil {
-		return nil, err
-	}
-	if connectionInfo == nil {
-		return nil, status.Error(codes.Internal, "Invalid authorize connection response.")
-	}
-	if len(connStates) == 0 {
-		return nil, status.Error(codes.Internal, "Invalid connection state in authorize response.")
-	}
-
-	ret := &pbs.AuthorizeConnectionResponse{
-		ConnectionId:    connectionInfo.GetPublicId(),
-		Status:          connStates[0].Status.ProtoVal(),
-		ConnectionsLeft: authzSummary.ConnectionLimit,
-	}
-	if ret.ConnectionsLeft != -1 {
-		ret.ConnectionsLeft -= int32(authzSummary.CurrentConnectionCount)
-	}
-
-	return ret, nil
-}
-
-func (ws *workerServiceServer) ConnectConnection(ctx context.Context, req *pbs.ConnectConnectionRequest) (*pbs.ConnectConnectionResponse, error) {
-	const op = "workers.(workerServiceServer).ConnectConnection"
-	connRepo, err := ws.connectionRepoFn()
-	if err != nil {
-		return nil, status.Errorf(codes.Internal, "error getting session repo: %v", err)
-	}
-
-	connectionInfo, connStates, err := connRepo.ConnectConnection(ctx, session.ConnectWith{
-		ConnectionId:       req.GetConnectionId(),
-		ClientTcpAddress:   req.GetClientTcpAddress(),
-		ClientTcpPort:      req.GetClientTcpPort(),
-		EndpointTcpAddress: req.GetEndpointTcpAddress(),
-		EndpointTcpPort:    req.GetEndpointTcpPort(),
-		UserClientIp:       req.GetUserClientIp(),
-	})
-	if err != nil {
-		return nil, err
-	}
-	if connectionInfo == nil {
-		return nil, status.Error(codes.Internal, "Invalid connect connection response.")
-	}
-
-	return &pbs.ConnectConnectionResponse{
-		Status: connStates[0].Status.ProtoVal(),
-	}, nil
-}
-
-func (ws *workerServiceServer) CloseConnection(ctx context.Context, req *pbs.CloseConnectionRequest) (*pbs.CloseConnectionResponse, error) {
-	const op = "workers.(workerServiceServer).CloseConnection"
-	numCloses := len(req.GetCloseRequestData())
-	if numCloses == 0 {
-		return &pbs.CloseConnectionResponse{}, nil
-	}
-
-	closeWiths := make([]session.CloseWith, 0, numCloses)
-	closeIds := make([]string, 0, numCloses)
-
-	for _, v := range req.GetCloseRequestData() {
-		closeIds = append(closeIds, v.GetConnectionId())
-		closeWiths = append(closeWiths, session.CloseWith{
-			ConnectionId: v.GetConnectionId(),
-			BytesUp:      v.GetBytesUp(),
-			BytesDown:    v.GetBytesDown(),
-			ClosedReason: session.ClosedReason(v.GetReason()),
-		})
-	}
-	connRepo, err := ws.connectionRepoFn()
-	if err != nil {
-		return nil, status.Errorf(codes.Internal, "error getting connection repo: %v", err)
-	}
-
-	sessRepo, err := ws.sessionRepoFn()
-	if err != nil {
-		return nil, status.Errorf(codes.Internal, "error getting session repo: %v", err)
-	}
-
-	closeInfos, err := session.CloseConnections(ctx, sessRepo, connRepo, closeWiths)
-	if err != nil {
-		return nil, err
-	}
-	if closeInfos == nil {
-		return nil, status.Error(codes.Internal, "Invalid close connection response.")
-	}
-
-	closeData := make([]*pbs.CloseConnectionResponseData, 0, numCloses)
-	for _, v := range closeInfos {
-		if v.Connection == nil {
-			return nil, status.Errorf(codes.Internal, "No connection found while closing one of the connection IDs: %v", closeIds)
-		}
-		if len(v.ConnectionStates) == 0 {
-			return nil, status.Errorf(codes.Internal, "No connection states found while closing one of the connection IDs: %v", closeIds)
-		}
-		closeData = append(closeData, &pbs.CloseConnectionResponseData{
-			ConnectionId: v.Connection.GetPublicId(),
-			Status:       v.ConnectionStates[0].Status.ProtoVal(),
-		})
-	}
-
-	ret := &pbs.CloseConnectionResponse{
-		CloseResponseData: closeData,
-	}
-
-	return ret, nil
-}

internal/daemon/controller/handlers/workers/worker_service_status_test.go

@@ -1,517 +0,0 @@
-package workers_test
-
-import (
-	"context"
-	"sync"
-	"testing"
-
-	"github.com/google/go-cmp/cmp"
-	"github.com/google/go-cmp/cmp/cmpopts"
-	"github.com/hashicorp/boundary/internal/authtoken"
-	"github.com/hashicorp/boundary/internal/daemon/controller/handlers/workers"
-	"github.com/hashicorp/boundary/internal/db"
-	pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services"
-	"github.com/hashicorp/boundary/internal/host/static"
-	"github.com/hashicorp/boundary/internal/iam"
-	"github.com/hashicorp/boundary/internal/kms"
-	"github.com/hashicorp/boundary/internal/servers"
-	"github.com/hashicorp/boundary/internal/servers/store"
-	"github.com/hashicorp/boundary/internal/session"
-	"github.com/hashicorp/boundary/internal/target"
-	"github.com/hashicorp/boundary/internal/target/tcp"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-)
-
-func TestStatus(t *testing.T) {
-	ctx := context.Background()
-	conn, _ := db.TestSetup(t, "postgres")
-	rw := db.New(conn)
-	wrapper := db.TestWrapper(t)
-	kms := kms.TestKms(t, conn, wrapper)
-	org, prj := iam.TestScopes(t, iam.TestRepo(t, conn, wrapper))
-
-	serverRepo, _ := servers.NewRepository(rw, rw, kms)
-	serverRepo.UpsertController(ctx, &store.Controller{
-		PrivateId: "test_controller1",
-		Address:   "127.0.0.1",
-	})
-	serverRepo.UpsertWorkerStatus(ctx, servers.NewWorkerStatus(
-		"test_worker1",
-		servers.WithAddress("127.0.0.1")))
-
-	serversRepoFn := func() (*servers.Repository, error) {
-		return serverRepo, nil
-	}
-	sessionRepoFn := func() (*session.Repository, error) {
-		return session.NewRepository(rw, rw, kms)
-	}
-	connRepoFn := func() (*session.ConnectionRepository, error) {
-		return session.NewConnectionRepository(ctx, rw, rw, kms)
-	}
-
-	repo, err := sessionRepoFn()
-	require.NoError(t, err)
-	connRepo, err := connRepoFn()
-	require.NoError(t, err)
-
-	at := authtoken.TestAuthToken(t, conn, kms, org.GetPublicId())
-	uId := at.GetIamUserId()
-	hc := static.TestCatalogs(t, conn, prj.GetPublicId(), 1)[0]
-	hs := static.TestSets(t, conn, hc.GetPublicId(), 1)[0]
-	h := static.TestHosts(t, conn, hc.GetPublicId(), 1)[0]
-	static.TestSetMembers(t, conn, hs.GetPublicId(), []*static.Host{h})
-	tar := tcp.TestTarget(
-		ctx,
-		t, conn, prj.GetPublicId(), "test",
-		target.WithHostSources([]string{hs.GetPublicId()}),
-		target.WithSessionConnectionLimit(-1),
-	)
-
-	worker1 := servers.TestWorker(t, conn, wrapper)
-
-	sess := session.TestSession(t, conn, wrapper, session.ComposedOf{
-		UserId:          uId,
-		HostId:          h.GetPublicId(),
-		TargetId:        tar.GetPublicId(),
-		HostSetId:       hs.GetPublicId(),
-		AuthTokenId:     at.GetPublicId(),
-		ScopeId:         prj.GetPublicId(),
-		Endpoint:        "tcp://127.0.0.1:22",
-		ConnectionLimit: 10,
-	})
-	tofu := session.TestTofu(t)
-	sess, _, err = repo.ActivateSession(ctx, sess.PublicId, sess.Version, tofu)
-	require.NoError(t, err)
-	require.NoError(t, err)
-
-	s := workers.NewWorkerServiceServer(serversRepoFn, sessionRepoFn, connRepoFn, new(sync.Map), kms)
-	require.NotNil(t, s)
-
-	connection, _, err := connRepo.AuthorizeConnection(ctx, sess.PublicId, worker1.PublicId)
-	require.NoError(t, err)
-
-	cases := []struct {
-		name       string
-		wantErr    bool
-		wantErrMsg string
-		req        *pbs.StatusRequest
-		want       *pbs.StatusResponse
-	}{
-		{
-			name:    "No Sessions",
-			wantErr: false,
-			req: &pbs.StatusRequest{
-				Worker: &servers.Server{
-					PrivateId:  worker1.PublicId,
-					Address:    worker1.CanonicalAddress(),
-					CreateTime: worker1.CreateTime,
-					UpdateTime: worker1.UpdateTime,
-				},
-			},
-			want: &pbs.StatusResponse{
-				Controllers: []*servers.Server{
-					{
-						PrivateId: "test_controller1",
-						Address:   "127.0.0.1",
-					},
-				},
-			},
-		},
-		{
-			name:    "Still Active",
-			wantErr: false,
-			req: &pbs.StatusRequest{
-				Worker: &servers.Server{
-					PrivateId:  worker1.PublicId,
-					Address:    worker1.CanonicalAddress(),
-					CreateTime: worker1.CreateTime,
-					UpdateTime: worker1.UpdateTime,
-				},
-				Jobs: []*pbs.JobStatus{
-					{
-						Job: &pbs.Job{
-							Type: pbs.JOBTYPE_JOBTYPE_SESSION,
-							JobInfo: &pbs.Job_SessionInfo{
-								SessionInfo: &pbs.SessionJobInfo{
-									SessionId: sess.PublicId,
-									Status:    pbs.SESSIONSTATUS_SESSIONSTATUS_ACTIVE,
-									Connections: []*pbs.Connection{
-										{
-											ConnectionId: connection.PublicId,
-											Status:       pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_CONNECTED,
-										},
-									},
-								},
-							},
-						},
-					},
-				},
-			},
-			want: &pbs.StatusResponse{
-				Controllers: []*servers.Server{
-					{
-						PrivateId: "test_controller1",
-						Address:   "127.0.0.1",
-					},
-				},
-			},
-		},
-	}
-
-	for _, tc := range cases {
-		t.Run(tc.name, func(t *testing.T) {
-			assert, require := assert.New(t), require.New(t)
-
-			got, err := s.Status(ctx, tc.req)
-			if tc.wantErr {
-				require.Error(err)
-				assert.Nil(got)
-				assert.Equal(tc.wantErrMsg, err.Error())
-				return
-			}
-			assert.Empty(
-				cmp.Diff(
-					tc.want,
-					got,
-					cmpopts.IgnoreUnexported(
-						pbs.StatusResponse{},
-						servers.Server{},
-						pbs.JobChangeRequest{},
-						pbs.Job{},
-						pbs.Job_SessionInfo{},
-						pbs.SessionJobInfo{},
-						pbs.Connection{},
-					),
-					cmpopts.IgnoreFields(servers.Server{}, "CreateTime", "UpdateTime"),
-				),
-			)
-		})
-	}
-}
-
-func TestStatusSessionClosed(t *testing.T) {
-	ctx := context.Background()
-	conn, _ := db.TestSetup(t, "postgres")
-	rw := db.New(conn)
-	wrapper := db.TestWrapper(t)
-	kms := kms.TestKms(t, conn, wrapper)
-	org, prj := iam.TestScopes(t, iam.TestRepo(t, conn, wrapper))
-
-	serverRepo, _ := servers.NewRepository(rw, rw, kms)
-	serverRepo.UpsertController(ctx, &store.Controller{
-		PrivateId: "test_controller1",
-		Address:   "127.0.0.1",
-	})
-	serverRepo.UpsertWorkerStatus(ctx, servers.NewWorkerStatus(
-		"test_worker1",
-		servers.WithAddress("127.0.0.1")))
-
-	serversRepoFn := func() (*servers.Repository, error) {
-		return serverRepo, nil
-	}
-	sessionRepoFn := func() (*session.Repository, error) {
-		return session.NewRepository(rw, rw, kms)
-	}
-	connRepoFn := func() (*session.ConnectionRepository, error) {
-		return session.NewConnectionRepository(ctx, rw, rw, kms)
-	}
-
-	repo, err := sessionRepoFn()
-	require.NoError(t, err)
-	connRepo, err := connRepoFn()
-	require.NoError(t, err)
-
-	at := authtoken.TestAuthToken(t, conn, kms, org.GetPublicId())
-	uId := at.GetIamUserId()
-	hc := static.TestCatalogs(t, conn, prj.GetPublicId(), 1)[0]
-	hs := static.TestSets(t, conn, hc.GetPublicId(), 1)[0]
-	h := static.TestHosts(t, conn, hc.GetPublicId(), 1)[0]
-	static.TestSetMembers(t, conn, hs.GetPublicId(), []*static.Host{h})
-	tar := tcp.TestTarget(
-		ctx,
-		t, conn, prj.GetPublicId(), "test",
-		target.WithHostSources([]string{hs.GetPublicId()}),
-		target.WithSessionConnectionLimit(-1),
-	)
-
-	worker1 := servers.TestWorker(t, conn, wrapper)
-
-	sess := session.TestSession(t, conn, wrapper, session.ComposedOf{
-		UserId:          uId,
-		HostId:          h.GetPublicId(),
-		TargetId:        tar.GetPublicId(),
-		HostSetId:       hs.GetPublicId(),
-		AuthTokenId:     at.GetPublicId(),
-		ScopeId:         prj.GetPublicId(),
-		Endpoint:        "tcp://127.0.0.1:22",
-		ConnectionLimit: 10,
-	})
-	tofu := session.TestTofu(t)
-	sess, _, err = repo.ActivateSession(ctx, sess.PublicId, sess.Version, tofu)
-	require.NoError(t, err)
-	sess2 := session.TestSession(t, conn, wrapper, session.ComposedOf{
-		UserId:          uId,
-		HostId:          h.GetPublicId(),
-		TargetId:        tar.GetPublicId(),
-		HostSetId:       hs.GetPublicId(),
-		AuthTokenId:     at.GetPublicId(),
-		ScopeId:         prj.GetPublicId(),
-		Endpoint:        "tcp://127.0.0.1:22",
-		ConnectionLimit: 10,
-	})
-	tofu2 := session.TestTofu(t)
-	sess2, _, err = repo.ActivateSession(ctx, sess2.PublicId, sess2.Version, tofu2)
-	require.NoError(t, err)
-
-	s := workers.NewWorkerServiceServer(serversRepoFn, sessionRepoFn, connRepoFn, new(sync.Map), kms)
-	require.NotNil(t, s)
-
-	connection, _, err := connRepo.AuthorizeConnection(ctx, sess.PublicId, worker1.PublicId)
-	require.NoError(t, err)
-
-	cases := []struct {
-		name       string
-		wantErr    bool
-		wantErrMsg string
-		setupFn    func(t *testing.T)
-		req        *pbs.StatusRequest
-		want       *pbs.StatusResponse
-	}{
-		{
-			name:    "Connection Canceled",
-			wantErr: false,
-			setupFn: func(t *testing.T) {
-				_, err := repo.CancelSession(ctx, sess2.PublicId, sess.Version)
-				require.NoError(t, err)
-			},
-			req: &pbs.StatusRequest{
-				Worker: &servers.Server{
-					PrivateId:  worker1.PublicId,
-					Address:    worker1.CanonicalAddress(),
-					CreateTime: worker1.CreateTime,
-					UpdateTime: worker1.UpdateTime,
-				},
-				Jobs: []*pbs.JobStatus{
-					{
-						Job: &pbs.Job{
-							Type: pbs.JOBTYPE_JOBTYPE_SESSION,
-							JobInfo: &pbs.Job_SessionInfo{
-								SessionInfo: &pbs.SessionJobInfo{
-									SessionId: sess2.PublicId,
-									Status:    pbs.SESSIONSTATUS_SESSIONSTATUS_ACTIVE,
-									Connections: []*pbs.Connection{
-										{
-											ConnectionId: connection.PublicId,
-											Status:       pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_CONNECTED,
-										},
-									},
-								},
-							},
-						},
-					},
-				},
-			},
-			want: &pbs.StatusResponse{
-				Controllers: []*servers.Server{
-					{
-						PrivateId: "test_controller1",
-						Address:   "127.0.0.1",
-					},
-				},
-				JobsRequests: []*pbs.JobChangeRequest{
-					{
-						Job: &pbs.Job{
-							Type: pbs.JOBTYPE_JOBTYPE_SESSION,
-							JobInfo: &pbs.Job_SessionInfo{
-								SessionInfo: &pbs.SessionJobInfo{
-									SessionId: sess2.PublicId,
-									Status:    pbs.SESSIONSTATUS_SESSIONSTATUS_CANCELING,
-								},
-							},
-						},
-						RequestType: pbs.CHANGETYPE_CHANGETYPE_UPDATE_STATE,
-					},
-				},
-			},
-		},
-	}
-
-	for _, tc := range cases {
-		t.Run(tc.name, func(t *testing.T) {
-			assert, require := assert.New(t), require.New(t)
-
-			if tc.setupFn != nil {
-				tc.setupFn(t)
-			}
-			got, err := s.Status(ctx, tc.req)
-			if tc.wantErr {
-				require.Error(err)
-				assert.Nil(got)
-				assert.Equal(tc.wantErrMsg, err.Error())
-				return
-			}
-			assert.Empty(
-				cmp.Diff(
-					tc.want,
-					got,
-					cmpopts.IgnoreUnexported(
-						pbs.StatusResponse{},
-						servers.Server{},
-						pbs.JobChangeRequest{},
-						pbs.Job{},
-						pbs.Job_SessionInfo{},
-						pbs.SessionJobInfo{},
-						pbs.Connection{},
-					),
-					cmpopts.IgnoreFields(servers.Server{}, "CreateTime", "UpdateTime"),
-				),
-			)
-		})
-	}
-}
-
-func TestStatusDeadConnection(t *testing.T) {
-	ctx := context.Background()
-	conn, _ := db.TestSetup(t, "postgres")
-	rw := db.New(conn)
-	wrapper := db.TestWrapper(t)
-	kms := kms.TestKms(t, conn, wrapper)
-	org, prj := iam.TestScopes(t, iam.TestRepo(t, conn, wrapper))
-
-	serverRepo, _ := servers.NewRepository(rw, rw, kms)
-	serverRepo.UpsertController(ctx, &store.Controller{
-		PrivateId: "test_controller1",
-		Address:   "127.0.0.1",
-	})
-	serverRepo.UpsertWorkerStatus(ctx, servers.NewWorkerStatus(
-		"test_worker1",
-		servers.WithAddress("127.0.0.1")))
-
-	serversRepoFn := func() (*servers.Repository, error) {
-		return serverRepo, nil
-	}
-	sessionRepoFn := func() (*session.Repository, error) {
-		return session.NewRepository(rw, rw, kms)
-	}
-	connRepoFn := func() (*session.ConnectionRepository, error) {
-		return session.NewConnectionRepository(ctx, rw, rw, kms, session.WithWorkerStateDelay(0))
-	}
-
-	repo, err := sessionRepoFn()
-	require.NoError(t, err)
-	connRepo, err := connRepoFn()
-	require.NoError(t, err)
-
-	at := authtoken.TestAuthToken(t, conn, kms, org.GetPublicId())
-	uId := at.GetIamUserId()
-	hc := static.TestCatalogs(t, conn, prj.GetPublicId(), 1)[0]
-	hs := static.TestSets(t, conn, hc.GetPublicId(), 1)[0]
-	h := static.TestHosts(t, conn, hc.GetPublicId(), 1)[0]
-	static.TestSetMembers(t, conn, hs.GetPublicId(), []*static.Host{h})
-	tar := tcp.TestTarget(
-		ctx,
-		t, conn, prj.GetPublicId(), "test",
-		target.WithHostSources([]string{hs.GetPublicId()}),
-		target.WithSessionConnectionLimit(-1),
-	)
-
-	worker1 := servers.TestWorker(t, conn, wrapper)
-
-	sess := session.TestSession(t, conn, wrapper, session.ComposedOf{
-		UserId:          uId,
-		HostId:          h.GetPublicId(),
-		TargetId:        tar.GetPublicId(),
-		HostSetId:       hs.GetPublicId(),
-		AuthTokenId:     at.GetPublicId(),
-		ScopeId:         prj.GetPublicId(),
-		Endpoint:        "tcp://127.0.0.1:22",
-		ConnectionLimit: 10,
-	})
-	tofu := session.TestTofu(t)
-	sess, _, err = repo.ActivateSession(ctx, sess.PublicId, sess.Version, tofu)
-	require.NoError(t, err)
-	sess2 := session.TestSession(t, conn, wrapper, session.ComposedOf{
-		UserId:          uId,
-		HostId:          h.GetPublicId(),
-		TargetId:        tar.GetPublicId(),
-		HostSetId:       hs.GetPublicId(),
-		AuthTokenId:     at.GetPublicId(),
-		ScopeId:         prj.GetPublicId(),
-		Endpoint:        "tcp://127.0.0.1:22",
-		ConnectionLimit: 10,
-	})
-	tofu2 := session.TestTofu(t)
-	sess2, _, err = repo.ActivateSession(ctx, sess2.PublicId, sess2.Version, tofu2)
-	require.NoError(t, err)
-
-	s := workers.NewWorkerServiceServer(serversRepoFn, sessionRepoFn, connRepoFn, new(sync.Map), kms)
-	require.NotNil(t, s)
-
-	connection, _, err := connRepo.AuthorizeConnection(ctx, sess.PublicId, worker1.PublicId)
-	require.NoError(t, err)
-	deadConn, _, err := connRepo.AuthorizeConnection(ctx, sess2.PublicId, worker1.PublicId)
-	require.NoError(t, err)
-	require.NotEqual(t, deadConn.PublicId, connection.PublicId)
-
-	req := &pbs.StatusRequest{
-		Worker: &servers.Server{
-			PrivateId:  worker1.PublicId,
-			Address:    worker1.CanonicalAddress(),
-			CreateTime: worker1.CreateTime,
-			UpdateTime: worker1.UpdateTime,
-		},
-		Jobs: []*pbs.JobStatus{
-			{
-				Job: &pbs.Job{
-					Type: pbs.JOBTYPE_JOBTYPE_SESSION,
-					JobInfo: &pbs.Job_SessionInfo{
-						SessionInfo: &pbs.SessionJobInfo{
-							SessionId: sess.PublicId,
-							Status:    pbs.SESSIONSTATUS_SESSIONSTATUS_ACTIVE,
-							Connections: []*pbs.Connection{
-								{
-									ConnectionId: connection.PublicId,
-									Status:       pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_CONNECTED,
-								},
-							},
-						},
-					},
-				},
-			},
-		},
-	}
-	want := &pbs.StatusResponse{
-		Controllers: []*servers.Server{
-			{
-				PrivateId: "test_controller1",
-				Address:   "127.0.0.1",
-			},
-		},
-	}
-
-	got, err := s.Status(ctx, req)
-	assert.Empty(t,
-		cmp.Diff(
-			want,
-			got,
-			cmpopts.IgnoreUnexported(
-				pbs.StatusResponse{},
-				servers.Server{},
-				pbs.JobChangeRequest{},
-				pbs.Job{},
-				pbs.Job_SessionInfo{},
-				pbs.SessionJobInfo{},
-				pbs.Connection{},
-			),
-			cmpopts.IgnoreFields(servers.Server{}, "CreateTime", "UpdateTime"),
-		),
-	)
-
-	gotConn, states, err := connRepo.LookupConnection(ctx, deadConn.PublicId)
-	require.NoError(t, err)
-	assert.Equal(t, session.ConnectionSystemError, session.ClosedReason(gotConn.ClosedReason))
-	assert.Equal(t, 2, len(states))
-	assert.Nil(t, states[0].EndTime)
-	assert.Equal(t, session.StatusClosed, states[0].Status)
-}

internal/daemon/controller/handlers/workers/worker_service_test.go

@@ -1,176 +0,0 @@
-package workers_test
-
-import (
-	"context"
-	"sync"
-	"testing"
-
-	"github.com/google/go-cmp/cmp"
-	"github.com/google/go-cmp/cmp/cmpopts"
-	"github.com/hashicorp/boundary/internal/authtoken"
-	"github.com/hashicorp/boundary/internal/daemon/controller/handlers/workers"
-	"github.com/hashicorp/boundary/internal/db"
-	pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services"
-	"github.com/hashicorp/boundary/internal/host/static"
-	"github.com/hashicorp/boundary/internal/iam"
-	"github.com/hashicorp/boundary/internal/kms"
-	"github.com/hashicorp/boundary/internal/servers"
-	"github.com/hashicorp/boundary/internal/session"
-	"github.com/hashicorp/boundary/internal/target"
-	"github.com/hashicorp/boundary/internal/target/tcp"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-	"google.golang.org/protobuf/proto"
-)
-
-func TestLookupSession(t *testing.T) {
-	ctx := context.Background()
-	conn, _ := db.TestSetup(t, "postgres")
-	rw := db.New(conn)
-	wrapper := db.TestWrapper(t)
-	kms := kms.TestKms(t, conn, wrapper)
-	org, prj := iam.TestScopes(t, iam.TestRepo(t, conn, wrapper))
-
-	serversRepoFn := func() (*servers.Repository, error) {
-		return servers.NewRepository(rw, rw, kms)
-	}
-	sessionRepoFn := func() (*session.Repository, error) {
-		return session.NewRepository(rw, rw, kms)
-	}
-	connectionRepoFn := func() (*session.ConnectionRepository, error) {
-		return session.NewConnectionRepository(ctx, rw, rw, kms)
-	}
-
-	at := authtoken.TestAuthToken(t, conn, kms, org.GetPublicId())
-	uId := at.GetIamUserId()
-	hc := static.TestCatalogs(t, conn, prj.GetPublicId(), 1)[0]
-	hs := static.TestSets(t, conn, hc.GetPublicId(), 1)[0]
-	h := static.TestHosts(t, conn, hc.GetPublicId(), 1)[0]
-	static.TestSetMembers(t, conn, hs.GetPublicId(), []*static.Host{h})
-	tar := tcp.TestTarget(ctx, t, conn, prj.GetPublicId(), "test", target.WithHostSources([]string{hs.GetPublicId()}))
-
-	sess := session.TestSession(t, conn, wrapper, session.ComposedOf{
-		UserId:      uId,
-		HostId:      h.GetPublicId(),
-		TargetId:    tar.GetPublicId(),
-		HostSetId:   hs.GetPublicId(),
-		AuthTokenId: at.GetPublicId(),
-		ScopeId:     prj.GetPublicId(),
-		Endpoint:    "tcp://127.0.0.1:22",
-	})
-
-	egressSess := session.TestSession(t, conn, wrapper, session.ComposedOf{
-		UserId:      uId,
-		HostId:      h.GetPublicId(),
-		TargetId:    tar.GetPublicId(),
-		HostSetId:   hs.GetPublicId(),
-		AuthTokenId: at.GetPublicId(),
-		ScopeId:     prj.GetPublicId(),
-		Endpoint:    "tcp://127.0.0.1:22",
-	})
-
-	repo, err := sessionRepoFn()
-	require.NoError(t, err)
-
-	creds := []*pbs.Credential{
-		{
-			Credential: &pbs.Credential_UserPassword{
-				UserPassword: &pbs.UserPassword{
-					Username: "username",
-					Password: "password",
-				},
-			},
-		},
-		{
-			Credential: &pbs.Credential_UserPassword{
-				UserPassword: &pbs.UserPassword{
-					Username: "another-username",
-					Password: "a different password",
-				},
-			},
-		},
-	}
-
-	workerCreds := make([]session.Credential, 0, len(creds))
-	for _, c := range creds {
-		data, err := proto.Marshal(c)
-		require.NoError(t, err)
-		workerCreds = append(workerCreds, data)
-	}
-	err = repo.AddSessionCredentials(ctx, egressSess.ScopeId, egressSess.GetPublicId(), workerCreds)
-	require.NoError(t, err)
-
-	s := workers.NewWorkerServiceServer(serversRepoFn, sessionRepoFn, connectionRepoFn, new(sync.Map), kms)
-	require.NotNil(t, s)
-
-	cases := []struct {
-		name       string
-		wantErr    bool
-		wantErrMsg string
-		want       *pbs.LookupSessionResponse
-		sessionId  string
-	}{
-		{
-			name:       "Invalid session id",
-			sessionId:  "s_fakesession",
-			wantErr:    true,
-			wantErrMsg: "rpc error: code = PermissionDenied desc = Unknown session ID.",
-		},
-		{
-			name:      "Valid",
-			sessionId: sess.PublicId,
-			want: &pbs.LookupSessionResponse{
-				ConnectionLimit: 1,
-				ConnectionsLeft: 1,
-				Version:         1,
-				Endpoint:        sess.Endpoint,
-				HostId:          sess.HostId,
-				HostSetId:       sess.HostSetId,
-				TargetId:        sess.TargetId,
-				UserId:          sess.UserId,
-				Status:          pbs.SESSIONSTATUS_SESSIONSTATUS_PENDING,
-			},
-		},
-		{
-			name:      "Valid-with-egress-creds",
-			sessionId: egressSess.PublicId,
-			want: &pbs.LookupSessionResponse{
-				ConnectionLimit: 1,
-				ConnectionsLeft: 1,
-				Version:         1,
-				Endpoint:        egressSess.Endpoint,
-				HostId:          egressSess.HostId,
-				HostSetId:       egressSess.HostSetId,
-				TargetId:        egressSess.TargetId,
-				UserId:          egressSess.UserId,
-				Status:          pbs.SESSIONSTATUS_SESSIONSTATUS_PENDING,
-				Credentials:     creds,
-			},
-		},
-	}
-
-	for _, tc := range cases {
-		t.Run(tc.name, func(t *testing.T) {
-			assert, require := assert.New(t), require.New(t)
-			req := &pbs.LookupSessionRequest{
-				SessionId: tc.sessionId,
-			}
-
-			got, err := s.LookupSession(ctx, req)
-			if tc.wantErr {
-				require.Error(err)
-				assert.Nil(got)
-				assert.Equal(tc.wantErrMsg, err.Error())
-				return
-			}
-			assert.Empty(
-				cmp.Diff(
-					tc.want,
-					got,
-					cmpopts.IgnoreUnexported(pbs.LookupSessionResponse{}, pbs.Credential{}, pbs.UserPassword{}),
-					cmpopts.IgnoreFields(pbs.LookupSessionResponse{}, "Expiration", "Authorization"),
-				),
-			)
-		})
-	}
-}

internal/daemon/controller/listeners.go

@@ -13,7 +13,7 @@ import (
 
 	grpc_middleware "github.com/grpc-ecosystem/go-grpc-middleware"
 	"github.com/hashicorp/boundary/internal/cmd/base"
-	"github.com/hashicorp/boundary/internal/daemon/controller/handlers/workers"
+	"github.com/hashicorp/boundary/internal/daemon/cluster/handlers"
 	"github.com/hashicorp/boundary/internal/daemon/controller/internal/metric"
 	pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services"
 	"github.com/hashicorp/go-multierror"
@@ -140,12 +140,12 @@ func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error
 		),
 	)
 
-	workerService := workers.NewWorkerServiceServer(c.ServersRepoFn, c.SessionRepoFn, c.ConnectionRepoFn,
+	workerService := handlers.NewWorkerServiceServer(c.ServersRepoFn, c.SessionRepoFn, c.ConnectionRepoFn,
 		c.workerStatusUpdateTimes, c.kms)
 	pbs.RegisterServerCoordinationServiceServer(workerServer, workerService)
 	pbs.RegisterSessionServiceServer(workerServer, workerService)
 
-	multihopService := workers.NewMultihopServiceServer(
+	multihopService := handlers.NewMultihopServiceServer(
 		nodeauth.MakeCurrentParametersFactory(c.baseContext, nodee.NopTransactionStorage(c.NodeeFileStorage)),
 	)
 	multihop.RegisterMultihopServiceServer(workerServer, multihopService)