Add fields to server_worker to be a boundary resource (#2083)

* Remove the server_id from session table, change it to worker_id on session_connection

This is step 1 in updating the worker table.  The worker is becoming a public resource so it must have
a public id.  We drop the server_id column on the session_connection table and add the worker_id column
so we can use the wt_public_key type.  Since db migrations are applied while the controller is offline,
it should be safe to assume all session connections are not in a running state, thus allowing us to null
out all the worker ids in the session_connection table.

For now we don't have very good enforcement of the wt_public_id contraints, since the id is currently
defined in the worker config, but this will soon change and the worker's id will be generated by the
controller.

* Updating server_id references to controller_id.

* Adding worker tag db writing test.

* Move the server table back to the top.

* Update comment for the worker object.

* Update comment for the worker table.

* Update the newWorkerId call to take a context and fix its error.

* Correct comment for the wt_network_address domain.

internal/cmd/commands/server/worker_tags_reload_test.go

@@ -100,11 +100,11 @@ func TestServer_ReloadWorkerTags(t *testing.T) {
 		require.NoError(err)
 		tags, err := serversRepo.ListTagsForWorkers(testController.Context(), []string{name})
 		require.NoError(err)
-		require.Len(tags, 2)
-		require.Equal(key, tags[0].Key)
-		require.Equal(values[0], tags[0].Value)
-		require.Equal(key, tags[1].Key)
-		require.Equal(values[1], tags[1].Value)
+		require.Len(tags[name], 2)
+		require.Equal(key, tags[name][0].Key)
+		require.Equal(values[0], tags[name][0].Value)
+		require.Equal(key, tags[name][1].Key)
+		require.Equal(values[1], tags[name][1].Value)
 	}
 
 	// Give time to populate up to the controller

internal/daemon/controller/handlers/targets/target_service.go

@@ -911,7 +911,7 @@ func (s Service) AuthorizeSession(ctx context.Context, req *pbs.AuthorizeSession
 	if hasWorkerFilter && len(workerIds) > 0 {
 		finalWorkers := make([]*pb.WorkerInfo, 0, len(workers))
 		// Fetch the tags for the given worker IDs
-		tags, err := serversRepo.ListTagsForWorkers(ctx, workerIds)
+		workerTags, err := serversRepo.ListTagsForWorkers(ctx, workerIds)
 		if err != nil {
 			return nil, err
 		}
@@ -919,13 +919,15 @@ func (s Service) AuthorizeSession(ctx context.Context, req *pbs.AuthorizeSession
 		// built from the worker config, but with one extra level: a map of the
 		// worker's ID to its filter map.
 		tagMap := make(map[string]map[string][]string)
-		for _, tag := range tags {
-			currWorkerMap := tagMap[tag.WorkerId]
+		for wId, tags := range workerTags {
+			currWorkerMap := tagMap[wId]
 			if currWorkerMap == nil {
 				currWorkerMap = make(map[string][]string)
-				tagMap[tag.WorkerId] = currWorkerMap
+				tagMap[wId] = currWorkerMap
+			}
+			for _, tag := range tags {
+				currWorkerMap[tag.Key] = append(currWorkerMap[tag.Key], tag.Value)
 			}
-			currWorkerMap[tag.Key] = append(currWorkerMap[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
 		}

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

@@ -7,8 +7,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/hashicorp/boundary/internal/servers/store"
-
 	"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"
@@ -16,6 +14,7 @@ import (
 	"github.com/hashicorp/boundary/internal/observability/event"
 	"github.com/hashicorp/boundary/internal/servers"
 	"github.com/hashicorp/boundary/internal/session"
+	"github.com/hashicorp/boundary/internal/types/scope"
 	"github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/targets"
 	"github.com/hashicorp/go-bexpr"
 	"google.golang.org/grpc/codes"
@@ -74,20 +73,20 @@ func (ws *workerServiceServer) Status(ctx context.Context, req *pbs.StatusReques
 
 	reqServer := req.GetWorker()
 	// Convert API tags to storage tags
-	workerTags := make(map[string]*store.TagValues)
+	workerTags := make([]*servers.Tag, 0, len(reqServer.Tags))
 	for k, v := range reqServer.Tags {
-		thisTag := &store.TagValues{
-			Values: v.GetValues(),
+		for _, val := range v.GetValues() {
+			workerTags = append(workerTags, &servers.Tag{
+				Key:   k,
+				Value: val,
+			})
 		}
-		workerTags[k] = thisTag
-	}
-	worker := &store.Worker{
-		PublicId:   reqServer.PrivateId,
-		Address:    reqServer.Address,
-		CreateTime: reqServer.CreateTime,
-		UpdateTime: reqServer.UpdateTime,
-		Tags:       workerTags,
 	}
+
+	worker := servers.NewWorker(scope.Global.String(),
+		servers.WithPublicId(reqServer.PrivateId),
+		servers.WithAddress(reqServer.Address),
+		servers.WithWorkerTags(workerTags...))
 	controllers, _, err := serverRepo.UpsertWorker(ctx, worker, servers.WithUpdateTags(req.GetUpdateTags()))
 	if err != nil {
 		event.WriteError(ctx, op, err, event.WithInfoMsg("error storing worker status"))
@@ -199,17 +198,19 @@ func (ws *workerServiceServer) LookupSession(ctx context.Context, req *pbs.Looku
 			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)
 		}
-		tags, err := serversRepo.ListTagsForWorkers(ctx, []string{req.ServerId})
+		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 _, 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
+		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

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

@@ -6,6 +6,7 @@ import (
 	"testing"
 
 	"github.com/hashicorp/boundary/internal/servers/store"
+	"github.com/hashicorp/boundary/internal/types/scope"
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
@@ -37,10 +38,10 @@ func TestStatus(t *testing.T) {
 		PrivateId: "test_controller1",
 		Address:   "127.0.0.1",
 	})
-	serverRepo.UpsertWorker(ctx, &store.Worker{
-		PublicId: "test_worker1",
-		Address:  "127.0.0.1",
-	})
+	serverRepo.UpsertWorker(ctx, servers.NewWorker(
+		scope.Global.String(),
+		servers.WithPublicId("test_worker1"),
+		servers.WithAddress("127.0.0.1")))
 
 	serversRepoFn := func() (*servers.Repository, error) {
 		return serverRepo, nil
@@ -205,10 +206,10 @@ func TestStatusSessionClosed(t *testing.T) {
 		PrivateId: "test_controller1",
 		Address:   "127.0.0.1",
 	})
-	serverRepo.UpsertWorker(ctx, &store.Worker{
-		PublicId: "test_worker1",
-		Address:  "127.0.0.1",
-	})
+	serverRepo.UpsertWorker(ctx, servers.NewWorker(
+		scope.Global.String(),
+		servers.WithPublicId("test_worker1"),
+		servers.WithAddress("127.0.0.1")))
 
 	serversRepoFn := func() (*servers.Repository, error) {
 		return serverRepo, nil
@@ -387,10 +388,10 @@ func TestStatusDeadConnection(t *testing.T) {
 		PrivateId: "test_controller1",
 		Address:   "127.0.0.1",
 	})
-	serverRepo.UpsertWorker(ctx, &store.Worker{
-		PublicId: "test_worker1",
-		Address:  "127.0.0.1",
-	})
+	serverRepo.UpsertWorker(ctx, servers.NewWorker(
+		scope.Global.String(),
+		servers.WithPublicId("test_worker1"),
+		servers.WithAddress("127.0.0.1")))
 
 	serversRepoFn := func() (*servers.Repository, error) {
 		return serverRepo, nil

internal/db/schema/migrations/oss/postgres/99/01_domain_types.up.sql

@@ -0,0 +1,15 @@
+begin;
+
+-- wt_network_address defines a type for a network address.
+-- While wt_dns_names does exist, a wt_network_address can be either a dns name
+-- or an ip address.
+create domain wt_network_address as text
+  constraint wt_network_address_too_short
+    check (length(trim(value)) > 0)
+  constraint wt_network_address_too_long
+    check (length(trim(value)) < 256);
+comment on domain wt_network_address is
+  'standard column for a network address.';
+
+
+commit;

internal/db/schema/migrations/oss/postgres/99/02_worker_controller_tables.up.sql

@@ -1,12 +1,11 @@
 begin;
 
 -- Split the server table into two new tables: controller and worker
+
 create table server_controller (
   private_id text primary key,
   description wt_description,
-  address text not null
-    constraint address_must_not_be_empty
-      check(length(trim(address)) > 0),
+  address wt_network_address not null,
   create_time wt_timestamp,
   update_time wt_timestamp
 );
@@ -30,19 +29,25 @@ create trigger controller_update_time_column before update on server_controller
 -- workers will now be exposed as resources in boundary.
 create table server_worker (
   public_id wt_public_id primary key,
+  scope_id wt_scope_id not null
+    references iam_scope_global(scope_id)
+      on delete cascade
+      on update cascade,
   description wt_description,
   name wt_name unique,
-  address text not null
-    constraint address_must_not_be_empty
-      check(length(trim(address)) > 0),
+  -- The address can be null since it is an optional value from the API.
+  address wt_network_address,
   create_time wt_timestamp,
-  update_time wt_timestamp
+  update_time wt_timestamp,
+  version wt_version,
+  constraint server_worker_scope_id_name_uq
+    unique(scope_id, name)
 );
 comment on table server_worker  is
   'server_worker is a table where each row represents a Boundary worker.';
 
 create trigger immutable_columns before update on server_worker
-  for each row execute procedure immutable_columns('public_id','create_time');
+  for each row execute procedure immutable_columns('public_id', 'scope_id', 'create_time');
 
 create trigger default_create_time_column before insert on server_worker
   for each row execute procedure default_create_time();
@@ -96,15 +101,39 @@ alter table session_connection
       on update cascade;
 
 -- Update job run table so that server id references controller id
+-- We are not migrating the values from server_id to controller_id. The fkey
+-- constraint says that server_id can be set to null when the server is deleted
+-- which is what this migration does (removing all records from the server table).
+-- Not migrating values make it easier to change types in the server_worker and
+-- server_controller tables (like from text to wt_public_id or text to wt_address)
+-- without having to worry about old values being valid in the new types.
+-- Finally, neither jobs nor servers are exposed out of boundary so the risk of
+-- losing data that would be useful later on is diminished.
+alter table job_run
+  add column controller_id wt_private_id,
+  drop column server_id;
 alter table job_run
-  drop constraint server_fkey,
   add constraint server_controller_fkey
-    foreign key (server_id)
+    foreign key (controller_id)
       references server_controller (private_id)
       on delete set null
       on update cascade;
 
--- Replaces the view created in 9/01 to remove references to the server/worker id.
+-- Since the above alter tables sets all controller_ids to null running jobs
+-- can no longer be reclaimed by any controller and should be considered
+-- interrupted.
+update job_run
+set
+  status = 'interrupted',
+  end_time = current_timestamp
+where
+    status = 'running';
+
+
+-- Replaces the view created in 9/01.
+-- Remove the worker id from this view.  In actuality this is almost a no-op
+-- because no server information was ever getting populated here due to a bug
+-- in the update mask when updating a session at the time we activate a session.
 create view session_list as
   select
     s.public_id, s.user_id, s.host_id, s.target_id,

internal/proto/controller/storage/job/store/v1/job.proto

@@ -26,6 +26,12 @@ message Job {
 }
 
 message JobRun {
+  // server_id referenced an server table which was replaced with both a
+  // server_controller table and a server_worker table.  For the purpose of
+  // jobs, only the controller table is relevant thus we replaced this column
+  // with controller_id.
+  reserved "server_id";
+
   // private_id is used to access the job run but not intended to be available via the API
   // @inject_tag: `gorm:"primary_key;default:null"`
   string private_id = 1;
@@ -38,10 +44,6 @@ message JobRun {
   // @inject_tag: `gorm:"not_null"`
   string job_name = 3;
 
-  // The server_id of the server running the job and must be set.
-  // @inject_tag: `gorm:"not_null"`
-  string server_id = 4;
-
   // The create_time is set by the database.
   // @inject_tag: `gorm:"default:current_timestamp"`
   timestamp.v1.Timestamp create_time = 5;
@@ -65,4 +67,8 @@ message JobRun {
   // status of the job run (running, completed, failed or interrupted).
   // @inject_tag: `gorm:"not_null"`
   string status = 10;
+
+  // The controller_id of the controller running the job and must be set.
+  // @inject_tag: `gorm:"not_null"`
+  string controller_id = 11;
 }

internal/proto/controller/storage/servers/store/v1/worker.proto

@@ -8,12 +8,11 @@ option go_package = "github.com/hashicorp/boundary/internal/servers/store;store"
 
 // Worker contains all fields related to a Worker resource
 message Worker {
-  // Public ID of the resource
+  // public_id is a surrogate key suitable for use in a public API
   // @inject_tag: `gorm:"primary_key"`
   string public_id = 10;
 
   // Name of the resource (optional)
-  // Description of the resource (optional)
   // @inject_tag: `gorm:"default:null"`
   string name = 20;
 
@@ -21,21 +20,39 @@ message Worker {
   // @inject_tag: `gorm:"default:null"`
   string description = 30;
 
-  // Address for the worker
+  // Address for the worker. This must be set
+  // @inject_tag: `gorm:"default:null"`
   string address = 40;
 
-  // First seen time from the RDBMS
-  storage.timestamp.v1.Timestamp create_time = 50;
+  // The create_time is set by the database.
+  // @inject_tag: `gorm:"default:current_timestamp"`
+  timestamp.v1.Timestamp create_time = 50;
 
-  // Last time there was an update
-  storage.timestamp.v1.Timestamp update_time = 60;
+  // The update_time is set by the database.
+  // @inject_tag: `gorm:"default:current_timestamp"`
+  timestamp.v1.Timestamp update_time = 60;
 
-  // @gotags: `gorm:"-"`
-  map<string, TagValues> tags = 70;
+  // The scope_id of the owning scope and must be set.
+  // @inject_tag: `gorm:"not_null"`
+  string scope_id = 70;
+
+  // version allows optimistic locking of the resource.
+  // @inject_tag: `gorm:"default:null"`
+  uint32 version = 80;
 }
 
-// TagValues is used because map fields cannot be repeated but can be a
-// message
-message TagValues {
-  repeated string values = 10;
+// WorkerTag is a tag for a worker.  The primary key is comprised of the
+// worker_id, key, value, and source.
+message WorkerTag {
+  // worker_id is the public key that key of the worker this tag is for.
+  // @inject_tag: `gorm:"default:null"`
+  string worker_id = 10;
+
+  // key is the key of the tag. This must be set.
+  // @inject_tag: `gorm:"default:null"`
+  string key = 20;
+
+  // value is the value
+  // @inject_tag: `gorm:"default:null"`
+  string value = 30;
 }

internal/scheduler/job/options.go

@@ -27,7 +27,7 @@ type options struct {
 	withRunJobsLimit uint
 	withLimit        int
 	withName         string
-	withServerId     string
+	withControllerId string
 }
 
 func getDefaultOptions() options {
@@ -72,9 +72,9 @@ func WithName(n string) Option {
 	}
 }
 
-// WithServerId provides an option to provide the server id to match when calling InterruptRuns
-func WithServerId(id string) Option {
+// WithControllerId provides an option to provide the server id to match when calling InterruptRuns
+func WithControllerId(id string) Option {
 	return func(o *options) {
-		o.withServerId = id
+		o.withControllerId = id
 	}
 }

internal/scheduler/job/options_test.go

@@ -49,12 +49,12 @@ func Test_GetOpts(t *testing.T) {
 		testOpts.withName = "name"
 		assert.Equal(opts, testOpts)
 	})
-	t.Run("WithServerId", func(t *testing.T) {
+	t.Run("WithControllerId", func(t *testing.T) {
 		assert := assert.New(t)
-		opts := getOpts(WithServerId("server_id"))
+		opts := getOpts(WithControllerId("controller_id"))
 		testOpts := getDefaultOptions()
 		assert.NotEqual(opts, testOpts)
-		testOpts.withServerId = "server_id"
+		testOpts.withControllerId = "controller_id"
 		assert.Equal(opts, testOpts)
 	})
 }

internal/scheduler/job/query.go

@@ -2,7 +2,7 @@ package job
 
 const runJobsQuery = `
 	insert into job_run (
-	  job_plugin_id, job_name, server_id
+	  job_plugin_id, job_name, controller_id
 	)
 	select 
 	  job_plugin_id, job_name, ?

internal/scheduler/job/repository_run.go

@@ -255,7 +255,7 @@ func (r *Repository) FailRun(ctx context.Context, runId string, completed, total
 // Once a run has been persisted with a final run status (completed, failed
 // or interrupted), any future calls to InterruptRuns will return an error with Code
 // errors.InvalidJobRunState.
-// WithServerId is the only valid option
+// WithControllerId is the only valid option
 func (r *Repository) InterruptRuns(ctx context.Context, interruptThreshold time.Duration, opt ...Option) ([]*Run, error) {
 	const op = "job.(Repository).InterruptRuns"
 
@@ -263,12 +263,12 @@ func (r *Repository) InterruptRuns(ctx context.Context, interruptThreshold time.
 
 	// interruptThreshold is seconds in past so * -1
 	args := []interface{}{-1 * int(interruptThreshold.Round(time.Second).Seconds())}
-	var whereServerId string
-	if opts.withServerId != "" {
-		whereServerId = "and server_id = ?"
-		args = append(args, opts.withServerId)
+	var whereControllerId string
+	if opts.withControllerId != "" {
+		whereControllerId = "and controller_id = ?"
+		args = append(args, opts.withControllerId)
 	}
-	query := fmt.Sprintf(interruptRunsQuery, whereServerId)
+	query := fmt.Sprintf(interruptRunsQuery, whereControllerId)
 
 	var runs []*Run
 	_, err := r.writer.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{},

internal/scheduler/job/repository_run_test.go

@@ -27,30 +27,30 @@ func TestRepository_RunJobs(t *testing.T) {
 	server := testController(t, conn, wrapper)
 
 	tests := []struct {
-		name        string
-		serverId    string
-		job         *Job
-		wantRun     bool
-		wantErr     bool
-		wantErrCode errors.Code
-		wantErrMsg  string
+		name         string
+		ControllerId string
+		job          *Job
+		wantRun      bool
+		wantErr      bool
+		wantErrCode  errors.Code
+		wantErrMsg   string
 	}{
 		{
-			name:        "missing-server-id",
-			serverId:    "",
-			wantErr:     true,
-			wantErrCode: errors.InvalidParameter,
-			wantErrMsg:  "job.(Repository).RunJobs: missing server id: parameter violation: error #100",
+			name:         "missing-server-id",
+			ControllerId: "",
+			wantErr:      true,
+			wantErrCode:  errors.InvalidParameter,
+			wantErrMsg:   "job.(Repository).RunJobs: missing server id: parameter violation: error #100",
 		},
 		{
-			name:     "no-work",
-			serverId: server.PrivateId,
-			wantRun:  false,
-			wantErr:  false,
+			name:         "no-work",
+			ControllerId: server.PrivateId,
+			wantRun:      false,
+			wantErr:      false,
 		},
 		{
-			name:     "valid",
-			serverId: server.PrivateId,
+			name:         "valid",
+			ControllerId: server.PrivateId,
 			job: &Job{
 				Job: &store.Job{
 					Name:        "valid-test",
@@ -60,8 +60,8 @@ func TestRepository_RunJobs(t *testing.T) {
 			wantRun: true,
 		},
 		{
-			name:     "fake-server-id",
-			serverId: "fake-server-id",
+			name:         "fake-server-id",
+			ControllerId: "fake-server-id",
 			job: &Job{
 				Job: &store.Job{
 					Name:        "fake-server-id-test",
@@ -84,7 +84,7 @@ func TestRepository_RunJobs(t *testing.T) {
 			assert.NoError(err)
 			require.NotNil(repo)
 
-			got, err := repo.RunJobs(context.Background(), tt.serverId)
+			got, err := repo.RunJobs(context.Background(), tt.ControllerId)
 			if tt.wantErr {
 				require.Error(err)
 				assert.Truef(errors.Match(errors.T(tt.wantErrCode), err), "Unexpected error %s", err)
@@ -253,10 +253,10 @@ func TestRepository_UpdateProgress(t *testing.T) {
 			name: "status-already-interrupted",
 			orig: &Run{
 				JobRun: &store.JobRun{
-					JobName:     job.Name,
-					JobPluginId: job.PluginId,
-					ServerId:    server.PrivateId,
-					Status:      Interrupted.string(),
+					JobName:      job.Name,
+					JobPluginId:  job.PluginId,
+					ControllerId: server.PrivateId,
+					Status:       Interrupted.string(),
 				},
 			},
 			wantErr:     true,
@@ -267,10 +267,10 @@ func TestRepository_UpdateProgress(t *testing.T) {
 			name: "status-already-failed",
 			orig: &Run{
 				JobRun: &store.JobRun{
-					JobName:     job.Name,
-					JobPluginId: job.PluginId,
-					ServerId:    server.PrivateId,
-					Status:      Failed.string(),
+					JobName:      job.Name,
+					JobPluginId:  job.PluginId,
+					ControllerId: server.PrivateId,
+					Status:       Failed.string(),
 				},
 			},
 			wantErr:     true,
@@ -281,10 +281,10 @@ func TestRepository_UpdateProgress(t *testing.T) {
 			name: "status-already-completed",
 			orig: &Run{
 				JobRun: &store.JobRun{
-					JobName:     job.Name,
-					JobPluginId: job.PluginId,
-					ServerId:    server.PrivateId,
-					Status:      Completed.string(),
+					JobName:      job.Name,
+					JobPluginId:  job.PluginId,
+					ControllerId: server.PrivateId,
+					Status:       Completed.string(),
 				},
 			},
 			wantErr:     true,
@@ -295,10 +295,10 @@ func TestRepository_UpdateProgress(t *testing.T) {
 			name: "valid-no-changes",
 			orig: &Run{
 				JobRun: &store.JobRun{
-					JobName:     job.Name,
-					JobPluginId: job.PluginId,
-					ServerId:    server.PrivateId,
-					Status:      Running.string(),
+					JobName:      job.Name,
+					JobPluginId:  job.PluginId,
+					ControllerId: server.PrivateId,
+					Status:       Running.string(),
 				},
 			},
 		},
@@ -306,10 +306,10 @@ func TestRepository_UpdateProgress(t *testing.T) {
 			name: "valid-update-total",
 			orig: &Run{
 				JobRun: &store.JobRun{
-					JobName:     job.Name,
-					JobPluginId: job.PluginId,
-					ServerId:    server.PrivateId,
-					Status:      Running.string(),
+					JobName:      job.Name,
+					JobPluginId:  job.PluginId,
+					ControllerId: server.PrivateId,
+					Status:       Running.string(),
 				},
 			},
 			args: args{
@@ -323,11 +323,11 @@ func TestRepository_UpdateProgress(t *testing.T) {
 			name: "valid-update-completed",
 			orig: &Run{
 				JobRun: &store.JobRun{
-					JobName:     job.Name,
-					JobPluginId: job.PluginId,
-					ServerId:    server.PrivateId,
-					Status:      Running.string(),
-					TotalCount:  10,
+					JobName:      job.Name,
+					JobPluginId:  job.PluginId,
+					ControllerId: server.PrivateId,
+					Status:       Running.string(),
+					TotalCount:   10,
 				},
 			},
 			args: args{
@@ -343,10 +343,10 @@ func TestRepository_UpdateProgress(t *testing.T) {
 			name: "valid-update-completed-and-total",
 			orig: &Run{
 				JobRun: &store.JobRun{
-					JobName:     job.Name,
-					JobPluginId: job.PluginId,
-					ServerId:    server.PrivateId,
-					Status:      Running.string(),
+					JobName:      job.Name,
+					JobPluginId:  job.PluginId,
+					ControllerId: server.PrivateId,
+					Status:       Running.string(),
 				},
 			},
 			args: args{
@@ -362,10 +362,10 @@ func TestRepository_UpdateProgress(t *testing.T) {
 			name: "invalid-completed-greater-than-total",
 			orig: &Run{
 				JobRun: &store.JobRun{
-					JobName:     job.Name,
-					JobPluginId: job.PluginId,
-					ServerId:    server.PrivateId,
-					Status:      Running.string(),
+					JobName:      job.Name,
+					JobPluginId:  job.PluginId,
+					ControllerId: server.PrivateId,
+					Status:       Running.string(),
 				},
 			},
 			args: args{
@@ -467,10 +467,10 @@ func TestRepository_CompleteRun(t *testing.T) {
 			name: "status-already-interrupted",
 			orig: &Run{
 				JobRun: &store.JobRun{
-					JobName:     job.Name,
-					JobPluginId: job.PluginId,
-					ServerId:    server.PrivateId,
-					Status:      Interrupted.string(),
+					JobName:      job.Name,
+					JobPluginId:  job.PluginId,
+					ControllerId: server.PrivateId,
+					Status:       Interrupted.string(),
 				},
 			},
 			wantErr:     true,
@@ -481,10 +481,10 @@ func TestRepository_CompleteRun(t *testing.T) {
 			name: "status-already-failed",
 			orig: &Run{
 				JobRun: &store.JobRun{
-					JobName:     job.Name,
-					JobPluginId: job.PluginId,
-					ServerId:    server.PrivateId,
-					Status:      Failed.string(),
+					JobName:      job.Name,
+					JobPluginId:  job.PluginId,
+					ControllerId: server.PrivateId,
+					Status:       Failed.string(),
 				},
 			},
 			wantErr:     true,
@@ -495,10 +495,10 @@ func TestRepository_CompleteRun(t *testing.T) {
 			name: "status-already-completed",
 			orig: &Run{
 				JobRun: &store.JobRun{
-					JobName:     job.Name,
-					JobPluginId: job.PluginId,
-					ServerId:    server.PrivateId,
-					Status:      Completed.string(),
+					JobName:      job.Name,
+					JobPluginId:  job.PluginId,
+					ControllerId: server.PrivateId,
+					Status:       Completed.string(),
 				},
 			},
 			wantErr:     true,
@@ -509,10 +509,10 @@ func TestRepository_CompleteRun(t *testing.T) {
 			name: "valid",
 			orig: &Run{
 				JobRun: &store.JobRun{
-					JobName:     job.Name,
-					JobPluginId: job.PluginId,
-					ServerId:    server.PrivateId,
-					Status:      Running.string(),
+					JobName:      job.Name,
+					JobPluginId:  job.PluginId,
+					ControllerId: server.PrivateId,
+					Status:       Running.string(),
 				},
 			},
 			nextRunIn: time.Hour,
@@ -521,10 +521,10 @@ func TestRepository_CompleteRun(t *testing.T) {
 			name: "valid-with-progress",
 			orig: &Run{
 				JobRun: &store.JobRun{
-					JobName:     job.Name,
-					JobPluginId: job.PluginId,
-					ServerId:    server.PrivateId,
-					Status:      Running.string(),
+					JobName:      job.Name,
+					JobPluginId:  job.PluginId,
+					ControllerId: server.PrivateId,
+					Status:       Running.string(),
 				},
 			},
 			args: args{completed: 10, total: 20},
@@ -631,10 +631,10 @@ func TestRepository_FailRun(t *testing.T) {
 			name: "status-already-interrupted",
 			orig: &Run{
 				JobRun: &store.JobRun{
-					JobName:     job.Name,
-					JobPluginId: job.PluginId,
-					ServerId:    server.PrivateId,
-					Status:      Interrupted.string(),
+					JobName:      job.Name,
+					JobPluginId:  job.PluginId,
+					ControllerId: server.PrivateId,
+					Status:       Interrupted.string(),
 				},
 			},
 			wantErr:     true,
@@ -645,10 +645,10 @@ func TestRepository_FailRun(t *testing.T) {
 			name: "status-already-failed",
 			orig: &Run{
 				JobRun: &store.JobRun{
-					JobName:     job.Name,
-					JobPluginId: job.PluginId,
-					ServerId:    server.PrivateId,
-					Status:      Failed.string(),
+					JobName:      job.Name,
+					JobPluginId:  job.PluginId,
+					ControllerId: server.PrivateId,
+					Status:       Failed.string(),
 				},
 			},
 			wantErr:     true,
@@ -659,10 +659,10 @@ func TestRepository_FailRun(t *testing.T) {
 			name: "status-already-completed",
 			orig: &Run{
 				JobRun: &store.JobRun{
-					JobName:     job.Name,
-					JobPluginId: job.PluginId,
-					ServerId:    server.PrivateId,
-					Status:      Completed.string(),
+					JobName:      job.Name,
+					JobPluginId:  job.PluginId,
+					ControllerId: server.PrivateId,
+					Status:       Completed.string(),
 				},
 			},
 			wantErr:     true,
@@ -673,10 +673,10 @@ func TestRepository_FailRun(t *testing.T) {
 			name: "valid",
 			orig: &Run{
 				JobRun: &store.JobRun{
-					JobName:     job.Name,
-					JobPluginId: job.PluginId,
-					ServerId:    server.PrivateId,
-					Status:      Running.string(),
+					JobName:      job.Name,
+					JobPluginId:  job.PluginId,
+					ControllerId: server.PrivateId,
+					Status:       Running.string(),
 				},
 			},
 		},
@@ -684,10 +684,10 @@ func TestRepository_FailRun(t *testing.T) {
 			name: "valid-with-progress",
 			orig: &Run{
 				JobRun: &store.JobRun{
-					JobName:     job.Name,
-					JobPluginId: job.PluginId,
-					ServerId:    server.PrivateId,
-					Status:      Running.string(),
+					JobName:      job.Name,
+					JobPluginId:  job.PluginId,
+					ControllerId: server.PrivateId,
+					Status:       Running.string(),
 				},
 			},
 			args: args{completed: 10, total: 20},
@@ -859,7 +859,7 @@ func TestRepository_InterruptServerRuns(t *testing.T) {
 	job3 := testJob(t, conn, "job3", "description", wrapper)
 
 	type args struct {
-		serverId     string
+		ControllerId string
 		opts         []Option
 		expectedJobs []*Job
 	}
@@ -872,7 +872,7 @@ func TestRepository_InterruptServerRuns(t *testing.T) {
 			name: "all-runs",
 			runs: []args{
 				{
-					serverId:     server1.PrivateId,
+					ControllerId: server1.PrivateId,
 					opts:         []Option{WithRunJobsLimit(3)},
 					expectedJobs: []*Job{job1, job2, job3},
 				},
@@ -887,22 +887,22 @@ func TestRepository_InterruptServerRuns(t *testing.T) {
 			name: "all-runs-on-single-server-with-server-id",
 			runs: []args{
 				{
-					serverId:     server2.PrivateId,
+					ControllerId: server2.PrivateId,
 					opts:         []Option{WithRunJobsLimit(3)},
 					expectedJobs: []*Job{job1, job2, job3},
 				},
 			},
 			interrupts: []args{
 				{
-					opts:         []Option{WithServerId(server1.PrivateId)},
+					opts:         []Option{WithControllerId(server1.PrivateId)},
 					expectedJobs: []*Job{},
 				},
 				{
-					opts:         []Option{WithServerId(server2.PrivateId)},
+					opts:         []Option{WithControllerId(server2.PrivateId)},
 					expectedJobs: []*Job{job1, job2, job3},
 				},
 				{
-					opts:         []Option{WithServerId(server3.PrivateId)},
+					opts:         []Option{WithControllerId(server3.PrivateId)},
 					expectedJobs: []*Job{},
 				},
 			},
@@ -921,15 +921,15 @@ func TestRepository_InterruptServerRuns(t *testing.T) {
 			runs: []args{},
 			interrupts: []args{
 				{
-					opts:         []Option{WithServerId(server1.PrivateId)},
+					opts:         []Option{WithControllerId(server1.PrivateId)},
 					expectedJobs: []*Job{},
 				},
 				{
-					opts:         []Option{WithServerId(server2.PrivateId)},
+					opts:         []Option{WithControllerId(server2.PrivateId)},
 					expectedJobs: []*Job{},
 				},
 				{
-					opts:         []Option{WithServerId(server3.PrivateId)},
+					opts:         []Option{WithControllerId(server3.PrivateId)},
 					expectedJobs: []*Job{},
 				},
 			},
@@ -938,17 +938,17 @@ func TestRepository_InterruptServerRuns(t *testing.T) {
 			name: "multiple-servers-interrupt-all",
 			runs: []args{
 				{
-					serverId:     server1.PrivateId,
+					ControllerId: server1.PrivateId,
 					opts:         []Option{WithRunJobsLimit(1)},
 					expectedJobs: []*Job{job1},
 				},
 				{
-					serverId:     server2.PrivateId,
+					ControllerId: server2.PrivateId,
 					opts:         []Option{WithRunJobsLimit(1)},
 					expectedJobs: []*Job{job2},
 				},
 				{
-					serverId:     server3.PrivateId,
+					ControllerId: server3.PrivateId,
 					opts:         []Option{WithRunJobsLimit(1)},
 					expectedJobs: []*Job{job3},
 				},
@@ -963,32 +963,32 @@ func TestRepository_InterruptServerRuns(t *testing.T) {
 			name: "multiple-servers-with-server-id",
 			runs: []args{
 				{
-					serverId:     server1.PrivateId,
+					ControllerId: server1.PrivateId,
 					opts:         []Option{WithRunJobsLimit(1)},
 					expectedJobs: []*Job{job1},
 				},
 				{
-					serverId:     server2.PrivateId,
+					ControllerId: server2.PrivateId,
 					opts:         []Option{WithRunJobsLimit(1)},
 					expectedJobs: []*Job{job2},
 				},
 				{
-					serverId:     server3.PrivateId,
+					ControllerId: server3.PrivateId,
 					opts:         []Option{WithRunJobsLimit(1)},
 					expectedJobs: []*Job{job3},
 				},
 			},
 			interrupts: []args{
 				{
-					opts:         []Option{WithServerId(server1.PrivateId)},
+					opts:         []Option{WithControllerId(server1.PrivateId)},
 					expectedJobs: []*Job{job1},
 				},
 				{
-					opts:         []Option{WithServerId(server2.PrivateId)},
+					opts:         []Option{WithControllerId(server2.PrivateId)},
 					expectedJobs: []*Job{job2},
 				},
 				{
-					opts:         []Option{WithServerId(server3.PrivateId)},
+					opts:         []Option{WithControllerId(server3.PrivateId)},
 					expectedJobs: []*Job{job3},
 				},
 			},
@@ -997,32 +997,32 @@ func TestRepository_InterruptServerRuns(t *testing.T) {
 			name: "multiple-servers-distributed-runs",
 			runs: []args{
 				{
-					serverId:     server1.PrivateId,
+					ControllerId: server1.PrivateId,
 					opts:         []Option{WithRunJobsLimit(2)},
 					expectedJobs: []*Job{job1, job2},
 				},
 				{
-					serverId:     server2.PrivateId,
+					ControllerId: server2.PrivateId,
 					opts:         []Option{WithRunJobsLimit(1)},
 					expectedJobs: []*Job{job3},
 				},
 				{
-					serverId:     server3.PrivateId,
+					ControllerId: server3.PrivateId,
 					opts:         []Option{WithRunJobsLimit(1)},
 					expectedJobs: []*Job{},
 				},
 			},
 			interrupts: []args{
 				{
-					opts:         []Option{WithServerId(server1.PrivateId)},
+					opts:         []Option{WithControllerId(server1.PrivateId)},
 					expectedJobs: []*Job{job1, job2},
 				},
 				{
-					opts:         []Option{WithServerId(server2.PrivateId)},
+					opts:         []Option{WithControllerId(server2.PrivateId)},
 					expectedJobs: []*Job{job3},
 				},
 				{
-					opts:         []Option{WithServerId(server3.PrivateId)},
+					opts:         []Option{WithControllerId(server3.PrivateId)},
 					expectedJobs: []*Job{},
 				},
 			},
@@ -1031,17 +1031,17 @@ func TestRepository_InterruptServerRuns(t *testing.T) {
 			name: "multiple-servers-distributed-runs-interrupt-all",
 			runs: []args{
 				{
-					serverId:     server1.PrivateId,
+					ControllerId: server1.PrivateId,
 					opts:         []Option{WithRunJobsLimit(2)},
 					expectedJobs: []*Job{job1, job2},
 				},
 				{
-					serverId:     server2.PrivateId,
+					ControllerId: server2.PrivateId,
 					opts:         []Option{WithRunJobsLimit(1)},
 					expectedJobs: []*Job{job3},
 				},
 				{
-					serverId:     server3.PrivateId,
+					ControllerId: server3.PrivateId,
 					opts:         []Option{WithRunJobsLimit(1)},
 					expectedJobs: []*Job{},
 				},
@@ -1060,7 +1060,7 @@ func TestRepository_InterruptServerRuns(t *testing.T) {
 			require.NoError(err)
 
 			for _, r := range tt.runs {
-				runs, err := repo.RunJobs(context.Background(), r.serverId, r.opts...)
+				runs, err := repo.RunJobs(context.Background(), r.ControllerId, r.opts...)
 				require.NoError(err)
 				assert.Len(runs, len(r.expectedJobs))
 				sort.Slice(runs, func(i, j int) bool { return runs[i].JobName < runs[j].JobName })

internal/scheduler/job/store/job.pb.go

@@ -116,9 +116,6 @@ type JobRun struct {
 	// The job_name is the name of the job and must be set.
 	// @inject_tag: `gorm:"not_null"`
 	JobName string `protobuf:"bytes,3,opt,name=job_name,json=jobName,proto3" json:"job_name,omitempty" gorm:"not_null"`
-	// The server_id of the server running the job and must be set.
-	// @inject_tag: `gorm:"not_null"`
-	ServerId string `protobuf:"bytes,4,opt,name=server_id,json=serverId,proto3" json:"server_id,omitempty" gorm:"not_null"`
 	// The create_time is set by the database.
 	// @inject_tag: `gorm:"default:current_timestamp"`
 	CreateTime *timestamp.Timestamp `protobuf:"bytes,5,opt,name=create_time,json=createTime,proto3" json:"create_time,omitempty" gorm:"default:current_timestamp"`
@@ -137,6 +134,9 @@ type JobRun struct {
 	// status of the job run (running, completed, failed or interrupted).
 	// @inject_tag: `gorm:"not_null"`
 	Status string `protobuf:"bytes,10,opt,name=status,proto3" json:"status,omitempty" gorm:"not_null"`
+	// The controller_id of the controller running the job and must be set.
+	// @inject_tag: `gorm:"not_null"`
+	ControllerId string `protobuf:"bytes,11,opt,name=controller_id,json=controllerId,proto3" json:"controller_id,omitempty" gorm:"not_null"`
 }
 
 func (x *JobRun) Reset() {
@@ -192,13 +192,6 @@ func (x *JobRun) GetJobName() string {
 	return ""
 }
 
-func (x *JobRun) GetServerId() string {
-	if x != nil {
-		return x.ServerId
-	}
-	return ""
-}
-
 func (x *JobRun) GetCreateTime() *timestamp.Timestamp {
 	if x != nil {
 		return x.CreateTime
@@ -241,6 +234,13 @@ func (x *JobRun) GetStatus() string {
 	return ""
 }
 
+func (x *JobRun) GetControllerId() string {
+	if x != nil {
+		return x.ControllerId
+	}
+	return ""
+}
+
 var File_controller_storage_job_store_v1_job_proto protoreflect.FileDescriptor
 
 var file_controller_storage_job_store_v1_job_proto_rawDesc = []byte{
@@ -263,40 +263,41 @@ var file_controller_storage_job_store_v1_job_proto_rawDesc = []byte{
 	0x72, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74,
 	0x61, 0x6d, 0x70, 0x2e, 0x76, 0x31, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70,
 	0x52, 0x10, 0x6e, 0x65, 0x78, 0x74, 0x53, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x64, 0x52,
-	0x75, 0x6e, 0x22, 0xc6, 0x03, 0x0a, 0x06, 0x4a, 0x6f, 0x62, 0x52, 0x75, 0x6e, 0x12, 0x1d, 0x0a,
+	0x75, 0x6e, 0x22, 0xd9, 0x03, 0x0a, 0x06, 0x4a, 0x6f, 0x62, 0x52, 0x75, 0x6e, 0x12, 0x1d, 0x0a,
 	0x0a, 0x70, 0x72, 0x69, 0x76, 0x61, 0x74, 0x65, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28,
 	0x09, 0x52, 0x09, 0x70, 0x72, 0x69, 0x76, 0x61, 0x74, 0x65, 0x49, 0x64, 0x12, 0x22, 0x0a, 0x0d,
 	0x6a, 0x6f, 0x62, 0x5f, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x5f, 0x69, 0x64, 0x18, 0x02, 0x20,
 	0x01, 0x28, 0x09, 0x52, 0x0b, 0x6a, 0x6f, 0x62, 0x50, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x49, 0x64,
 	0x12, 0x19, 0x0a, 0x08, 0x6a, 0x6f, 0x62, 0x5f, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x03, 0x20, 0x01,
-	0x28, 0x09, 0x52, 0x07, 0x6a, 0x6f, 0x62, 0x4e, 0x61, 0x6d, 0x65, 0x12, 0x1b, 0x0a, 0x09, 0x73,
-	0x65, 0x72, 0x76, 0x65, 0x72, 0x5f, 0x69, 0x64, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08,
-	0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x49, 0x64, 0x12, 0x4b, 0x0a, 0x0b, 0x63, 0x72, 0x65, 0x61,
-	0x74, 0x65, 0x5f, 0x74, 0x69, 0x6d, 0x65, 0x18, 0x05, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x2a, 0x2e,
+	0x28, 0x09, 0x52, 0x07, 0x6a, 0x6f, 0x62, 0x4e, 0x61, 0x6d, 0x65, 0x12, 0x4b, 0x0a, 0x0b, 0x63,
+	0x72, 0x65, 0x61, 0x74, 0x65, 0x5f, 0x74, 0x69, 0x6d, 0x65, 0x18, 0x05, 0x20, 0x01, 0x28, 0x0b,
+	0x32, 0x2a, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x74,
+	0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e,
+	0x76, 0x31, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x0a, 0x63, 0x72,
+	0x65, 0x61, 0x74, 0x65, 0x54, 0x69, 0x6d, 0x65, 0x12, 0x4b, 0x0a, 0x0b, 0x75, 0x70, 0x64, 0x61,
+	0x74, 0x65, 0x5f, 0x74, 0x69, 0x6d, 0x65, 0x18, 0x06, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x2a, 0x2e,
 	0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61,
 	0x67, 0x65, 0x2e, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e, 0x76, 0x31, 0x2e,
-	0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x0a, 0x63, 0x72, 0x65, 0x61, 0x74,
-	0x65, 0x54, 0x69, 0x6d, 0x65, 0x12, 0x4b, 0x0a, 0x0b, 0x75, 0x70, 0x64, 0x61, 0x74, 0x65, 0x5f,
-	0x74, 0x69, 0x6d, 0x65, 0x18, 0x06, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x2a, 0x2e, 0x63, 0x6f, 0x6e,
-	0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e,
-	0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e, 0x76, 0x31, 0x2e, 0x54, 0x69, 0x6d,
-	0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x0a, 0x75, 0x70, 0x64, 0x61, 0x74, 0x65, 0x54, 0x69,
-	0x6d, 0x65, 0x12, 0x45, 0x0a, 0x08, 0x65, 0x6e, 0x64, 0x5f, 0x74, 0x69, 0x6d, 0x65, 0x18, 0x07,
-	0x20, 0x01, 0x28, 0x0b, 0x32, 0x2a, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65,
-	0x72, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74,
-	0x61, 0x6d, 0x70, 0x2e, 0x76, 0x31, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70,
-	0x52, 0x07, 0x65, 0x6e, 0x64, 0x54, 0x69, 0x6d, 0x65, 0x12, 0x27, 0x0a, 0x0f, 0x63, 0x6f, 0x6d,
-	0x70, 0x6c, 0x65, 0x74, 0x65, 0x64, 0x5f, 0x63, 0x6f, 0x75, 0x6e, 0x74, 0x18, 0x08, 0x20, 0x01,
-	0x28, 0x0d, 0x52, 0x0e, 0x63, 0x6f, 0x6d, 0x70, 0x6c, 0x65, 0x74, 0x65, 0x64, 0x43, 0x6f, 0x75,
-	0x6e, 0x74, 0x12, 0x1f, 0x0a, 0x0b, 0x74, 0x6f, 0x74, 0x61, 0x6c, 0x5f, 0x63, 0x6f, 0x75, 0x6e,
-	0x74, 0x18, 0x09, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0a, 0x74, 0x6f, 0x74, 0x61, 0x6c, 0x43, 0x6f,
-	0x75, 0x6e, 0x74, 0x12, 0x16, 0x0a, 0x06, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x18, 0x0a, 0x20,
-	0x01, 0x28, 0x09, 0x52, 0x06, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x42, 0x42, 0x5a, 0x40, 0x67,
-	0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63,
-	0x6f, 0x72, 0x70, 0x2f, 0x62, 0x6f, 0x75, 0x6e, 0x64, 0x61, 0x72, 0x79, 0x2f, 0x69, 0x6e, 0x74,
-	0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x73, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x2f,
-	0x6a, 0x6f, 0x62, 0x2f, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x3b, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x62,
-	0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+	0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x0a, 0x75, 0x70, 0x64, 0x61, 0x74,
+	0x65, 0x54, 0x69, 0x6d, 0x65, 0x12, 0x45, 0x0a, 0x08, 0x65, 0x6e, 0x64, 0x5f, 0x74, 0x69, 0x6d,
+	0x65, 0x18, 0x07, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x2a, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f,
+	0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, 0x74, 0x69, 0x6d,
+	0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e, 0x76, 0x31, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74,
+	0x61, 0x6d, 0x70, 0x52, 0x07, 0x65, 0x6e, 0x64, 0x54, 0x69, 0x6d, 0x65, 0x12, 0x27, 0x0a, 0x0f,
+	0x63, 0x6f, 0x6d, 0x70, 0x6c, 0x65, 0x74, 0x65, 0x64, 0x5f, 0x63, 0x6f, 0x75, 0x6e, 0x74, 0x18,
+	0x08, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0e, 0x63, 0x6f, 0x6d, 0x70, 0x6c, 0x65, 0x74, 0x65, 0x64,
+	0x43, 0x6f, 0x75, 0x6e, 0x74, 0x12, 0x1f, 0x0a, 0x0b, 0x74, 0x6f, 0x74, 0x61, 0x6c, 0x5f, 0x63,
+	0x6f, 0x75, 0x6e, 0x74, 0x18, 0x09, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0a, 0x74, 0x6f, 0x74, 0x61,
+	0x6c, 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x12, 0x16, 0x0a, 0x06, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73,
+	0x18, 0x0a, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x12, 0x23,
+	0x0a, 0x0d, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x5f, 0x69, 0x64, 0x18,
+	0x0b, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0c, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65,
+	0x72, 0x49, 0x64, 0x52, 0x09, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x5f, 0x69, 0x64, 0x42, 0x42,
+	0x5a, 0x40, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x68, 0x61, 0x73,
+	0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2f, 0x62, 0x6f, 0x75, 0x6e, 0x64, 0x61, 0x72, 0x79, 0x2f,
+	0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x73, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c,
+	0x65, 0x72, 0x2f, 0x6a, 0x6f, 0x62, 0x2f, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x3b, 0x73, 0x74, 0x6f,
+	0x72, 0x65, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (

internal/scheduler/job/testing.go

@@ -35,7 +35,7 @@ func testJob(t testing.TB, conn *db.DB, name, description string, wrapper wrappi
 func testRun(conn *db.DB, pluginId, name, cId string) (*Run, error) {
 	query := `
 		insert into job_run (
-			job_plugin_id, job_name, server_id
+			job_plugin_id, job_name, controller_id
 		)
 		values (?,?,?)
 		on conflict (job_plugin_id, job_name) where status = 'running'
@@ -65,7 +65,7 @@ func testRun(conn *db.DB, pluginId, name, cId string) (*Run, error) {
 func testRunWithUpdateTime(conn *db.DB, pluginId, name, cId string, updateTime time.Time) (*Run, error) {
 	query := `
 		insert into job_run (
-		  job_plugin_id, job_name, server_id, update_time
+		  job_plugin_id, job_name, controller_id, update_time
 		)
 		values (?,?,?,?)
 		returning *;

internal/scheduler/job/testing_test.go

@@ -68,7 +68,7 @@ func Test_TestRun(t *testing.T) {
 	var got Run
 	err = rw.LookupWhere(context.Background(), &got, "private_id = ?", []interface{}{run.PrivateId})
 	require.NoError(err)
-	assert.Equal(server.PrivateId, got.ServerId)
+	assert.Equal(server.PrivateId, got.ControllerId)
 	assert.Equal(job.Name, got.JobName)
 	assert.Equal(job.PluginId, got.JobPluginId)
 	assert.NotEmpty(got.CreateTime)

internal/scheduler/scheduler.go

@@ -154,7 +154,7 @@ func (s *Scheduler) Start(ctx context.Context, wg *sync.WaitGroup) error {
 	}
 
 	// Interrupt all runs previously allocated to this server
-	_, err = repo.InterruptRuns(ctx, 0, job.WithServerId(s.serverId))
+	_, err = repo.InterruptRuns(ctx, 0, job.WithControllerId(s.serverId))
 	if err != nil {
 		return errors.Wrap(ctx, err, op)
 	}

internal/servers/gorm.go

@@ -1,6 +0,0 @@
-package servers
-
-// TableName overrides the table name used by WorkerTag to `worker_tag`
-func (WorkerTag) TableName() string {
-	return "server_worker_tag"
-}

internal/servers/options.go

@@ -1,6 +1,8 @@
 package servers
 
-import "time"
+import (
+	"time"
+)
 
 // getOpts - iterate the inbound Options and return a struct
 func getOpts(opt ...Option) options {
@@ -16,15 +18,45 @@ type Option func(*options)
 
 // options = how options are represented
 type options struct {
-	withLimit      int
-	withLiveness   time.Duration
-	withUpdateTags bool
+	withName        string
+	withDescription string
+	withAddress     string
+	withPublicId    string
+	withLimit       int
+	withLiveness    time.Duration
+	withUpdateTags  bool
+	withWorkerTags  []*Tag
 }
 
 func getDefaultOptions() options {
-	return options{
-		withLimit:    0,
-		withLiveness: 0,
+	return options{}
+}
+
+// WithDescription provides an optional description.
+func WithDescription(desc string) Option {
+	return func(o *options) {
+		o.withDescription = desc
+	}
+}
+
+// WithName provides an optional name.
+func WithName(name string) Option {
+	return func(o *options) {
+		o.withName = name
+	}
+}
+
+// WithAddress provides an optional address.
+func WithAddress(address string) Option {
+	return func(o *options) {
+		o.withAddress = address
+	}
+}
+
+// WithPublicId provides an optional public id
+func WithPublicId(id string) Option {
+	return func(o *options) {
+		o.withPublicId = id
 	}
 }
 
@@ -54,3 +86,10 @@ func WithUpdateTags(updateTags bool) Option {
 		o.withUpdateTags = updateTags
 	}
 }
+
+// WithWorkerTags provides worker tags.
+func WithWorkerTags(tags ...*Tag) Option {
+	return func(o *options) {
+		o.withWorkerTags = tags
+	}
+}

internal/servers/options_test.go

@@ -13,15 +13,56 @@ func Test_GetOpts(t *testing.T) {
 	opts := getOpts()
 	assert.Equal(t, options{}, opts)
 
-	opts = getOpts(
-		WithLimit(5),
-		WithLiveness(time.Hour),
-		WithUpdateTags(true),
-	)
-	exp := options{
-		withLimit:      5,
-		withLiveness:   time.Hour,
-		withUpdateTags: true,
-	}
-	assert.Equal(t, exp, opts)
+	t.Run("WithName", func(t *testing.T) {
+		opts := getOpts(WithName("test"))
+		testOpts := getDefaultOptions()
+		testOpts.withName = "test"
+		assert.Equal(t, opts, testOpts)
+	})
+	t.Run("WithDescription", func(t *testing.T) {
+		opts := getOpts(WithDescription("test desc"))
+		testOpts := getDefaultOptions()
+		testOpts.withDescription = "test desc"
+		assert.Equal(t, opts, testOpts)
+	})
+	t.Run("WithLimit", func(t *testing.T) {
+		opts := getOpts(WithLimit(5))
+		testOpts := getDefaultOptions()
+		testOpts.withLimit = 5
+		assert.Equal(t, opts, testOpts)
+	})
+	t.Run("WithAddress", func(t *testing.T) {
+		opts := getOpts(WithAddress("test"))
+		testOpts := getDefaultOptions()
+		testOpts.withAddress = "test"
+		assert.Equal(t, opts, testOpts)
+	})
+	t.Run("WithPublicId", func(t *testing.T) {
+		opts := getOpts(WithPublicId("test"))
+		testOpts := getDefaultOptions()
+		testOpts.withPublicId = "test"
+		assert.Equal(t, opts, testOpts)
+	})
+	t.Run("WithLiveness", func(t *testing.T) {
+		opts := getOpts(WithLiveness(time.Hour))
+		testOpts := getDefaultOptions()
+		testOpts.withLiveness = time.Hour
+		assert.Equal(t, opts, testOpts)
+	})
+	t.Run("WithUpdateTags", func(t *testing.T) {
+		opts := getOpts(WithUpdateTags(true))
+		testOpts := getDefaultOptions()
+		testOpts.withUpdateTags = true
+		assert.Equal(t, opts, testOpts)
+	})
+	t.Run("WithWorkerTags", func(t *testing.T) {
+		tags := []*Tag{
+			{Key: "key1", Value: "val1"},
+			{Key: "key2", Value: "val2"},
+		}
+		opts := getOpts(WithWorkerTags(tags...))
+		testOpts := getDefaultOptions()
+		testOpts.withWorkerTags = tags
+		assert.Equal(t, opts, testOpts)
+	})
 }

internal/servers/public_ids.go

@@ -0,0 +1,21 @@
+package servers
+
+import (
+	"context"
+
+	"github.com/hashicorp/boundary/internal/db"
+	"github.com/hashicorp/boundary/internal/errors"
+)
+
+// PublicId prefixes for the resources in the servers package.
+const (
+	WorkerPrefix = "w"
+)
+
+func newWorkerId(ctx context.Context) (string, error) {
+	id, err := db.NewPublicId(WorkerPrefix)
+	if err != nil {
+		return "", errors.Wrap(ctx, err, "servers.newWorkerId")
+	}
+	return id, nil
+}

internal/servers/repository.go

@@ -24,17 +24,6 @@ const (
 	DefaultLiveness = 15 * time.Second
 )
 
-type ServerType string
-
-const (
-	ServerTypeController ServerType = "controller"
-	ServerTypeWorker     ServerType = "worker"
-)
-
-func (s ServerType) String() string {
-	return string(s)
-}
-
 // Repository is the servers database repository
 type Repository struct {
 	reader db.Reader
@@ -63,13 +52,13 @@ func NewRepository(r db.Reader, w db.Writer, kms *kms.Kms) (*Repository, error)
 }
 
 // ListWorkers is a passthrough to listWorkersWithReader that uses the repo's normal reader.
-func (r *Repository) ListWorkers(ctx context.Context, opt ...Option) ([]*store.Worker, error) {
+func (r *Repository) ListWorkers(ctx context.Context, opt ...Option) ([]*Worker, error) {
 	return r.listWorkersWithReader(ctx, r.reader, opt...)
 }
 
 // listWorkersWithReader will return a listing of resources and honor the WithLimit option or the repo
 // defaultLimit. It accepts a reader, allowing it to be used within a transaction or without.
-func (r *Repository) listWorkersWithReader(ctx context.Context, reader db.Reader, opt ...Option) ([]*store.Worker, error) {
+func (r *Repository) listWorkersWithReader(ctx context.Context, reader db.Reader, opt ...Option) ([]*Worker, error) {
 	opts := getOpts(opt...)
 	liveness := opts.withLiveness
 	if liveness == 0 {
@@ -81,7 +70,7 @@ func (r *Repository) listWorkersWithReader(ctx context.Context, reader db.Reader
 		where = fmt.Sprintf("update_time > now() - interval '%d seconds'", uint32(liveness.Seconds()))
 	}
 
-	var workers []*store.Worker
+	var workers []*Worker
 	if err := reader.SearchWhere(
 		ctx,
 		&workers,
@@ -128,16 +117,10 @@ func (r *Repository) listControllersWithReader(ctx context.Context, reader db.Re
 	return controllers, nil
 }
 
-// WorkerTag holds the information for the worker_tag table for Gorm.
-type WorkerTag struct {
-	WorkerId string
-	Key      string
-	Value    string
-}
-
-// ListTagsForWorkers pulls out tag tuples into WorkerTag structs for the given worker
-func (r *Repository) ListTagsForWorkers(ctx context.Context, workerIds []string, opt ...Option) ([]*WorkerTag, error) {
-	var workerTags []*WorkerTag
+// ListTagsForWorkers returns a map from the worker's id to the list of Tags
+// that are for that worker.  All options are ignored.
+func (r *Repository) ListTagsForWorkers(ctx context.Context, workerIds []string, opt ...Option) (map[string][]*Tag, error) {
+	var workerTags []*store.WorkerTag
 	if err := r.reader.SearchWhere(
 		ctx,
 		&workerTags,
@@ -147,14 +130,29 @@ func (r *Repository) ListTagsForWorkers(ctx context.Context, workerIds []string,
 	); err != nil {
 		return nil, errors.Wrap(ctx, err, "servers.ListTagsForWorkers", errors.WithMsg(fmt.Sprintf("worker IDs %v", workerIds)))
 	}
-	return workerTags, nil
+
+	ret := make(map[string][]*Tag, len(workerIds))
+	for _, t := range workerTags {
+		ret[t.WorkerId] = append(ret[t.WorkerId], &Tag{Key: t.Key, Value: t.Value})
+	}
+	return ret, nil
 }
 
-func (r *Repository) UpsertWorker(ctx context.Context, worker *store.Worker, opt ...Option) ([]*store.Controller, int, error) {
+// UpsertWorker creates a new worker if one with the provided public id doesn't
+// already exist. If it does, UpsertWorker updates the worker. The
+// WithUpdateTags option is the only one used. All others are ignored.
+func (r *Repository) UpsertWorker(ctx context.Context, worker *Worker, opt ...Option) ([]*store.Controller, int, error) {
 	const op = "servers.UpsertWorker"
 
-	if worker == nil {
+	switch {
+	case worker == nil:
 		return nil, db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "worker is nil")
+	case worker.Address == "":
+		return nil, db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "address is empty")
+	case worker.ScopeId == "":
+		return nil, db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "scope id is empty")
+	case worker.PublicId == "":
+		return nil, db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "public id is empty")
 	}
 
 	opts := getOpts(opt...)
@@ -187,9 +185,9 @@ func (r *Repository) UpsertWorker(ctx context.Context, worker *store.Worker, opt
 			// delete all tags for the given worker, then add the new ones
 			// we've been sent.
 			if opts.withUpdateTags {
-				_, err = w.Delete(ctx, &WorkerTag{}, db.WithWhere(deleteTagsSql, worker.PublicId))
+				_, err = w.Delete(ctx, &store.WorkerTag{}, db.WithWhere(deleteTagsSql, worker.GetPublicId()))
 				if err != nil {
-					return errors.Wrap(ctx, err, op+":DeleteTags", errors.WithMsg(worker.PublicId))
+					return errors.Wrap(ctx, err, op+":DeleteTags", errors.WithMsg(worker.GetPublicId()))
 				}
 
 				// If tags were cleared out entirely, then we'll have nothing
@@ -198,20 +196,18 @@ func (r *Repository) UpsertWorker(ctx context.Context, worker *store.Worker, opt
 				// below.
 				if len(worker.Tags) > 0 {
 					tags := make([]interface{}, 0, len(worker.Tags))
-					for k, v := range worker.Tags {
+					for _, v := range worker.Tags {
 						if v == nil {
-							return errors.New(ctx, errors.InvalidParameter, op+":RangeTags", fmt.Sprintf("found nil tag value for worker %s and key %s", worker.PublicId, k))
-						}
-						for _, val := range v.Values {
-							tags = append(tags, WorkerTag{
-								WorkerId: worker.PublicId,
-								Key:      k,
-								Value:    val,
-							})
+							return errors.New(ctx, errors.InvalidParameter, op+":RangeTags", fmt.Sprintf("found nil tag value for worker %s", worker.GetPublicId()))
 						}
+						tags = append(tags, &store.WorkerTag{
+							WorkerId: worker.GetPublicId(),
+							Key:      v.Key,
+							Value:    v.Value,
+						})
 					}
 					if err = w.CreateItems(ctx, tags); err != nil {
-						return errors.Wrap(ctx, err, op+":CreateTags", errors.WithMsg(worker.PublicId))
+						return errors.Wrap(ctx, err, op+":CreateTags", errors.WithMsg(worker.GetPublicId()))
 					}
 				}
 			}

internal/servers/repository_test.go

@@ -5,8 +5,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/hashicorp/boundary/internal/servers/store"
-
 	"github.com/hashicorp/boundary/api/recovery"
 	"github.com/hashicorp/boundary/api/roles"
 	"github.com/hashicorp/boundary/globals"
@@ -99,28 +97,34 @@ func TestTagUpdatingListing(t *testing.T) {
 	defer tc.Shutdown()
 
 	repo := tc.ServersRepo()
-	srv := &store.Worker{
-		PublicId: "test_worker_1",
-		Address:  "127.0.0.1",
-		Tags: map[string]*store.TagValues{
-			"tag1": {
-				Values: []string{"value1", "value2"},
+	srv := servers.NewWorker(scope.Global.String(),
+		servers.WithPublicId("test_worker_1"),
+		servers.WithAddress("127.0.0.1"),
+		servers.WithWorkerTags(
+			&servers.Tag{
+				Key:   "tag1",
+				Value: "value1",
 			},
-		},
-	}
+			&servers.Tag{
+				Key:   "tag1",
+				Value: "value2",
+			}))
 
 	_, _, err := repo.UpsertWorker(tc.Context(), srv, servers.WithUpdateTags(true))
 	require.NoError(err)
 
-	srv = &store.Worker{
-		PublicId: "test_worker_2",
-		Address:  "127.0.0.1",
-		Tags: map[string]*store.TagValues{
-			"tag2": {
-				Values: []string{"value1", "value2"},
+	srv = servers.NewWorker(scope.Global.String(),
+		servers.WithPublicId("test_worker_2"),
+		servers.WithAddress("127.0.0.1"),
+		servers.WithWorkerTags(
+			&servers.Tag{
+				Key:   "tag2",
+				Value: "value1",
 			},
-		},
-	}
+			&servers.Tag{
+				Key:   "tag2",
+				Value: "value2",
+			}))
 	_, _, err = repo.UpsertWorker(tc.Context(), srv, servers.WithUpdateTags(true))
 	require.NoError(err)
 
@@ -128,40 +132,42 @@ func TestTagUpdatingListing(t *testing.T) {
 	require.NoError(err)
 
 	// Base case
-	exp := []*servers.WorkerTag{
-		{
-			WorkerId: "test_worker_1",
-			Key:      "tag1",
-			Value:    "value1",
+	exp := map[string][]*servers.Tag{
+		"test_worker_1": {
+			{
+				Key:   "tag1",
+				Value: "value1",
+			}, {
+				Key:   "tag1",
+				Value: "value2",
+			},
 		},
-		{
-			WorkerId: "test_worker_1",
-			Key:      "tag1",
-			Value:    "value2",
-		},
-		{
-			WorkerId: "test_worker_2",
-			Key:      "tag2",
-			Value:    "value1",
-		},
-		{
-			WorkerId: "test_worker_2",
-			Key:      "tag2",
-			Value:    "value2",
+		"test_worker_2": {
+			{
+				Key:   "tag2",
+				Value: "value1",
+			},
+			{
+				Key:   "tag2",
+				Value: "value2",
+			},
 		},
 	}
 	require.Equal(exp, tags)
 
 	// Update without saying to update tags
-	srv = &store.Worker{
-		PublicId: "test_worker_2",
-		Address:  "192.168.1.1",
-		Tags: map[string]*store.TagValues{
-			"tag22": {
-				Values: []string{"value21", "value22"},
+	srv = servers.NewWorker(scope.Global.String(),
+		servers.WithPublicId("test_worker_2"),
+		servers.WithAddress("192.168.1.1"),
+		servers.WithWorkerTags(
+			&servers.Tag{
+				Key:   "tag22",
+				Value: "value21",
 			},
-		},
-	}
+			&servers.Tag{
+				Key:   "tag22",
+				Value: "value22",
+			}))
 	_, _, err = repo.UpsertWorker(tc.Context(), srv)
 	require.NoError(err)
 	tags, err = repo.ListTagsForWorkers(tc.Context(), []string{"test_worker_1", "test_worker_2"})
@@ -175,15 +181,15 @@ func TestTagUpdatingListing(t *testing.T) {
 	require.NoError(err)
 	require.NotEqual(exp, tags)
 	// Update and try again
-	exp[2] = &servers.WorkerTag{
-		WorkerId: "test_worker_2",
-		Key:      "tag22",
-		Value:    "value21",
-	}
-	exp[3] = &servers.WorkerTag{
-		WorkerId: "test_worker_2",
-		Key:      "tag22",
-		Value:    "value22",
+	exp["test_worker_2"] = []*servers.Tag{
+		{
+			Key:   "tag22",
+			Value: "value21",
+		},
+		{
+			Key:   "tag22",
+			Value: "value22",
+		},
 	}
 	require.Equal(exp, tags)
 }
@@ -199,11 +205,10 @@ func TestListServersWithLiveness(t *testing.T) {
 	require.NoError(err)
 	ctx := context.Background()
 
-	newWorker := func(publicId string) *store.Worker {
-		result := &store.Worker{
-			PublicId: publicId,
-			Address:  "127.0.0.1",
-		}
+	newWorker := func(publicId string) *servers.Worker {
+		result := servers.NewWorker(scope.Global.String(),
+			servers.WithPublicId(publicId),
+			servers.WithAddress("127.0.0.1"))
 		_, rowsUpdated, err := serversRepo.UpsertWorker(ctx, result)
 		require.NoError(err)
 		require.Equal(1, rowsUpdated)
@@ -224,7 +229,7 @@ func TestListServersWithLiveness(t *testing.T) {
 	require.NoError(err)
 	require.Equal(1, rowsUpdated)
 
-	requireIds := func(expected []string, actual []*store.Worker) {
+	requireIds := func(expected []string, actual []*servers.Worker) {
 		require.Len(expected, len(actual))
 		want := make(map[string]struct{})
 		for _, v := range expected {

internal/servers/store/gorm.go

@@ -1,11 +1,11 @@
 package store
 
-// TableName overrides the table name used by Worker to `server_worker`
-func (Worker) TableName() string {
-	return "server_worker"
-}
-
 // TableName overrides the table name used by Controller to `server_controller`
 func (Controller) TableName() string {
 	return "server_controller"
 }
+
+// TableName overrides the table name used by WorkerTag to `worker_tag`
+func (WorkerTag) TableName() string {
+	return "server_worker_tag"
+}

internal/servers/store/worker.pb.go

@@ -27,24 +27,30 @@ type Worker struct {
 	sizeCache     protoimpl.SizeCache
 	unknownFields protoimpl.UnknownFields
 
-	// Public ID of the resource
+	// public_id is a surrogate key suitable for use in a public API
 	// @inject_tag: `gorm:"primary_key"`
 	PublicId string `protobuf:"bytes,10,opt,name=public_id,json=publicId,proto3" json:"public_id,omitempty" gorm:"primary_key"`
 	// Name of the resource (optional)
-	// Description of the resource (optional)
 	// @inject_tag: `gorm:"default:null"`
 	Name string `protobuf:"bytes,20,opt,name=name,proto3" json:"name,omitempty" gorm:"default:null"`
 	// Description of the resource (optional)
 	// @inject_tag: `gorm:"default:null"`
 	Description string `protobuf:"bytes,30,opt,name=description,proto3" json:"description,omitempty" gorm:"default:null"`
-	// Address for the worker
-	Address string `protobuf:"bytes,40,opt,name=address,proto3" json:"address,omitempty"`
-	// First seen time from the RDBMS
-	CreateTime *timestamp.Timestamp `protobuf:"bytes,50,opt,name=create_time,json=createTime,proto3" json:"create_time,omitempty"`
-	// Last time there was an update
-	UpdateTime *timestamp.Timestamp `protobuf:"bytes,60,opt,name=update_time,json=updateTime,proto3" json:"update_time,omitempty"`
-	// @gotags: `gorm:"-"`
-	Tags map[string]*TagValues `protobuf:"bytes,70,rep,name=tags,proto3" json:"tags,omitempty" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3" gorm:"-"`
+	// Address for the worker. This must be set
+	// @inject_tag: `gorm:"default:null"`
+	Address string `protobuf:"bytes,40,opt,name=address,proto3" json:"address,omitempty" gorm:"default:null"`
+	// The create_time is set by the database.
+	// @inject_tag: `gorm:"default:current_timestamp"`
+	CreateTime *timestamp.Timestamp `protobuf:"bytes,50,opt,name=create_time,json=createTime,proto3" json:"create_time,omitempty" gorm:"default:current_timestamp"`
+	// The update_time is set by the database.
+	// @inject_tag: `gorm:"default:current_timestamp"`
+	UpdateTime *timestamp.Timestamp `protobuf:"bytes,60,opt,name=update_time,json=updateTime,proto3" json:"update_time,omitempty" gorm:"default:current_timestamp"`
+	// The scope_id of the owning scope and must be set.
+	// @inject_tag: `gorm:"not_null"`
+	ScopeId string `protobuf:"bytes,70,opt,name=scope_id,json=scopeId,proto3" json:"scope_id,omitempty" gorm:"not_null"`
+	// version allows optimistic locking of the resource.
+	// @inject_tag: `gorm:"default:null"`
+	Version uint32 `protobuf:"varint,80,opt,name=version,proto3" json:"version,omitempty" gorm:"default:null"`
 }
 
 func (x *Worker) Reset() {
@@ -121,25 +127,40 @@ func (x *Worker) GetUpdateTime() *timestamp.Timestamp {
 	return nil
 }
 
-func (x *Worker) GetTags() map[string]*TagValues {
+func (x *Worker) GetScopeId() string {
 	if x != nil {
-		return x.Tags
+		return x.ScopeId
 	}
-	return nil
+	return ""
 }
 
-// TagValues is used because map fields cannot be repeated but can be a
-// message
-type TagValues struct {
+func (x *Worker) GetVersion() uint32 {
+	if x != nil {
+		return x.Version
+	}
+	return 0
+}
+
+// WorkerTag is a tag for a worker.  The primary key is comprised of the
+// worker_id, key, value, and source.
+type WorkerTag struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
 	unknownFields protoimpl.UnknownFields
 
-	Values []string `protobuf:"bytes,10,rep,name=values,proto3" json:"values,omitempty"`
+	// worker_id is the public key that key of the worker this tag is for.
+	// @inject_tag: `gorm:"default:null"`
+	WorkerId string `protobuf:"bytes,10,opt,name=worker_id,json=workerId,proto3" json:"worker_id,omitempty" gorm:"default:null"`
+	// key is the key of the tag. This must be set.
+	// @inject_tag: `gorm:"default:null"`
+	Key string `protobuf:"bytes,20,opt,name=key,proto3" json:"key,omitempty" gorm:"default:null"`
+	// value is the value
+	// @inject_tag: `gorm:"default:null"`
+	Value string `protobuf:"bytes,30,opt,name=value,proto3" json:"value,omitempty" gorm:"default:null"`
 }
 
-func (x *TagValues) Reset() {
-	*x = TagValues{}
+func (x *WorkerTag) Reset() {
+	*x = WorkerTag{}
 	if protoimpl.UnsafeEnabled {
 		mi := &file_controller_storage_servers_store_v1_worker_proto_msgTypes[1]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
@@ -147,13 +168,13 @@ func (x *TagValues) Reset() {
 	}
 }
 
-func (x *TagValues) String() string {
+func (x *WorkerTag) String() string {
 	return protoimpl.X.MessageStringOf(x)
 }
 
-func (*TagValues) ProtoMessage() {}
+func (*WorkerTag) ProtoMessage() {}
 
-func (x *TagValues) ProtoReflect() protoreflect.Message {
+func (x *WorkerTag) ProtoReflect() protoreflect.Message {
 	mi := &file_controller_storage_servers_store_v1_worker_proto_msgTypes[1]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
@@ -165,16 +186,30 @@ func (x *TagValues) ProtoReflect() protoreflect.Message {
 	return mi.MessageOf(x)
 }
 
-// Deprecated: Use TagValues.ProtoReflect.Descriptor instead.
-func (*TagValues) Descriptor() ([]byte, []int) {
+// Deprecated: Use WorkerTag.ProtoReflect.Descriptor instead.
+func (*WorkerTag) Descriptor() ([]byte, []int) {
 	return file_controller_storage_servers_store_v1_worker_proto_rawDescGZIP(), []int{1}
 }
 
-func (x *TagValues) GetValues() []string {
+func (x *WorkerTag) GetWorkerId() string {
 	if x != nil {
-		return x.Values
+		return x.WorkerId
 	}
-	return nil
+	return ""
+}
+
+func (x *WorkerTag) GetKey() string {
+	if x != nil {
+		return x.Key
+	}
+	return ""
+}
+
+func (x *WorkerTag) GetValue() string {
+	if x != nil {
+		return x.Value
+	}
+	return ""
 }
 
 var File_controller_storage_servers_store_v1_worker_proto protoreflect.FileDescriptor
@@ -188,7 +223,7 @@ var file_controller_storage_servers_store_v1_worker_proto_rawDesc = []byte{
 	0x74, 0x6f, 0x72, 0x65, 0x2e, 0x76, 0x31, 0x1a, 0x2f, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c,
 	0x6c, 0x65, 0x72, 0x2f, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2f, 0x74, 0x69, 0x6d, 0x65,
 	0x73, 0x74, 0x61, 0x6d, 0x70, 0x2f, 0x76, 0x31, 0x2f, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61,
-	0x6d, 0x70, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0xc3, 0x03, 0x0a, 0x06, 0x57, 0x6f, 0x72,
+	0x6d, 0x70, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0xc4, 0x02, 0x0a, 0x06, 0x57, 0x6f, 0x72,
 	0x6b, 0x65, 0x72, 0x12, 0x1b, 0x0a, 0x09, 0x70, 0x75, 0x62, 0x6c, 0x69, 0x63, 0x5f, 0x69, 0x64,
 	0x18, 0x0a, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x70, 0x75, 0x62, 0x6c, 0x69, 0x63, 0x49, 0x64,
 	0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x14, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04,
@@ -205,25 +240,20 @@ var file_controller_storage_servers_store_v1_worker_proto_rawDesc = []byte{
 	0x28, 0x0b, 0x32, 0x2a, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e,
 	0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d,
 	0x70, 0x2e, 0x76, 0x31, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x0a,
-	0x75, 0x70, 0x64, 0x61, 0x74, 0x65, 0x54, 0x69, 0x6d, 0x65, 0x12, 0x49, 0x0a, 0x04, 0x74, 0x61,
-	0x67, 0x73, 0x18, 0x46, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x35, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72,
-	0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, 0x73, 0x65,
-	0x72, 0x76, 0x65, 0x72, 0x73, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x2e, 0x76, 0x31, 0x2e, 0x57,
-	0x6f, 0x72, 0x6b, 0x65, 0x72, 0x2e, 0x54, 0x61, 0x67, 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52,
-	0x04, 0x74, 0x61, 0x67, 0x73, 0x1a, 0x67, 0x0a, 0x09, 0x54, 0x61, 0x67, 0x73, 0x45, 0x6e, 0x74,
-	0x72, 0x79, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52,
-	0x03, 0x6b, 0x65, 0x79, 0x12, 0x44, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20,
-	0x01, 0x28, 0x0b, 0x32, 0x2e, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72,
-	0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73,
-	0x2e, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x2e, 0x76, 0x31, 0x2e, 0x54, 0x61, 0x67, 0x56, 0x61, 0x6c,
-	0x75, 0x65, 0x73, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x3a, 0x02, 0x38, 0x01, 0x22, 0x23,
-	0x0a, 0x09, 0x54, 0x61, 0x67, 0x56, 0x61, 0x6c, 0x75, 0x65, 0x73, 0x12, 0x16, 0x0a, 0x06, 0x76,
-	0x61, 0x6c, 0x75, 0x65, 0x73, 0x18, 0x0a, 0x20, 0x03, 0x28, 0x09, 0x52, 0x06, 0x76, 0x61, 0x6c,
-	0x75, 0x65, 0x73, 0x42, 0x3c, 0x5a, 0x3a, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f,
-	0x6d, 0x2f, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2f, 0x62, 0x6f, 0x75, 0x6e,
-	0x64, 0x61, 0x72, 0x79, 0x2f, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x73, 0x65,
-	0x72, 0x76, 0x65, 0x72, 0x73, 0x2f, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x3b, 0x73, 0x74, 0x6f, 0x72,
-	0x65, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+	0x75, 0x70, 0x64, 0x61, 0x74, 0x65, 0x54, 0x69, 0x6d, 0x65, 0x12, 0x19, 0x0a, 0x08, 0x73, 0x63,
+	0x6f, 0x70, 0x65, 0x5f, 0x69, 0x64, 0x18, 0x46, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x73, 0x63,
+	0x6f, 0x70, 0x65, 0x49, 0x64, 0x12, 0x18, 0x0a, 0x07, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e,
+	0x18, 0x50, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x07, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x22,
+	0x50, 0x0a, 0x09, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x54, 0x61, 0x67, 0x12, 0x1b, 0x0a, 0x09,
+	0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x5f, 0x69, 0x64, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x09, 0x52,
+	0x08, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x49, 0x64, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79,
+	0x18, 0x14, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x14, 0x0a, 0x05, 0x76,
+	0x61, 0x6c, 0x75, 0x65, 0x18, 0x1e, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75,
+	0x65, 0x42, 0x3c, 0x5a, 0x3a, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f,
+	0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2f, 0x62, 0x6f, 0x75, 0x6e, 0x64, 0x61,
+	0x72, 0x79, 0x2f, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x73, 0x65, 0x72, 0x76,
+	0x65, 0x72, 0x73, 0x2f, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x3b, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x62,
+	0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (
@@ -238,23 +268,20 @@ func file_controller_storage_servers_store_v1_worker_proto_rawDescGZIP() []byte
 	return file_controller_storage_servers_store_v1_worker_proto_rawDescData
 }
 
-var file_controller_storage_servers_store_v1_worker_proto_msgTypes = make([]protoimpl.MessageInfo, 3)
+var file_controller_storage_servers_store_v1_worker_proto_msgTypes = make([]protoimpl.MessageInfo, 2)
 var file_controller_storage_servers_store_v1_worker_proto_goTypes = []interface{}{
 	(*Worker)(nil),              // 0: controller.storage.servers.store.v1.Worker
-	(*TagValues)(nil),           // 1: controller.storage.servers.store.v1.TagValues
-	nil,                         // 2: controller.storage.servers.store.v1.Worker.TagsEntry
-	(*timestamp.Timestamp)(nil), // 3: controller.storage.timestamp.v1.Timestamp
+	(*WorkerTag)(nil),           // 1: controller.storage.servers.store.v1.WorkerTag
+	(*timestamp.Timestamp)(nil), // 2: controller.storage.timestamp.v1.Timestamp
 }
 var file_controller_storage_servers_store_v1_worker_proto_depIdxs = []int32{
-	3, // 0: controller.storage.servers.store.v1.Worker.create_time:type_name -> controller.storage.timestamp.v1.Timestamp
-	3, // 1: controller.storage.servers.store.v1.Worker.update_time:type_name -> controller.storage.timestamp.v1.Timestamp
-	2, // 2: controller.storage.servers.store.v1.Worker.tags:type_name -> controller.storage.servers.store.v1.Worker.TagsEntry
-	1, // 3: controller.storage.servers.store.v1.Worker.TagsEntry.value:type_name -> controller.storage.servers.store.v1.TagValues
-	4, // [4:4] is the sub-list for method output_type
-	4, // [4:4] is the sub-list for method input_type
-	4, // [4:4] is the sub-list for extension type_name
-	4, // [4:4] is the sub-list for extension extendee
-	0, // [0:4] is the sub-list for field type_name
+	2, // 0: controller.storage.servers.store.v1.Worker.create_time:type_name -> controller.storage.timestamp.v1.Timestamp
+	2, // 1: controller.storage.servers.store.v1.Worker.update_time:type_name -> controller.storage.timestamp.v1.Timestamp
+	2, // [2:2] is the sub-list for method output_type
+	2, // [2:2] is the sub-list for method input_type
+	2, // [2:2] is the sub-list for extension type_name
+	2, // [2:2] is the sub-list for extension extendee
+	0, // [0:2] is the sub-list for field type_name
 }
 
 func init() { file_controller_storage_servers_store_v1_worker_proto_init() }
@@ -276,7 +303,7 @@ func file_controller_storage_servers_store_v1_worker_proto_init() {
 			}
 		}
 		file_controller_storage_servers_store_v1_worker_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*TagValues); i {
+			switch v := v.(*WorkerTag); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -294,7 +321,7 @@ func file_controller_storage_servers_store_v1_worker_proto_init() {
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: file_controller_storage_servers_store_v1_worker_proto_rawDesc,
 			NumEnums:      0,
-			NumMessages:   3,
+			NumMessages:   2,
 			NumExtensions: 0,
 			NumServices:   0,
 		},

internal/servers/worker.go

@@ -0,0 +1,55 @@
+package servers
+
+import (
+	"github.com/hashicorp/boundary/internal/servers/store"
+	"google.golang.org/protobuf/proto"
+)
+
+// A Worker is a server that provides an address which can be used to proxy
+// session connections. It can be tagged with custom tags and is used when
+// authorizing and establishing a session.  It is owned by a scope.
+type Worker struct {
+	*store.Worker
+	Tags []*Tag `gorm:"-"`
+}
+
+// A Tag is a custom key/value pair which can be attached to a Worker.
+// Multiple Tags may contain the same key and different values in which
+// case both key/value pairs are valid.
+type Tag struct {
+	Key   string
+	Value string
+}
+
+// NewWorker returns a new Worker. Valid options are WithName, WithDescription
+// WithAddress, and WithWorkerTags. All other options are ignored.
+func NewWorker(scopeId string, opt ...Option) *Worker {
+	opts := getOpts(opt...)
+	return &Worker{
+		Worker: &store.Worker{
+			ScopeId:     scopeId,
+			PublicId:    opts.withPublicId,
+			Name:        opts.withName,
+			Description: opts.withDescription,
+			Address:     opts.withAddress,
+		},
+		Tags: opts.withWorkerTags,
+	}
+}
+
+func (w *Worker) clone() *Worker {
+	tags := make([]*Tag, 0, len(w.Tags))
+	for _, t := range w.Tags {
+		tags = append(tags, &Tag{Key: t.Key, Value: t.Value})
+	}
+	cw := proto.Clone(w.Worker)
+	return &Worker{
+		Worker: cw.(*store.Worker),
+		Tags:   tags,
+	}
+}
+
+// TableName overrides the table name used by Worker to `server_worker`
+func (Worker) TableName() string {
+	return "server_worker"
+}

internal/servers/worker_tags_test.go

@@ -0,0 +1,78 @@
+package servers
+
+import (
+	"context"
+	"testing"
+
+	"github.com/hashicorp/boundary/internal/db"
+	"github.com/hashicorp/boundary/internal/servers/store"
+	"github.com/hashicorp/boundary/internal/types/scope"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestWorkerTags_Create(t *testing.T) {
+	t.Parallel()
+	conn, _ := db.TestSetup(t, "postgres")
+	rw := db.New(conn)
+
+	workerPublicid, err := newWorkerId(context.Background())
+	require.NoError(t, err)
+	worker := NewWorker(scope.Global.String(), WithPublicId(workerPublicid))
+
+	tests := []struct {
+		name          string
+		want          *store.WorkerTag
+		wantCreateErr bool
+	}{
+		{
+			name: "success",
+			want: &store.WorkerTag{
+				WorkerId: worker.GetPublicId(),
+				Key:      "key",
+				Value:    "value",
+			},
+		},
+		{
+			name: "bad worker id",
+			want: &store.WorkerTag{
+				WorkerId: "w_badworkeridthatdoesntexist",
+				Key:      "key",
+				Value:    "value",
+			},
+			wantCreateErr: true,
+		},
+		{
+			name: "missing worker id",
+			want: &store.WorkerTag{
+				Key:   "key",
+				Value: "value",
+			},
+			wantCreateErr: true,
+		},
+		{
+			name: "missing key",
+			want: &store.WorkerTag{
+				WorkerId: worker.GetPublicId(),
+				Value:    "value",
+			},
+		},
+		{
+			name: "missing value",
+			want: &store.WorkerTag{
+				WorkerId: worker.GetPublicId(),
+				Key:      "key",
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			err := rw.Create(context.Background(), tt.want)
+			if tt.wantCreateErr {
+				assert.Error(t, err)
+			}
+		})
+	}
+}

internal/servers/worker_test.go

@@ -0,0 +1,185 @@
+package servers
+
+import (
+	"context"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/hashicorp/boundary/internal/db"
+	"github.com/hashicorp/boundary/internal/iam"
+	"github.com/hashicorp/boundary/internal/servers/store"
+	"github.com/hashicorp/boundary/internal/types/scope"
+	"github.com/stretchr/testify/assert"
+	"google.golang.org/protobuf/testing/protocmp"
+)
+
+func TestWorker_New(t *testing.T) {
+	t.Parallel()
+	conn, _ := db.TestSetup(t, "postgres")
+	wrapper := db.TestWrapper(t)
+	rw := db.New(conn)
+
+	org, prj := iam.TestScopes(t, iam.TestRepo(t, conn, wrapper))
+
+	type args struct {
+		scopeId string
+		opts    []Option
+	}
+
+	tests := []struct {
+		name          string
+		args          args
+		want          *Worker
+		wantCreateErr bool
+	}{
+		{
+			name: "missing-scope-id",
+			args: args{},
+			want: &Worker{
+				Worker: &store.Worker{},
+			},
+			wantCreateErr: true,
+		},
+		{
+			name: "global-scope-id",
+			args: args{
+				scopeId: scope.Global.String(),
+			},
+			want: &Worker{
+				Worker: &store.Worker{
+					ScopeId: scope.Global.String(),
+				},
+			},
+		},
+		{
+			name: "org-scope-id",
+			args: args{
+				scopeId: org.GetPublicId(),
+			},
+			want: &Worker{
+				Worker: &store.Worker{
+					ScopeId: org.GetPublicId(),
+				},
+			},
+			wantCreateErr: true,
+		},
+		{
+			name: "project-scope-id",
+			args: args{
+				scopeId: prj.GetPublicId(),
+			},
+			want: &Worker{
+				Worker: &store.Worker{
+					ScopeId: prj.GetPublicId(),
+				},
+			},
+			wantCreateErr: true,
+		},
+		{
+			name: "with public id",
+			args: args{
+				scopeId: scope.Global.String(),
+				opts: []Option{
+					WithPublicId("w_test_public_id"),
+				},
+			},
+			want: &Worker{
+				Worker: &store.Worker{
+					PublicId: "w_test_public_id",
+					ScopeId:  scope.Global.String(),
+				},
+			},
+		},
+		{
+			name: "with name",
+			args: args{
+				scopeId: scope.Global.String(),
+				opts: []Option{
+					WithName("foo"),
+				},
+			},
+			want: &Worker{
+				Worker: &store.Worker{
+					ScopeId: scope.Global.String(),
+					Name:    "foo",
+				},
+			},
+		},
+		{
+			name: "with description",
+			args: args{
+				scopeId: scope.Global.String(),
+				opts: []Option{
+					WithDescription("foo"),
+				},
+			},
+			want: &Worker{
+				Worker: &store.Worker{
+					ScopeId:     scope.Global.String(),
+					Description: "foo",
+				},
+			},
+		},
+		{
+			name: "with address",
+			args: args{
+				scopeId: scope.Global.String(),
+				opts: []Option{
+					WithAddress("foo"),
+				},
+			},
+			want: &Worker{
+				Worker: &store.Worker{
+					ScopeId: scope.Global.String(),
+					Address: "foo",
+				},
+			},
+		},
+		{
+			name: "with Tags",
+			args: args{
+				scopeId: scope.Global.String(),
+				opts: []Option{
+					WithWorkerTags(&Tag{
+						Key:   "key",
+						Value: "val",
+					}),
+				},
+			},
+			want: &Worker{
+				Worker: &store.Worker{
+					ScopeId: scope.Global.String(),
+				},
+				Tags: []*Tag{
+					{
+						Key:   "key",
+						Value: "val",
+					},
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			assert := assert.New(t)
+			got := NewWorker(tt.args.scopeId, tt.args.opts...)
+			assert.Empty(cmp.Diff(tt.want, got.clone(), protocmp.Transform()))
+
+			id, err := newWorkerId(context.Background())
+			assert.NoError(err)
+
+			tt.want.PublicId = id
+			got.PublicId = id
+
+			err2 := rw.Create(context.Background(), got)
+			if tt.wantCreateErr {
+				assert.Error(err2)
+			} else {
+				assert.NoError(err2)
+				assert.Equal(uint32(1), got.Version)
+			}
+		})
+	}
+}

internal/session/service_worker_status_report.go

@@ -42,7 +42,7 @@ func WorkerStatusReport(ctx context.Context, repo *Repository, connRepo *Connect
 		return notActive, errors.New(ctx, errors.Internal, op, fmt.Sprintf("Error closing orphaned connections for worker %s: %v", workerId, err))
 	}
 	if len(closed) > 0 {
-		event.WriteSysEvent(ctx, op, "marked unclaimed connections as closed", "server_id", workerId, "count", len(closed))
+		event.WriteSysEvent(ctx, op, "marked unclaimed connections as closed", "controller_id", workerId, "count", len(closed))
 	}
 	return notActive, err
 }

internal/session/service_worker_status_report_test.go

@@ -5,6 +5,7 @@ import (
 	"testing"
 
 	"github.com/hashicorp/boundary/internal/servers/store"
+	"github.com/hashicorp/boundary/internal/types/scope"
 
 	"github.com/hashicorp/boundary/internal/authtoken"
 	"github.com/hashicorp/boundary/internal/db"
@@ -32,10 +33,9 @@ func TestWorkerStatusReport(t *testing.T) {
 		PrivateId: "test_controller1",
 		Address:   "127.0.0.1",
 	})
-	serverRepo.UpsertWorker(ctx, &store.Worker{
-		PublicId: "test_worker1",
-		Address:  "127.0.0.1",
-	})
+	serverRepo.UpsertWorker(ctx, servers.NewWorker(scope.Global.String(),
+		servers.WithPublicId("test_worker1"),
+		servers.WithAddress("127.0.0.1")))
 
 	repo, err := session.NewRepository(rw, rw, kms)
 	require.NoError(t, err)
@@ -56,7 +56,7 @@ func TestWorkerStatusReport(t *testing.T) {
 	)
 
 	type testCase struct {
-		worker              *store.Worker
+		worker              *servers.Worker
 		req                 []session.StateReport
 		want                []session.StateReport
 		orphanedConnections []string

internal/session/session_cleanup_job_test.go

@@ -12,7 +12,6 @@ import (
 	"github.com/hashicorp/boundary/internal/kms"
 	"github.com/hashicorp/boundary/internal/scheduler"
 	"github.com/hashicorp/boundary/internal/servers"
-	"github.com/hashicorp/boundary/internal/servers/store"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -289,7 +288,7 @@ func TestCloseConnectionsForDeadWorkers(t *testing.T) {
 	// updateServer is a helper for updating the update time for our
 	// servers. The controller is read back so that we can reference
 	// the most up-to-date fields.
-	updateServer := func(t *testing.T, w *store.Worker) *store.Worker {
+	updateServer := func(t *testing.T, w *servers.Worker) *servers.Worker {
 		t.Helper()
 		_, rowsUpdated, err := serversRepo.UpsertWorker(ctx, w)
 		require.NoError(err)

internal/session/testing.go

@@ -6,7 +6,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/hashicorp/boundary/internal/servers/store"
+	"github.com/hashicorp/boundary/internal/types/scope"
 
 	"github.com/hashicorp/boundary/internal/auth/password"
 	"github.com/hashicorp/boundary/internal/authtoken"
@@ -181,7 +181,7 @@ func TestTofu(t testing.TB) []byte {
 }
 
 // TestWorker inserts a worker into the db to satisfy foreign key constraints.
-func TestWorker(t *testing.T, conn *db.DB, wrapper wrapping.Wrapper, opt ...Option) *store.Worker {
+func TestWorker(t *testing.T, conn *db.DB, wrapper wrapping.Wrapper, opt ...Option) *servers.Worker {
 	t.Helper()
 	rw := db.New(conn)
 	kms := kms.TestKms(t, conn, wrapper)
@@ -193,11 +193,10 @@ func TestWorker(t *testing.T, conn *db.DB, wrapper wrapping.Wrapper, opt ...Opti
 	id = "test_session_worker-" + id
 
 	name := "test-worker-" + id
-	worker := &store.Worker{
-		PublicId: id,
-		Name:     name,
-		Address:  "127.0.0.1",
-	}
+	worker := servers.NewWorker(scope.Global.String(),
+		servers.WithPublicId(id),
+		servers.WithName(name),
+		servers.WithAddress("127.0.0.1"))
 	_, _, err = serversRepo.UpsertWorker(context.Background(), worker)
 	require.NoError(t, err)
 	return worker

testing/dbtest/session_list_benchmarks_dump_generation_test.go

@@ -19,10 +19,10 @@ import (
 	"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/hashicorp/boundary/internal/types/scope"
 	"github.com/hashicorp/boundary/testing/dbtest"
 	"github.com/stretchr/testify/require"
 	"golang.org/x/sync/errgroup"
@@ -132,10 +132,9 @@ func TestGenerateSessionBenchmarkTemplateDumps(t *testing.T) {
 			require.NoError(err)
 			serversRepo, err := servers.NewRepository(rw, rw, kms)
 			require.NoError(err)
-			worker := &store.Worker{
-				PublicId: "test_worker_1",
-				Address:  "127.0.0.1",
-			}
+			worker := servers.NewWorker(scope.Global.String(),
+				servers.WithPublicId("test_worker_1"),
+				servers.WithAddress("127.0.0.1"))
 			_, _, err = serversRepo.UpsertWorker(ctx, worker)
 			require.NoError(err)
 
@@ -196,7 +195,7 @@ func TestGenerateSessionBenchmarkTemplateDumps(t *testing.T) {
 						ScopeId:     users[userIndex].scopeId,
 						Endpoint:    "tcp://127.0.0.1:22",
 					})
-					cycleSessionStates(t, ctx, sess, sessRepo, connRepo, conn, worker, scenario.connsPerSession)
+					cycleSessionStates(t, ctx, sess, sessRepo, connRepo, conn, scenario.connsPerSession)
 					return nil
 				})
 			}
@@ -220,7 +219,7 @@ func TestGenerateSessionBenchmarkTemplateDumps(t *testing.T) {
 	}
 }
 
-func cycleSessionStates(t testing.TB, ctx context.Context, sess *session.Session, sessRepo *session.Repository, connRepo *session.ConnectionRepository, conn *db.DB, worker *store.Worker, numConns int) {
+func cycleSessionStates(t testing.TB, ctx context.Context, sess *session.Session, sessRepo *session.Repository, connRepo *session.ConnectionRepository, conn *db.DB, numConns int) {
 	sess, _, err := sessRepo.ActivateSession(ctx, sess.PublicId, sess.Version, []byte(`tofu`))
 	require.NoError(t, err)
 	var closeWiths []session.CloseWith