Add worker config table and view (#2090)

* 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 constraints, 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.

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

@@ -27,6 +27,7 @@ import (
 	"github.com/hashicorp/boundary/internal/kms"
 	"github.com/hashicorp/boundary/internal/perms"
 	"github.com/hashicorp/boundary/internal/requests"
+	"github.com/hashicorp/boundary/internal/servers"
 	"github.com/hashicorp/boundary/internal/session"
 	"github.com/hashicorp/boundary/internal/target"
 	"github.com/hashicorp/boundary/internal/types/action"
@@ -894,79 +895,27 @@ func (s Service) AuthorizeSession(ctx context.Context, req *pbs.AuthorizeSession
 	// worker IDs below is used to contain their IDs in the same order. This is
 	// used to fetch tags for filtering. But we avoid allocation unless we
 	// actually need it.
-	var workers []*pb.WorkerInfo
-	var workerIds []string
-	hasWorkerFilter := len(t.GetWorkerFilter()) > 0
-	servers, err := serversRepo.ListWorkers(ctx)
+	selectedWorkers, err := serversRepo.ListWorkers(ctx)
 	if err != nil {
 		return nil, err
 	}
-	for _, v := range servers {
-		if hasWorkerFilter {
-			workerIds = append(workerIds, v.GetPublicId())
-		}
-		workers = append(workers, &pb.WorkerInfo{Address: v.Address})
-	}
 
-	if hasWorkerFilter && len(workerIds) > 0 {
-		finalWorkers := make([]*pb.WorkerInfo, 0, len(workers))
-		// Fetch the tags for the given worker IDs
-		workerTags, err := serversRepo.ListTagsForWorkers(ctx, workerIds)
-		if err != nil {
-			return nil, err
-		}
-		// Build the map for filtering. This is similar to the filter map we
-		// 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 wId, tags := range workerTags {
-			currWorkerMap := tagMap[wId]
-			if currWorkerMap == nil {
-				currWorkerMap = make(map[string][]string)
-				tagMap[wId] = currWorkerMap
-			}
-			for _, tag := range tags {
-				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
-		}
-
-		// Create the evaluator
+	if len(t.GetWorkerFilter()) > 0 && len(selectedWorkers) > 0 {
 		eval, err := bexpr.CreateEvaluator(t.GetWorkerFilter())
 		if err != nil {
 			return nil, err
 		}
-
-		// Iterate through the known worker IDs, and evaluate. If evaluation
-		// returns true, add to the final worker slice, which is assigned back
-		// to workers after this.
-		for i, worker := range workerIds {
-			filterInput := map[string]interface{}{
-				"name": worker,
-				"tags": tagMap[worker],
-			}
-			ok, err := eval.Evaluate(filterInput)
-			if err != nil && !stderrors.Is(err, pointerstructure.ErrNotFound) {
-				return nil, handlers.ApiErrorWithCodeAndMessage(
-					codes.FailedPrecondition,
-					fmt.Sprintf("Worker filter expression evaluation resulted in error: %s", err))
-			}
-			if ok {
-				finalWorkers = append(finalWorkers, workers[i])
-			}
+		selectedWorkers, err = workerList(selectedWorkers).filtered(eval)
+		if err != nil {
+			return nil, err
 		}
-		workers = finalWorkers
 	}
-	if len(workers) == 0 {
+
+	if len(selectedWorkers) == 0 {
 		return nil, handlers.ApiErrorWithCodeAndMessage(
 			codes.FailedPrecondition,
 			"No workers are available to handle this session, or all have been filtered.")
 	}
-	workerAddresses := make([]string, 0, len(workers))
-	for _, worker := range workers {
-		workerAddresses = append(workerAddresses, worker.GetAddress())
-	}
 
 	requestedId := req.GetHostId()
 	staticHostRepo, err := s.staticHostRepoFn()
@@ -1074,7 +1023,7 @@ func (s Service) AuthorizeSession(ctx context.Context, req *pbs.AuthorizeSession
 	if err != nil {
 		return nil, err
 	}
-	sess, privKey, err := sessionRepo.CreateSession(ctx, wrapper, sess, workerAddresses)
+	sess, privKey, err := sessionRepo.CreateSession(ctx, wrapper, sess, workerList(selectedWorkers).addresses())
 	if err != nil {
 		return nil, err
 	}
@@ -1202,7 +1151,7 @@ func (s Service) AuthorizeSession(ctx context.Context, req *pbs.AuthorizeSession
 		PrivateKey:      privKey,
 		HostId:          chosenEndpoint.HostId,
 		Endpoint:        endpointUrl.String(),
-		WorkerInfo:      workers,
+		WorkerInfo:      workerList(selectedWorkers).workerInfos(),
 		ConnectionLimit: t.GetSessionConnectionLimit(),
 	}
 	marshaledSad, err := proto.Marshal(sad)
@@ -2172,3 +2121,53 @@ func credentialToProto(ctx context.Context, cred credential.Credential) (*server
 		return nil, errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("unsupported credential %T", c))
 	}
 }
+
+// workerList is a helper type to make the selection of workers clearer and more declarative.
+type workerList []*servers.Worker
+
+// addresses converts the slice of workers to a slice of their addresses
+func (w workerList) addresses() []string {
+	ret := make([]string, 0, len(w))
+	for _, worker := range w {
+		ret = append(ret, worker.CanonicalAddress())
+	}
+	return ret
+}
+
+// workerInfos converts the slice of workers to a slice of their workerInfo protos
+func (w workerList) workerInfos() []*pb.WorkerInfo {
+	ret := make([]*pb.WorkerInfo, 0, len(w))
+	for _, worker := range w {
+		ret = append(ret, &pb.WorkerInfo{Address: worker.CanonicalAddress()})
+	}
+	return ret
+}
+
+// filtered returns a new workerList where all elements contained in it are the
+// ones which from the original workerList that pass the evaluator's evaluation.
+func (w workerList) filtered(eval *bexpr.Evaluator) (workerList, error) {
+	var ret []*servers.Worker
+	for _, worker := range w {
+
+		// TODO: Decide how we are actually going to filter by name now that
+		//   we are separating reported status name and the resource name.
+		name := worker.GetName()
+		if name == "" {
+			name = worker.ReportedStatus.GetName()
+		}
+		filterInput := map[string]interface{}{
+			"name": name,
+			"tags": worker.CanonicalTags(),
+		}
+		ok, err := eval.Evaluate(filterInput)
+		if err != nil && !stderrors.Is(err, pointerstructure.ErrNotFound) {
+			return nil, handlers.ApiErrorWithCodeAndMessage(
+				codes.FailedPrecondition,
+				fmt.Sprintf("Worker filter expression evaluation resulted in error: %s", err))
+		}
+		if ok {
+			ret = append(ret, worker)
+		}
+	}
+	return ret, nil
+}

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

@@ -0,0 +1,74 @@
+package targets
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/hashicorp/boundary/internal/servers"
+	"github.com/hashicorp/boundary/internal/types/scope"
+	pb "github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/targets"
+	"github.com/hashicorp/go-bexpr"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestWorkerList_Addresses(t *testing.T) {
+	addresses := []string{
+		"test1",
+		"test2",
+		"test3",
+		"test4",
+	}
+	var workerInfos []*pb.WorkerInfo
+	var tested workerList
+	for _, a := range addresses {
+		workerInfos = append(workerInfos, &pb.WorkerInfo{Address: a})
+		tested = append(tested, servers.NewWorker(scope.Global.String(),
+			servers.WithName(a),
+			servers.WithAddress(a)))
+	}
+	assert.Equal(t, addresses, tested.addresses())
+	assert.Equal(t, workerInfos, tested.workerInfos())
+}
+
+func TestWorkerList_Filter(t *testing.T) {
+	var workers []*servers.Worker
+	for i := 0; i < 5; i++ {
+		w := servers.NewWorker(scope.Global.String(),
+			servers.WithName(fmt.Sprintf("test_worker_%d", i)),
+			servers.WithWorkerTags(&servers.Tag{
+				Key:   fmt.Sprintf("key%d", i),
+				Value: fmt.Sprintf("value%d", i),
+			}))
+		w.ReportedStatus = servers.NewWorkerStatus("",
+			servers.WithName(fmt.Sprintf("config%d", i)),
+			servers.WithWorkerTags(&servers.Tag{
+				Key: "key",
+				// Note that the value is either 0 or 1
+				Value: fmt.Sprintf("configvalue%d", i%2),
+			}))
+		workers = append(workers, w)
+	}
+
+	{
+		ev, err := bexpr.CreateEvaluator(`"/name" matches "test_worker_[13]" and "configvalue2" in "/tags/key"`)
+		require.NoError(t, err)
+		got, err := workerList(workers).filtered(ev)
+		require.NoError(t, err)
+		assert.Len(t, got, 0)
+	}
+	{
+		ev, err := bexpr.CreateEvaluator(`"/name" matches "test_worker_[12]" and "configvalue0" in "/tags/key"`)
+		require.NoError(t, err)
+		got, err := workerList(workers).filtered(ev)
+		require.NoError(t, err)
+		assert.Len(t, got, 1)
+	}
+	{
+		ev, err := bexpr.CreateEvaluator(`"configvalue0" in "/tags/key"`)
+		require.NoError(t, err)
+		got, err := workerList(workers).filtered(ev)
+		require.NoError(t, err)
+		assert.Len(t, got, 3)
+	}
+}

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

@@ -14,7 +14,6 @@ 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"
@@ -83,11 +82,12 @@ func (ws *workerServiceServer) Status(ctx context.Context, req *pbs.StatusReques
 		}
 	}
 
-	worker := servers.NewWorker(scope.Global.String(),
-		servers.WithPublicId(reqServer.PrivateId),
+	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.UpsertWorker(ctx, worker, servers.WithUpdateTags(req.GetUpdateTags()))
+	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)

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

@@ -5,9 +5,6 @@ import (
 	"sync"
 	"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"
 	"github.com/hashicorp/boundary/internal/authtoken"
@@ -18,6 +15,7 @@ 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"
@@ -38,9 +36,8 @@ func TestStatus(t *testing.T) {
 		PrivateId: "test_controller1",
 		Address:   "127.0.0.1",
 	})
-	serverRepo.UpsertWorker(ctx, servers.NewWorker(
-		scope.Global.String(),
-		servers.WithPublicId("test_worker1"),
+	serverRepo.UpsertWorkerStatus(ctx, servers.NewWorkerStatus(
+		"test_worker1",
 		servers.WithAddress("127.0.0.1")))
 
 	serversRepoFn := func() (*servers.Repository, error) {
@@ -107,7 +104,7 @@ func TestStatus(t *testing.T) {
 			req: &pbs.StatusRequest{
 				Worker: &servers.Server{
 					PrivateId:  worker1.PublicId,
-					Address:    worker1.Address,
+					Address:    worker1.CanonicalAddress(),
 					CreateTime: worker1.CreateTime,
 					UpdateTime: worker1.UpdateTime,
 				},
@@ -127,7 +124,7 @@ func TestStatus(t *testing.T) {
 			req: &pbs.StatusRequest{
 				Worker: &servers.Server{
 					PrivateId:  worker1.PublicId,
-					Address:    worker1.Address,
+					Address:    worker1.CanonicalAddress(),
 					CreateTime: worker1.CreateTime,
 					UpdateTime: worker1.UpdateTime,
 				},
@@ -206,9 +203,8 @@ func TestStatusSessionClosed(t *testing.T) {
 		PrivateId: "test_controller1",
 		Address:   "127.0.0.1",
 	})
-	serverRepo.UpsertWorker(ctx, servers.NewWorker(
-		scope.Global.String(),
-		servers.WithPublicId("test_worker1"),
+	serverRepo.UpsertWorkerStatus(ctx, servers.NewWorkerStatus(
+		"test_worker1",
 		servers.WithAddress("127.0.0.1")))
 
 	serversRepoFn := func() (*servers.Repository, error) {
@@ -292,7 +288,7 @@ func TestStatusSessionClosed(t *testing.T) {
 			req: &pbs.StatusRequest{
 				Worker: &servers.Server{
 					PrivateId:  worker1.PublicId,
-					Address:    worker1.Address,
+					Address:    worker1.CanonicalAddress(),
 					CreateTime: worker1.CreateTime,
 					UpdateTime: worker1.UpdateTime,
 				},
@@ -388,9 +384,8 @@ func TestStatusDeadConnection(t *testing.T) {
 		PrivateId: "test_controller1",
 		Address:   "127.0.0.1",
 	})
-	serverRepo.UpsertWorker(ctx, servers.NewWorker(
-		scope.Global.String(),
-		servers.WithPublicId("test_worker1"),
+	serverRepo.UpsertWorkerStatus(ctx, servers.NewWorkerStatus(
+		"test_worker1",
 		servers.WithAddress("127.0.0.1")))
 
 	serversRepoFn := func() (*servers.Repository, error) {
@@ -462,7 +457,7 @@ func TestStatusDeadConnection(t *testing.T) {
 	req := &pbs.StatusRequest{
 		Worker: &servers.Server{
 			PrivateId:  worker1.PublicId,
-			Address:    worker1.Address,
+			Address:    worker1.CanonicalAddress(),
 			CreateTime: worker1.CreateTime,
 			UpdateTime: worker1.UpdateTime,
 		},

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

@@ -58,6 +58,33 @@ create trigger worker_insert_time_column before insert on server_worker
 create trigger worker_update_time_column before update on server_worker
   for each row execute procedure update_time_column();
 
+create table server_worker_status (
+  worker_id wt_public_id primary key
+    constraint server_worker_fkey
+      references server_worker(public_id)
+      on delete cascade
+      on update cascade,
+  create_time wt_timestamp,
+  update_time wt_timestamp,
+  -- This is the calculated address that the worker reports it is reachable on.
+  address wt_network_address not null,
+  name wt_name
+);
+comment on table server_worker_status  is
+  'server_worker_status is a table where each row represents values that a Boundary worker reports to a controller.';
+
+create trigger immutable_columns before update on server_worker_status
+  for each row execute procedure immutable_columns('worker_id', 'create_time');
+
+create trigger default_create_time_column before insert on server_worker_status
+  for each row execute procedure default_create_time();
+
+create trigger worker_insert_time_column before insert on server_worker_status
+  for each row execute procedure update_time_column();
+
+create trigger worker_update_time_column before update on server_worker_status
+  for each row execute procedure update_time_column();
+
 -- Create table worker tag
 create table server_worker_tag_enm (
   source text primary key
@@ -88,6 +115,45 @@ create table server_worker_tag (
   primary key(worker_id, key, value, source)
 );
 
+-- worker_aggregate view allows the worker and configuration to be read at the
+-- same time.
+create view server_worker_aggregate as
+  with worker_config_tags(worker_id, source, tags) as (
+    select
+      ct.worker_id,
+      ct.source,
+      -- keys and tags can be any lowercase printable character so use uppercase characters as delimitors.
+      string_agg(distinct concat_ws('Y', ct.key, ct.value), 'Z') as tags
+    from server_worker_tag ct
+    group by ct.worker_id, ct.source
+  )
+select
+  w.public_id,
+  w.scope_id,
+  w.description,
+  w.name,
+  w.address,
+  w.create_time,
+  w.update_time,
+  w.version,
+  ws.name as worker_status_name,
+  ws.address as worker_status_address,
+  ws.update_time as worker_status_update_time,
+  ws.create_time as worker_status_create_time,
+  -- keys and tags can be any lowercase printable character so use uppercase characters as delimitors.
+  wt.tags as api_tags,
+  ct.tags as worker_config_tags
+from server_worker w
+  left join server_worker_status ws on
+    w.public_id = ws.worker_id
+  left join worker_config_tags wt on
+      w.public_id = wt.worker_id and wt.source = 'api'
+  left join worker_config_tags ct on
+    w.public_id = ct.worker_id and ct.source = 'configuration';
+comment on view server_worker_aggregate is
+'server_worker_aggregate contains the worker resource with its worker provided config values and its configuration and api provided tags.';
+
+
 -- Aaand drop server_tag
 drop table server_tag;
 

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

@@ -20,7 +20,7 @@ message Worker {
   // @inject_tag: `gorm:"default:null"`
   string description = 30;
 
-  // Address for the worker. This must be set
+  // Address for the worker. This is optional.
   // @inject_tag: `gorm:"default:null"`
   string address = 40;
 
@@ -45,18 +45,42 @@ message Worker {
 // 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"`
+  // @inject_tag: `gorm:"primary_key"`
   string worker_id = 10;
 
   // key is the key of the tag. This must be set.
-  // @inject_tag: `gorm:"default:null"`
+  // @inject_tag: `gorm:"primary_key"`
   string key = 20;
 
   // value is the value
-  // @inject_tag: `gorm:"default:null"`
+  // @inject_tag: `gorm:"primary_key"`
   string value = 30;
 
   // source is the source of the tag. Either 'configuration' or 'api'.
   // @inject_tag: `gorm:"default:not_null"`
   string source = 40;
 }
+
+// WorkerStatus contains the fields that the worker reports to the controller
+// in its status  updates.
+message WorkerStatus {
+  // worker_id is the public key that key of the worker this tag is for.
+  // @inject_tag: `gorm:"primary_key"`
+  string worker_id = 10;
+
+  // The create_time is set by the database.
+  // @inject_tag: `gorm:"default:current_timestamp"`
+  timestamp.v1.Timestamp create_time = 20;
+
+  // The update_time is set by the database.
+  // @inject_tag: `gorm:"default:current_timestamp"`
+  timestamp.v1.Timestamp update_time = 30;
+
+  // Name of the resource (optional)
+  // @inject_tag: `gorm:"default:null"`
+  string name = 40;
+
+  // Address for the worker. This must be set.
+  // @inject_tag: `gorm:"default:null"`
+  string address = 50;
+}

internal/servers/query.go

@@ -2,5 +2,11 @@ package servers
 
 const (
 	deleteWhereCreateTimeSql = `create_time < ?`
-	deleteConfigTagsSql      = `source = 'configuration' and worker_id = ?`
+	deleteTagsByWorkerIdSql  = `
+	delete 
+	from server_worker_tag 
+	where 
+		source = ?
+	and
+		worker_id = ?`
 )

internal/servers/repository.go

@@ -1,6 +1,7 @@
 package servers
 
 import (
+	"reflect"
 	"time"
 
 	"github.com/hashicorp/boundary/internal/db"
@@ -46,3 +47,14 @@ func NewRepository(r db.Reader, w db.Writer, kms *kms.Kms) (*Repository, error)
 		kms:    kms,
 	}, nil
 }
+
+func isNil(i interface{}) bool {
+	if i == nil {
+		return true
+	}
+	switch reflect.TypeOf(i).Kind() {
+	case reflect.Ptr, reflect.Map, reflect.Array, reflect.Chan, reflect.Slice:
+		return reflect.ValueOf(i).IsNil()
+	}
+	return false
+}

internal/servers/repository_worker.go

@@ -7,6 +7,7 @@ import (
 	"github.com/hashicorp/boundary/internal/db"
 	"github.com/hashicorp/boundary/internal/errors"
 	"github.com/hashicorp/boundary/internal/servers/store"
+	"github.com/hashicorp/boundary/internal/types/scope"
 )
 
 // ListWorkers is a passthrough to listWorkersWithReader that uses the repo's normal reader.
@@ -14,9 +15,16 @@ func (r *Repository) ListWorkers(ctx context.Context, opt ...Option) ([]*Worker,
 	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.
+// listWorkersWithReader will return a listing of resources and honor the
+// WithLimit option. If WithLiveness is zero the default liveness value is used,
+// if it is negative then the last status update time is ignored. This method
+// 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) ([]*Worker, error) {
+	const op = "workers.listWorkersWithReader"
+	switch {
+	case isNil(reader):
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "reader is nil")
+	}
 	opts := getOpts(opt...)
 	liveness := opts.withLiveness
 	if liveness == 0 {
@@ -25,20 +33,28 @@ func (r *Repository) listWorkersWithReader(ctx context.Context, reader db.Reader
 
 	var where string
 	if liveness > 0 {
-		where = fmt.Sprintf("update_time > now() - interval '%d seconds'", uint32(liveness.Seconds()))
+		where = fmt.Sprintf("worker_status_update_time > now() - interval '%d seconds'", uint32(liveness.Seconds()))
 	}
 
-	var workers []*Worker
+	var wAggs []*workerAggregate
 	if err := reader.SearchWhere(
 		ctx,
-		&workers,
+		&wAggs,
 		where,
 		[]interface{}{},
-		db.WithLimit(-1),
+		db.WithLimit(opts.withLimit),
 	); err != nil {
-		return nil, errors.Wrap(ctx, err, "workers.listWorkersWithReader")
+		return nil, errors.Wrap(ctx, err, op, errors.WithMsg("error searching for workers"))
 	}
 
+	workers := make([]*Worker, 0, len(wAggs))
+	for _, a := range wAggs {
+		w, err := a.toWorker(ctx)
+		if err != nil {
+			return nil, errors.Wrap(ctx, err, op, errors.WithMsg("error converting workerAggregate to Worker"))
+		}
+		workers = append(workers, w)
+	}
 	return workers, nil
 }
 
@@ -64,21 +80,20 @@ func (r *Repository) ListTagsForWorkers(ctx context.Context, workerIds []string,
 	return ret, nil
 }
 
-// UpsertWorker creates a new worker if one with the provided public id doesn't
-// already exist. If it does, UpsertWorker updates the worker. The
+// UpsertWorkerStatus creates a new worker if one with the provided public id doesn't
+// already exist. If it does, UpsertWorkerStatus 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"
+// Workers are intentionally not oplogged.
+func (r *Repository) UpsertWorkerStatus(ctx context.Context, wStatus *WorkerStatus, opt ...Option) ([]*store.Controller, int, error) {
+	const op = "servers.UpsertWorkerStatus"
 
 	switch {
-	case worker == nil:
-		return nil, db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "worker is nil")
-	case worker.Address == "":
+	case wStatus == nil:
+		return nil, db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "wStatus is nil")
+	case wStatus.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")
+	case wStatus.WorkerId == "":
+		return nil, db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "worker id is empty")
 	}
 
 	opts := getOpts(opt...)
@@ -89,19 +104,28 @@ func (r *Repository) UpsertWorker(ctx context.Context, worker *Worker, opt ...Op
 		ctx,
 		db.StdRetryCnt,
 		db.ExpBackoff{},
-		func(read db.Reader, w db.Writer) error {
-			var err error
-			onConflict := &db.OnConflict{
+		func(reader db.Reader, w db.Writer) error {
+			worker := NewWorker(scope.Global.String(), WithPublicId(wStatus.GetWorkerId()))
+			workerCreateConflict := &db.OnConflict{
 				Target: db.Columns{"public_id"},
-				Action: append(db.SetColumns([]string{"description", "address"}), db.SetColumnValues(map[string]interface{}{"update_time": "now()"})...),
+				Action: db.DoNothing(true),
 			}
-			err = w.Create(ctx, worker, db.WithOnConflict(onConflict), db.WithReturnRowsAffected(&rowsUpdated))
+			if err := w.Create(ctx, worker, db.WithOnConflict(workerCreateConflict)); err != nil {
+				return errors.Wrap(ctx, err, op, errors.WithMsg("error creating a worker"))
+			}
+
+			var err error
+			onConfigConflict := &db.OnConflict{
+				Target: db.Columns{"worker_id"},
+				Action: append(db.SetColumns([]string{"name", "address"}), db.SetColumnValues(map[string]interface{}{"update_time": "now()"})...),
+			}
+			err = w.Create(ctx, wStatus, db.WithOnConflict(onConfigConflict), db.WithReturnRowsAffected(&rowsUpdated))
 			if err != nil {
 				return errors.Wrap(ctx, err, op+":Upsert")
 			}
 
 			// Fetch current controllers to feed to the workers
-			controllers, err = r.listControllersWithReader(ctx, read)
+			controllers, err = r.listControllersWithReader(ctx, reader)
 			if err != nil {
 				return errors.Wrap(ctx, err, op+":ListController")
 			}
@@ -111,32 +135,7 @@ func (r *Repository) UpsertWorker(ctx context.Context, worker *Worker, opt ...Op
 			// delete all tags for the given worker, then add the new ones
 			// we've been sent.
 			if opts.withUpdateTags {
-				_, err = w.Delete(ctx, &store.WorkerTag{}, db.WithWhere(deleteConfigTagsSql, worker.GetPublicId()))
-				if err != nil {
-					return errors.Wrap(ctx, err, op+":DeleteTags", errors.WithMsg(worker.GetPublicId()))
-				}
-
-				// If tags were cleared out entirely, then we'll have nothing
-				// to do here, e.g., it will result in deletion of all tags.
-				// Otherwise, go through and stage each tuple for insertion
-				// below.
-				if len(worker.Tags) > 0 {
-					tags := make([]interface{}, 0, len(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", worker.GetPublicId()))
-						}
-						tags = append(tags, &store.WorkerTag{
-							WorkerId: worker.GetPublicId(),
-							Key:      v.Key,
-							Value:    v.Value,
-							Source:   string(ConfigurationTagSource),
-						})
-					}
-					if err = w.CreateItems(ctx, tags); err != nil {
-						return errors.Wrap(ctx, err, op+":CreateTags", errors.WithMsg(worker.GetPublicId()))
-					}
-				}
+				setWorkerTags(ctx, w, wStatus.GetWorkerId(), ConfigurationTagSource, wStatus.Tags)
 			}
 
 			return nil
@@ -148,3 +147,47 @@ func (r *Repository) UpsertWorker(ctx context.Context, worker *Worker, opt ...Op
 
 	return controllers, int(rowsUpdated), nil
 }
+
+// setWorkerTags removes all existing tags from the same source and worker id
+// and creates new ones based on the ones provided.  This function should be
+// called from inside a db transaction.
+// Workers/worker tags are intentionally not oplogged.
+func setWorkerTags(ctx context.Context, w db.Writer, id string, ts TagSource, tags []*Tag) error {
+	const op = "servers.setWorkerTags"
+	switch {
+	case !ts.isValid():
+		return errors.New(ctx, errors.InvalidParameter, op, "invalid tag source provided")
+	case id == "":
+		return errors.New(ctx, errors.InvalidParameter, op, "worker id is empty")
+	case isNil(w):
+		return errors.New(ctx, errors.InvalidParameter, op, "db.Writer is nil")
+	}
+	_, err := w.Exec(ctx, deleteTagsByWorkerIdSql, []interface{}{ts.String(), id})
+	if err != nil {
+		return errors.Wrap(ctx, err, op, errors.WithMsg(fmt.Sprintf("couldn't delete exist tags for worker %q", id)))
+	}
+
+	// If tags were cleared out entirely, then we'll have nothing
+	// to do here, e.g., it will result in deletion of all tags.
+	// Otherwise, go through and stage each tuple for insertion
+	// below.
+	if len(tags) > 0 {
+		uTags := make([]interface{}, 0, len(tags))
+		for _, v := range tags {
+			if v == nil {
+				return errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("found nil tag value for worker %s", id))
+			}
+			uTags = append(uTags, &store.WorkerTag{
+				WorkerId: id,
+				Key:      v.Key,
+				Value:    v.Value,
+				Source:   ts.String(),
+			})
+		}
+		if err = w.CreateItems(ctx, uTags); err != nil {
+			return errors.Wrap(ctx, err, op, errors.WithMsg(fmt.Sprintf("error creating tags for worker %q", id)))
+		}
+	}
+
+	return nil
+}

internal/servers/repository_worker_test.go

@@ -2,29 +2,153 @@ package servers_test
 
 import (
 	"context"
+	"fmt"
 	"testing"
 	"time"
 
-	"github.com/hashicorp/boundary/internal/daemon/controller"
 	"github.com/hashicorp/boundary/internal/db"
+	"github.com/hashicorp/boundary/internal/errors"
 	"github.com/hashicorp/boundary/internal/kms"
 	"github.com/hashicorp/boundary/internal/servers"
-	"github.com/hashicorp/boundary/internal/types/scope"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
+func TestUpsertWorkerStatus(t *testing.T) {
+	conn, _ := db.TestSetup(t, "postgres")
+	rw := db.New(conn)
+	wrapper := db.TestWrapper(t)
+	kms := kms.TestKms(t, conn, wrapper)
+	repo, err := servers.NewRepository(rw, rw, kms)
+	require.NoError(t, err)
+	ctx := context.Background()
+
+	wStatus1 := servers.NewWorkerStatus("test_worker_1",
+		servers.WithAddress("address"),
+		servers.WithName("config_name1"))
+	_, _, err = repo.UpsertWorkerStatus(ctx, wStatus1)
+	require.NoError(t, err)
+	{
+		workers, err := repo.ListWorkers(ctx)
+		require.NoError(t, err)
+		assert.Len(t, workers, 1)
+		worker := workers[0]
+		assert.Equal(t, wStatus1.GetWorkerId(), worker.GetPublicId())
+		assert.Equal(t, wStatus1.Address, worker.CanonicalAddress())
+		assert.Empty(t, worker.Name)
+		assert.Equal(t, worker.ReportedStatus.CreateTime, worker.CreateTime)
+		assert.Equal(t, worker.LastConnectionUpdate(), worker.UpdateTime)
+		assert.Equal(t, uint32(1), worker.Version)
+		assert.Empty(t, worker.Address)
+		assert.Empty(t, worker.Description)
+	}
+
+	// Remove the status name, verify that it doesn't remove the resource's name.
+	updatedStatus1 := servers.NewWorkerStatus(wStatus1.WorkerId,
+		servers.WithAddress(wStatus1.Address))
+	_, _, err = repo.UpsertWorkerStatus(ctx, updatedStatus1)
+	require.NoError(t, err)
+	{
+		workers, err := repo.ListWorkers(ctx)
+		require.NoError(t, err)
+		assert.Len(t, workers, 1)
+		worker := workers[0]
+		// The resource's name remains untouched
+		assert.Empty(t, worker.Name)
+		// The ReportedStatus's name is correctly cleared
+		assert.Empty(t, worker.ReportedStatus.Name)
+		// The last connection update time was increased
+		assert.Greater(t, worker.LastConnectionUpdate().AsTime().UnixMilli(), worker.UpdateTime.AsTime().UnixMilli())
+		assert.Equal(t, worker.UpdateTime, worker.CreateTime)
+
+		assert.Equal(t, updatedStatus1.GetWorkerId(), worker.GetPublicId())
+		assert.Equal(t, updatedStatus1.Address, worker.CanonicalAddress())
+		assert.Equal(t, worker.ReportedStatus.CreateTime, worker.CreateTime)
+		assert.Equal(t, uint32(1), worker.Version)
+		assert.Empty(t, worker.Address)
+		assert.Empty(t, worker.Description)
+	}
+
+	failureCases := []struct {
+		name      string
+		status    *servers.WorkerStatus
+		errAssert func(*testing.T, error)
+	}{
+		{
+			name: "no address",
+			status: servers.NewWorkerStatus("worker_with_no_address",
+				servers.WithName("worker_with_no_address")),
+			errAssert: func(t *testing.T, err error) {
+				t.Helper()
+				assert.True(t, errors.Match(errors.T(errors.InvalidParameter), err))
+			},
+		},
+		{
+			name: "invalid id",
+			status: servers.NewWorkerStatus("short",
+				servers.WithAddress("someaddress"),
+				servers.WithName("worker_with_short_id")),
+			errAssert: func(t *testing.T, err error) {
+				t.Helper()
+				assert.True(t, errors.IsCheckConstraintError(err))
+			},
+		},
+		{
+			name: "no id",
+			status: servers.NewWorkerStatus("",
+				servers.WithAddress("anotheraddress"),
+				servers.WithName("no_address")),
+			errAssert: func(t *testing.T, err error) {
+				t.Helper()
+				assert.True(t, errors.Match(errors.T(errors.InvalidParameter), err))
+			},
+		},
+		{
+			name:   "no status",
+			status: nil,
+			errAssert: func(t *testing.T, err error) {
+				t.Helper()
+				assert.True(t, errors.Match(errors.T(errors.InvalidParameter), err))
+			},
+		},
+	}
+	for _, tc := range failureCases {
+		t.Run(fmt.Sprintf("Failures %s", tc.name), func(t *testing.T) {
+			_, _, err = repo.UpsertWorkerStatus(ctx, tc.status)
+			assert.Error(t, err)
+			tc.errAssert(t, err)
+
+			// Still only the original worker exists.
+			workers, err := repo.ListWorkers(ctx)
+			require.NoError(t, err)
+			assert.Len(t, workers, 1)
+		})
+	}
+
+	{
+		anotherStatus := servers.NewWorkerStatus("another_test_worker",
+			servers.WithAddress("address"))
+		_, _, err = repo.UpsertWorkerStatus(ctx, anotherStatus)
+		require.NoError(t, err)
+		{
+			workers, err := repo.ListWorkers(ctx)
+			require.NoError(t, err)
+			assert.Len(t, workers, 2)
+		}
+	}
+}
+
 func TestTagUpdatingListing(t *testing.T) {
 	require := require.New(t)
-
+	conn, _ := db.TestSetup(t, "postgres")
+	rw := db.New(conn)
 	wrapper := db.TestWrapper(t)
-	tc := controller.NewTestController(t, &controller.TestControllerOpts{
-		RecoveryKms: wrapper,
-	})
-	defer tc.Shutdown()
+	kms := kms.TestKms(t, conn, wrapper)
+	repo, err := servers.NewRepository(rw, rw, kms)
+	require.NoError(err)
+	ctx := context.Background()
 
-	repo := tc.ServersRepo()
-	srv := servers.NewWorker(scope.Global.String(),
-		servers.WithPublicId("test_worker_1"),
+	wStatus := servers.NewWorkerStatus("test_worker_1",
 		servers.WithAddress("127.0.0.1"),
 		servers.WithWorkerTags(
 			&servers.Tag{
@@ -36,11 +160,10 @@ func TestTagUpdatingListing(t *testing.T) {
 				Value: "value2",
 			}))
 
-	_, _, err := repo.UpsertWorker(tc.Context(), srv, servers.WithUpdateTags(true))
+	_, _, err = repo.UpsertWorkerStatus(ctx, wStatus, servers.WithUpdateTags(true))
 	require.NoError(err)
 
-	srv = servers.NewWorker(scope.Global.String(),
-		servers.WithPublicId("test_worker_2"),
+	wStatus = servers.NewWorkerStatus("test_worker_2",
 		servers.WithAddress("127.0.0.1"),
 		servers.WithWorkerTags(
 			&servers.Tag{
@@ -51,10 +174,10 @@ func TestTagUpdatingListing(t *testing.T) {
 				Key:   "tag2",
 				Value: "value2",
 			}))
-	_, _, err = repo.UpsertWorker(tc.Context(), srv, servers.WithUpdateTags(true))
+	_, _, err = repo.UpsertWorkerStatus(ctx, wStatus, servers.WithUpdateTags(true))
 	require.NoError(err)
 
-	tags, err := repo.ListTagsForWorkers(tc.Context(), []string{"test_worker_1", "test_worker_2"})
+	tags, err := repo.ListTagsForWorkers(ctx, []string{"test_worker_1", "test_worker_2"})
 	require.NoError(err)
 
 	// Base case
@@ -82,8 +205,7 @@ func TestTagUpdatingListing(t *testing.T) {
 	require.Equal(exp, tags)
 
 	// Update without saying to update tags
-	srv = servers.NewWorker(scope.Global.String(),
-		servers.WithPublicId("test_worker_2"),
+	wStatus = servers.NewWorkerStatus("test_worker_2",
 		servers.WithAddress("192.168.1.1"),
 		servers.WithWorkerTags(
 			&servers.Tag{
@@ -94,16 +216,16 @@ func TestTagUpdatingListing(t *testing.T) {
 				Key:   "tag22",
 				Value: "value22",
 			}))
-	_, _, err = repo.UpsertWorker(tc.Context(), srv)
+	_, _, err = repo.UpsertWorkerStatus(ctx, wStatus)
 	require.NoError(err)
-	tags, err = repo.ListTagsForWorkers(tc.Context(), []string{"test_worker_1", "test_worker_2"})
+	tags, err = repo.ListTagsForWorkers(ctx, []string{"test_worker_1", "test_worker_2"})
 	require.NoError(err)
 	require.Equal(exp, tags)
 
 	// Update tags and test again
-	_, _, err = repo.UpsertWorker(tc.Context(), srv, servers.WithUpdateTags(true))
+	_, _, err = repo.UpsertWorkerStatus(ctx, wStatus, servers.WithUpdateTags(true))
 	require.NoError(err)
-	tags, err = repo.ListTagsForWorkers(tc.Context(), []string{"test_worker_1", "test_worker_2"})
+	tags, err = repo.ListTagsForWorkers(ctx, []string{"test_worker_1", "test_worker_2"})
 	require.NoError(err)
 	require.NotEqual(exp, tags)
 	// Update and try again
@@ -131,27 +253,26 @@ func TestListWorkersWithLiveness(t *testing.T) {
 	require.NoError(err)
 	ctx := context.Background()
 
-	newWorker := func(publicId string) *servers.Worker {
-		result := servers.NewWorker(scope.Global.String(),
-			servers.WithPublicId(publicId),
+	newWorkerStatus := func(publicId string) *servers.WorkerStatus {
+		result := servers.NewWorkerStatus(publicId,
 			servers.WithAddress("127.0.0.1"))
-		_, rowsUpdated, err := serversRepo.UpsertWorker(ctx, result)
+		_, rowsUpdated, err := serversRepo.UpsertWorkerStatus(ctx, result)
 		require.NoError(err)
 		require.Equal(1, rowsUpdated)
 
 		return result
 	}
 
-	server1 := newWorker("test_worker_1")
-	server2 := newWorker("test_worker_2")
-	server3 := newWorker("test_worker_3")
+	workConf1 := newWorkerStatus("test_worker_1")
+	workConf2 := newWorkerStatus("test_worker_2")
+	workConf3 := newWorkerStatus("test_worker_3")
 
 	// Sleep the default liveness time (15sec currently) +1s
 	time.Sleep(time.Second * 16)
 
 	// Push an upsert to the first worker so that its status has been
 	// updated.
-	_, rowsUpdated, err := serversRepo.UpsertWorker(ctx, server1)
+	_, rowsUpdated, err := serversRepo.UpsertWorkerStatus(ctx, workConf1)
 	require.NoError(err)
 	require.Equal(1, rowsUpdated)
 
@@ -174,10 +295,10 @@ func TestListWorkersWithLiveness(t *testing.T) {
 	result, err := serversRepo.ListWorkers(ctx)
 	require.NoError(err)
 	require.Len(result, 1)
-	requireIds([]string{server1.PublicId}, result)
+	requireIds([]string{workConf1.WorkerId}, result)
 
 	// Upsert second server.
-	_, rowsUpdated, err = serversRepo.UpsertWorker(ctx, server2)
+	_, rowsUpdated, err = serversRepo.UpsertWorkerStatus(ctx, workConf2)
 	require.NoError(err)
 	require.Equal(1, rowsUpdated)
 
@@ -186,11 +307,11 @@ func TestListWorkersWithLiveness(t *testing.T) {
 	result, err = serversRepo.ListWorkers(ctx, servers.WithLiveness(time.Second*5))
 	require.NoError(err)
 	require.Len(result, 2)
-	requireIds([]string{server1.PublicId, server2.PublicId}, result)
+	requireIds([]string{workConf1.WorkerId, workConf2.WorkerId}, result)
 
 	// Liveness disabled, should get all three workers.
 	result, err = serversRepo.ListWorkers(ctx, servers.WithLiveness(-1))
 	require.NoError(err)
 	require.Len(result, 3)
-	requireIds([]string{server1.PublicId, server2.PublicId, server3.PublicId}, result)
+	requireIds([]string{workConf1.WorkerId, workConf2.WorkerId, workConf3.WorkerId}, result)
 }

internal/servers/store/worker.pb.go

@@ -36,7 +36,7 @@ type Worker struct {
 	// 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. This must be set
+	// Address for the worker. This is optional.
 	// @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.
@@ -149,14 +149,14 @@ type WorkerTag struct {
 	unknownFields protoimpl.UnknownFields
 
 	// 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"`
+	// @inject_tag: `gorm:"primary_key"`
+	WorkerId string `protobuf:"bytes,10,opt,name=worker_id,json=workerId,proto3" json:"worker_id,omitempty" gorm:"primary_key"`
 	// 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"`
+	// @inject_tag: `gorm:"primary_key"`
+	Key string `protobuf:"bytes,20,opt,name=key,proto3" json:"key,omitempty" gorm:"primary_key"`
 	// value is the value
-	// @inject_tag: `gorm:"default:null"`
-	Value string `protobuf:"bytes,30,opt,name=value,proto3" json:"value,omitempty" gorm:"default:null"`
+	// @inject_tag: `gorm:"primary_key"`
+	Value string `protobuf:"bytes,30,opt,name=value,proto3" json:"value,omitempty" gorm:"primary_key"`
 	// source is the source of the tag. Either 'configuration' or 'api'.
 	// @inject_tag: `gorm:"default:not_null"`
 	Source string `protobuf:"bytes,40,opt,name=source,proto3" json:"source,omitempty" gorm:"default:not_null"`
@@ -222,6 +222,97 @@ func (x *WorkerTag) GetSource() string {
 	return ""
 }
 
+// WorkerStatus contains the fields that the worker reports to the controller
+// in its status  updates.
+type WorkerStatus struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	// worker_id is the public key that key of the worker this tag is for.
+	// @inject_tag: `gorm:"primary_key"`
+	WorkerId string `protobuf:"bytes,10,opt,name=worker_id,json=workerId,proto3" json:"worker_id,omitempty" gorm:"primary_key"`
+	// The create_time is set by the database.
+	// @inject_tag: `gorm:"default:current_timestamp"`
+	CreateTime *timestamp.Timestamp `protobuf:"bytes,20,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,30,opt,name=update_time,json=updateTime,proto3" json:"update_time,omitempty" gorm:"default:current_timestamp"`
+	// Name of the resource (optional)
+	// @inject_tag: `gorm:"default:null"`
+	Name string `protobuf:"bytes,40,opt,name=name,proto3" json:"name,omitempty" gorm:"default:null"`
+	// Address for the worker. This must be set.
+	// @inject_tag: `gorm:"default:null"`
+	Address string `protobuf:"bytes,50,opt,name=address,proto3" json:"address,omitempty" gorm:"default:null"`
+}
+
+func (x *WorkerStatus) Reset() {
+	*x = WorkerStatus{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_controller_storage_servers_store_v1_worker_proto_msgTypes[2]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *WorkerStatus) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*WorkerStatus) ProtoMessage() {}
+
+func (x *WorkerStatus) ProtoReflect() protoreflect.Message {
+	mi := &file_controller_storage_servers_store_v1_worker_proto_msgTypes[2]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use WorkerStatus.ProtoReflect.Descriptor instead.
+func (*WorkerStatus) Descriptor() ([]byte, []int) {
+	return file_controller_storage_servers_store_v1_worker_proto_rawDescGZIP(), []int{2}
+}
+
+func (x *WorkerStatus) GetWorkerId() string {
+	if x != nil {
+		return x.WorkerId
+	}
+	return ""
+}
+
+func (x *WorkerStatus) GetCreateTime() *timestamp.Timestamp {
+	if x != nil {
+		return x.CreateTime
+	}
+	return nil
+}
+
+func (x *WorkerStatus) GetUpdateTime() *timestamp.Timestamp {
+	if x != nil {
+		return x.UpdateTime
+	}
+	return nil
+}
+
+func (x *WorkerStatus) GetName() string {
+	if x != nil {
+		return x.Name
+	}
+	return ""
+}
+
+func (x *WorkerStatus) GetAddress() string {
+	if x != nil {
+		return x.Address
+	}
+	return ""
+}
+
 var File_controller_storage_servers_store_v1_worker_proto protoreflect.FileDescriptor
 
 var file_controller_storage_servers_store_v1_worker_proto_rawDesc = []byte{
@@ -260,11 +351,27 @@ var file_controller_storage_servers_store_v1_worker_proto_rawDesc = []byte{
 	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, 0x12, 0x16, 0x0a, 0x06, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x18, 0x28, 0x20, 0x01, 0x28,
-	0x09, 0x52, 0x06, 0x73, 0x6f, 0x75, 0x72, 0x63, 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,
+	0x09, 0x52, 0x06, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x22, 0xf3, 0x01, 0x0a, 0x0c, 0x57, 0x6f,
+	0x72, 0x6b, 0x65, 0x72, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 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, 0x4b, 0x0a, 0x0b, 0x63, 0x72, 0x65, 0x61, 0x74,
+	0x65, 0x5f, 0x74, 0x69, 0x6d, 0x65, 0x18, 0x14, 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, 0x1e, 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, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x28, 0x20, 0x01, 0x28, 0x09, 0x52,
+	0x04, 0x6e, 0x61, 0x6d, 0x65, 0x12, 0x18, 0x0a, 0x07, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73,
+	0x18, 0x32, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 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,
 }
 
 var (
@@ -279,20 +386,23 @@ 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, 2)
+var file_controller_storage_servers_store_v1_worker_proto_msgTypes = make([]protoimpl.MessageInfo, 3)
 var file_controller_storage_servers_store_v1_worker_proto_goTypes = []interface{}{
 	(*Worker)(nil),              // 0: controller.storage.servers.store.v1.Worker
 	(*WorkerTag)(nil),           // 1: controller.storage.servers.store.v1.WorkerTag
-	(*timestamp.Timestamp)(nil), // 2: controller.storage.timestamp.v1.Timestamp
+	(*WorkerStatus)(nil),        // 2: controller.storage.servers.store.v1.WorkerStatus
+	(*timestamp.Timestamp)(nil), // 3: controller.storage.timestamp.v1.Timestamp
 }
 var file_controller_storage_servers_store_v1_worker_proto_depIdxs = []int32{
-	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
+	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
+	3, // 2: controller.storage.servers.store.v1.WorkerStatus.create_time:type_name -> controller.storage.timestamp.v1.Timestamp
+	3, // 3: controller.storage.servers.store.v1.WorkerStatus.update_time:type_name -> controller.storage.timestamp.v1.Timestamp
+	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
 }
 
 func init() { file_controller_storage_servers_store_v1_worker_proto_init() }
@@ -325,6 +435,18 @@ func file_controller_storage_servers_store_v1_worker_proto_init() {
 				return nil
 			}
 		}
+		file_controller_storage_servers_store_v1_worker_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*WorkerStatus); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
 	}
 	type x struct{}
 	out := protoimpl.TypeBuilder{
@@ -332,7 +454,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:   2,
+			NumMessages:   3,
 			NumExtensions: 0,
 			NumServices:   0,
 		},

internal/servers/testing.go

@@ -79,6 +79,7 @@ func TestWorkerAuth(ctx context.Context, t *testing.T, conn *db.DB, worker *Work
 	return workerAuth
 }
 
+// TestWorker inserts a worker into the db to satisfy foreign key constraints.
 func TestWorker(t *testing.T, conn *db.DB, wrapper wrapping.Wrapper) *Worker {
 	t.Helper()
 	rw := db.New(conn)
@@ -89,12 +90,15 @@ func TestWorker(t *testing.T, conn *db.DB, wrapper wrapping.Wrapper) *Worker {
 	id, err := newWorkerId(context.Background())
 	require.NoError(t, err)
 	name := "test-worker-" + id
-
-	worker := NewWorker(scope.Global.String(),
-		WithPublicId(id),
+	wStatus := NewWorkerStatus(id,
 		WithName(name),
 		WithAddress("127.0.0.1"))
-	_, _, err = serversRepo.UpsertWorker(context.Background(), worker)
+	_, _, err = serversRepo.UpsertWorkerStatus(context.Background(), wStatus)
 	require.NoError(t, err)
-	return worker
+
+	wrk := NewWorker(scope.Global.String(), WithPublicId(id))
+	require.NoError(t, rw.LookupById(context.Background(), wrk))
+	wrk.ReportedStatus = wStatus
+
+	return wrk
 }

internal/servers/worker.go

@@ -1,6 +1,11 @@
 package servers
 
 import (
+	"context"
+	"strings"
+
+	"github.com/hashicorp/boundary/internal/db/timestamp"
+	"github.com/hashicorp/boundary/internal/errors"
 	"github.com/hashicorp/boundary/internal/servers/store"
 	"google.golang.org/protobuf/proto"
 )
@@ -10,7 +15,9 @@ import (
 // authorizing and establishing a session.  It is owned by a scope.
 type Worker struct {
 	*store.Worker
-	Tags []*Tag `gorm:"-"`
+
+	ReportedStatus *WorkerStatus `gorm:"-"`
+	Tags           []*Tag        `gorm:"-"`
 }
 
 // NewWorker returns a new Worker. Valid options are WithName, WithDescription
@@ -30,18 +37,163 @@ func NewWorker(scopeId string, opt ...Option) *Worker {
 }
 
 func (w *Worker) clone() *Worker {
+	if w == nil {
+		return nil
+	}
 	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)
+	crs := w.ReportedStatus.clone()
 	return &Worker{
-		Worker: cw.(*store.Worker),
-		Tags:   tags,
+		Worker:         cw.(*store.Worker),
+		Tags:           tags,
+		ReportedStatus: crs,
 	}
 }
 
+// CanonicalAddress returns the actual address boundary believes should be used
+// to communicate with this worker.  This will be the worker resource's address
+// unless it is not set in which case it will use address the worker provides
+// in its connection status updates.  If neither is available, an empty string
+// is returned.
+func (w *Worker) CanonicalAddress() string {
+	switch {
+	case w.Address != "":
+		return w.GetAddress()
+	case w.ReportedStatus != nil:
+		return w.ReportedStatus.GetAddress()
+	default:
+		return ""
+	}
+}
+
+// CanonicalTags is the deduplicated set of tags contained on both the resource
+// set over the API as well as the tags reported by the worker itself.
+func (w *Worker) CanonicalTags() map[string][]string {
+	dedupedTags := make(map[Tag]struct{})
+	for _, t := range w.Tags {
+		dedupedTags[*t] = struct{}{}
+	}
+	if w.ReportedStatus != nil {
+		for _, t := range w.ReportedStatus.Tags {
+			dedupedTags[*t] = struct{}{}
+		}
+	}
+	tags := make(map[string][]string)
+	for t := range dedupedTags {
+		tags[t.Key] = append(tags[t.Key], t.Value)
+	}
+	return tags
+}
+
+// LastConnectionUpdate contains the last time the worker has reported to the
+// controller its connection status.  If the worker has never reported to a
+// controller then nil is returned.
+func (w *Worker) LastConnectionUpdate() *timestamp.Timestamp {
+	if w.ReportedStatus == nil {
+		return nil
+	}
+	return w.ReportedStatus.GetUpdateTime()
+}
+
 // TableName overrides the table name used by Worker to `server_worker`
 func (Worker) TableName() string {
 	return "server_worker"
 }
+
+// workerAggregate contains an aggregated view of the values associated with
+// a single worker.
+type workerAggregate struct {
+	PublicId               string `gorm:"primary_key"`
+	ScopeId                string
+	Name                   string
+	Description            string
+	CreateTime             *timestamp.Timestamp
+	UpdateTime             *timestamp.Timestamp
+	Address                string
+	Version                uint32
+	ApiTags                string
+	WorkerStatusName       string
+	WorkerStatusAddress    string
+	WorkerStatusCreateTime *timestamp.Timestamp
+	WorkerStatusUpdateTime *timestamp.Timestamp
+	WorkerConfigTags       string
+}
+
+func (a *workerAggregate) toWorker(ctx context.Context) (*Worker, error) {
+	const op = "servers.(workerAggregate).toWorker"
+	workerOptions := []Option{
+		WithPublicId(a.PublicId),
+		WithName(a.Name),
+		WithDescription(a.Description),
+		WithAddress(a.Address),
+	}
+	tags, err := tagsFromAggregatedTagString(ctx, a.ApiTags)
+	if err != nil {
+		return nil, errors.Wrap(ctx, err, op, errors.WithMsg("error parsing config tag string"))
+	}
+	if len(tags) > 0 {
+		workerOptions = append(workerOptions, WithWorkerTags(tags...))
+	}
+	worker := NewWorker(a.ScopeId, workerOptions...)
+	worker.CreateTime = a.CreateTime
+	worker.UpdateTime = a.UpdateTime
+	worker.Version = a.Version
+
+	if a.WorkerStatusCreateTime == nil {
+		return worker, nil
+	}
+	statusOptions := []Option{
+		WithName(a.WorkerStatusName),
+		WithAddress(a.WorkerStatusAddress),
+	}
+	tags, err = tagsFromAggregatedTagString(ctx, a.WorkerConfigTags)
+	if err != nil {
+		return nil, errors.Wrap(ctx, err, op, errors.WithMsg("error parsing config tag string"))
+	}
+	if len(tags) > 0 {
+		statusOptions = append(statusOptions, WithWorkerTags(tags...))
+	}
+	cfg := NewWorkerStatus(a.PublicId, statusOptions...)
+	cfg.CreateTime = a.WorkerStatusCreateTime
+	cfg.UpdateTime = a.WorkerStatusUpdateTime
+
+	worker.ReportedStatus = cfg
+	return worker, nil
+}
+
+// tagsForAggregatedTagString parses a deliminated string in the format returned
+// by the database for the server_worker_aggregate view and returns []*Tag.
+// The string is in the format of key1Yvalue1Zkey2Yvalue2Zkey3Yvalue3. Y and Z
+// ares chosen for deliminators since tag keys and values are restricted from
+// having capitalized letters in them.
+func tagsFromAggregatedTagString(ctx context.Context, s string) ([]*Tag, error) {
+	if s == "" {
+		return nil, nil
+	}
+	const op = "servers.tagsFromAggregatedTagString"
+	const aggregateDelimiter = "Z"
+	const pairDelimiter = "Y"
+	var tags []*Tag
+	for _, kv := range strings.Split(s, aggregateDelimiter) {
+		res := strings.SplitN(kv, pairDelimiter, 3)
+		if len(res) != 2 {
+			return nil, errors.New(ctx, errors.Internal, op, "invalid aggregated tag pairs")
+		}
+		tags = append(tags, &Tag{
+			Key:   res[0],
+			Value: res[1],
+		})
+	}
+	return tags, nil
+}
+
+func (a *workerAggregate) GetPublicId() string {
+	return a.PublicId
+}
+
+func (workerAggregate) TableName() string {
+	return "server_worker_aggregate"
+}

internal/servers/worker_status.go

@@ -0,0 +1,50 @@
+package servers
+
+import (
+	"github.com/hashicorp/boundary/internal/servers/store"
+	"google.golang.org/protobuf/proto"
+)
+
+// A WorkerStatus is a container for the fields that a worker reports to the
+// controller about itself in its status updates.  While this may include a
+// Name, this is different from the Worker's name in that the Worker's name is
+// the name of the Worker resource in boundary where the name in the
+// WorkerStatus is the name the worker has reported about itself.
+type WorkerStatus struct {
+	*store.WorkerStatus
+	Tags []*Tag `gorm:"-"`
+}
+
+// NewWorkerStatus returns a new WorkerStatus. Valid options are WithName,
+// WithAddress, and WithWorkerTags. All other options are ignored.
+func NewWorkerStatus(workerId string, opt ...Option) *WorkerStatus {
+	opts := getOpts(opt...)
+	return &WorkerStatus{
+		WorkerStatus: &store.WorkerStatus{
+			WorkerId: workerId,
+			Name:     opts.withName,
+			Address:  opts.withAddress,
+		},
+		Tags: opts.withWorkerTags,
+	}
+}
+
+func (w *WorkerStatus) clone() *WorkerStatus {
+	if w == nil {
+		return nil
+	}
+	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.WorkerStatus)
+	return &WorkerStatus{
+		WorkerStatus: cw.(*store.WorkerStatus),
+		Tags:         tags,
+	}
+}
+
+// TableName overrides the table name used by WorkerStatus to `server_worker_status`
+func (WorkerStatus) TableName() string {
+	return "server_worker_status"
+}

internal/servers/worker_test.go

@@ -10,9 +10,219 @@ import (
 	"github.com/hashicorp/boundary/internal/servers/store"
 	"github.com/hashicorp/boundary/internal/types/scope"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"google.golang.org/protobuf/testing/protocmp"
 )
 
+func TestWorkerCanonicalAddress(t *testing.T) {
+	worker := NewWorker(scope.Global.String())
+	worker.ReportedStatus = NewWorkerStatus("foo", WithAddress("status"))
+	assert.Equal(t, "status", worker.CanonicalAddress())
+	worker.Address = "worker"
+	assert.Equal(t, "worker", worker.CanonicalAddress())
+}
+
+func TestWorkerCanonicalTags(t *testing.T) {
+	w := NewWorker(scope.Global.String(),
+		WithWorkerTags(
+			&Tag{Key: "key", Value: "apis unique"},
+			&Tag{Key: "key", Value: "shared"},
+			&Tag{Key: "key2", Value: "apis key2 unique"},
+		))
+	w.ReportedStatus = NewWorkerStatus("",
+		WithWorkerTags(
+			&Tag{Key: "key", Value: "configs unique"},
+			&Tag{Key: "key", Value: "shared"},
+			&Tag{Key: "key3", Value: "configs key3 unique"},
+		))
+
+	got := w.CanonicalTags()
+	assert.Len(t, got, 3, "2 keys expected, 'key' and 'key2'")
+	assert.ElementsMatch(t, got["key"], []string{"shared", "apis unique", "configs unique"})
+	assert.ElementsMatch(t, got["key2"], []string{"apis key2 unique"})
+	assert.ElementsMatch(t, got["key3"], []string{"configs key3 unique"})
+}
+
+func TestWorkerAggregate(t *testing.T) {
+	t.Parallel()
+	conn, _ := db.TestSetup(t, "postgres")
+	rw := db.New(conn)
+	ctx := context.Background()
+
+	getAggWorker := func(id string) *Worker {
+		agg := &workerAggregate{PublicId: id}
+		require.NoError(t, rw.LookupById(ctx, agg))
+		got, err := agg.toWorker(ctx)
+		assert.NoError(t, err)
+		return got
+	}
+
+	// Worker without a status
+	{
+		id, err := newWorkerId(ctx)
+		require.NoError(t, err)
+		require.NoError(t, rw.Create(ctx,
+			NewWorker(scope.Global.String(),
+				WithPublicId(id),
+				WithName(id),
+				WithAddress("address"))))
+
+		got := getAggWorker(id)
+		assert.Equal(t, id, got.GetPublicId())
+		assert.Equal(t, scope.Global.String(), got.GetScopeId())
+		assert.Equal(t, id, got.GetName())
+		assert.Equal(t, "address", got.GetAddress())
+		assert.Equal(t, uint32(1), got.GetVersion())
+		assert.Nil(t, got.ReportedStatus)
+		assert.Empty(t, got.CanonicalTags())
+	}
+
+	// Worker with status
+	{
+		id, err := newWorkerId(ctx)
+		require.NoError(t, err)
+		require.NoError(t, rw.Create(ctx,
+			NewWorker(scope.Global.String(), WithPublicId(id))))
+		require.NoError(t, rw.Create(ctx,
+			NewWorkerStatus(id,
+				WithAddress("address"),
+				WithName(id))))
+
+		got := getAggWorker(id)
+		assert.Equal(t, id, got.GetPublicId())
+		assert.Equal(t, uint32(1), got.GetVersion())
+		assert.NotNil(t, got.ReportedStatus)
+		assert.Equal(t, id, got.ReportedStatus.GetWorkerId())
+		assert.Equal(t, id, got.ReportedStatus.GetName())
+		assert.NotNil(t, got.ReportedStatus.CreateTime)
+		assert.NotNil(t, got.ReportedStatus.UpdateTime)
+		assert.Equal(t, "address", got.ReportedStatus.GetAddress())
+	}
+
+	// Worker with a config tag
+	{
+		id, err := newWorkerId(ctx)
+		require.NoError(t, err)
+		require.NoError(t, rw.Create(ctx,
+			NewWorker(scope.Global.String(), WithPublicId(id))))
+		require.NoError(t, rw.Create(ctx,
+			NewWorkerStatus(id, WithAddress("address"))))
+		require.NoError(t, rw.Create(ctx,
+			&store.WorkerTag{
+				WorkerId: id,
+				Key:      "key",
+				Value:    "val",
+				Source:   ConfigurationTagSource.String(),
+			}))
+
+		got := getAggWorker(id)
+		assert.Equal(t, id, got.GetPublicId())
+		assert.Equal(t, uint32(1), got.GetVersion())
+		assert.NotNil(t, got.ReportedStatus)
+		assert.Empty(t, got.Tags)
+		assert.Equal(t, got.ReportedStatus.Tags, []*Tag{{Key: "key", Value: "val"}})
+	}
+
+	// Worker with many config tag
+	{
+		id, err := newWorkerId(ctx)
+		require.NoError(t, err)
+		require.NoError(t, rw.Create(ctx,
+			NewWorker(scope.Global.String(), WithPublicId(id))))
+		require.NoError(t, rw.Create(ctx,
+			NewWorkerStatus(id, WithAddress("address"))))
+		require.NoError(t, rw.Create(ctx, &store.WorkerTag{
+			WorkerId: id,
+			Key:      "key",
+			Value:    "val",
+			Source:   ConfigurationTagSource.String(),
+		}))
+		require.NoError(t, rw.Create(ctx, &store.WorkerTag{
+			WorkerId: id,
+			Key:      "key",
+			Value:    "val2",
+			Source:   ConfigurationTagSource.String(),
+		}))
+		require.NoError(t, rw.Create(ctx, &store.WorkerTag{
+			WorkerId: id,
+			Key:      "key2",
+			Value:    "val2",
+			Source:   ConfigurationTagSource.String(),
+		}))
+
+		got := getAggWorker(id)
+		require.NotNil(t, got.ReportedStatus)
+		assert.Empty(t, got.Tags)
+		assert.ElementsMatch(t, got.ReportedStatus.Tags, []*Tag{
+			{Key: "key", Value: "val"},
+			{Key: "key", Value: "val2"},
+			{Key: "key2", Value: "val2"},
+		})
+	}
+
+	// Worker with an api tag
+	{
+		id, err := newWorkerId(ctx)
+		require.NoError(t, err)
+		require.NoError(t, rw.Create(ctx,
+			NewWorker(scope.Global.String(), WithPublicId(id))))
+		require.NoError(t, rw.Create(ctx,
+			NewWorkerStatus(id, WithAddress("address"))))
+		require.NoError(t, rw.Create(ctx,
+			&store.WorkerTag{
+				WorkerId: id,
+				Key:      "key",
+				Value:    "val",
+				Source:   ApiTagSource.String(),
+			}))
+
+		got := getAggWorker(id)
+		assert.Equal(t, id, got.GetPublicId())
+		assert.Equal(t, uint32(1), got.GetVersion())
+		assert.NotNil(t, got.ReportedStatus)
+		assert.Empty(t, got.ReportedStatus.Tags)
+		assert.Equal(t, got.Tags, []*Tag{{Key: "key", Value: "val"}})
+	}
+
+	// Worker with mix of tag sources
+	{
+		id, err := newWorkerId(ctx)
+		require.NoError(t, err)
+		require.NoError(t, rw.Create(ctx,
+			NewWorker(scope.Global.String(), WithPublicId(id))))
+		require.NoError(t, rw.Create(ctx,
+			NewWorkerStatus(id, WithAddress("address"))))
+		require.NoError(t, rw.Create(ctx, &store.WorkerTag{
+			WorkerId: id,
+			Key:      "key",
+			Value:    "val",
+			Source:   ConfigurationTagSource.String(),
+		}))
+		require.NoError(t, rw.Create(ctx, &store.WorkerTag{
+			WorkerId: id,
+			Key:      "key",
+			Value:    "val2",
+			Source:   ApiTagSource.String(),
+		}))
+		require.NoError(t, rw.Create(ctx, &store.WorkerTag{
+			WorkerId: id,
+			Key:      "key2",
+			Value:    "val2",
+			Source:   ApiTagSource.String(),
+		}))
+
+		got := getAggWorker(id)
+		require.NotNil(t, got.ReportedStatus)
+		assert.ElementsMatch(t, got.Tags, []*Tag{
+			{Key: "key", Value: "val2"},
+			{Key: "key2", Value: "val2"},
+		})
+		assert.ElementsMatch(t, got.ReportedStatus.Tags, []*Tag{
+			{Key: "key", Value: "val"},
+		})
+	}
+}
+
 func TestWorker_New(t *testing.T) {
 	t.Parallel()
 	conn, _ := db.TestSetup(t, "postgres")

internal/session/query.go

@@ -315,18 +315,25 @@ where
 	// * not closed
 	// * belong to servers that have not reported in within an acceptable
 	// threshold of time
+	// * belong to servers where we do not know when they last reported.
 	//
 	// and marks them as closed.
 	//
 	// The query returns the set of servers that have had connections closed
 	// along with their last update time and the number of connections closed on
-	// each.
+	// each.  If we do not know the last update time, we use the current time.
 	closeConnectionsForDeadServersCte = `
    with
    dead_workers (worker_id, last_update_time) as (
-         select public_id, update_time
-           from server_worker
-          where update_time < wt_sub_seconds_from_now(@grace_period_seconds)
+         select
+			w.public_id as worker_id,
+			ws.update_time as last_update_time
+           from server_worker w
+			left join server_worker_status ws
+				on w.public_id = ws.worker_id
+          where
+			ws.update_time is null
+			or ws.update_time < wt_sub_seconds_from_now(@grace_period_seconds)
    ),
    closed_connections (connection_id, worker_id) as (
          update session_connection
@@ -336,7 +343,7 @@ where
       returning public_id, worker_id
    )
    select closed_connections.worker_id,
-          dead_workers.last_update_time,
+          coalesce(dead_workers.last_update_time, current_timestamp) as last_update_time,
           count(closed_connections.connection_id) as number_connections_closed
      from closed_connections
      join dead_workers

internal/session/service_worker_status_report_test.go

@@ -4,15 +4,13 @@ import (
 	"context"
 	"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"
 	"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"
@@ -33,8 +31,8 @@ func TestWorkerStatusReport(t *testing.T) {
 		PrivateId: "test_controller1",
 		Address:   "127.0.0.1",
 	})
-	serverRepo.UpsertWorker(ctx, servers.NewWorker(scope.Global.String(),
-		servers.WithPublicId("test_worker1"),
+	serverRepo.UpsertWorkerStatus(ctx, servers.NewWorkerStatus(
+		"test_worker1",
 		servers.WithAddress("127.0.0.1")))
 
 	repo, err := session.NewRepository(rw, rw, kms)

internal/session/session_cleanup_job_test.go

@@ -108,7 +108,7 @@ func TestSessionConnectionCleanupJob(t *testing.T) {
 
 	// Push an upsert to the first worker so that its status has been
 	// updated.
-	_, rowsUpdated, err := serversRepo.UpsertWorker(ctx, worker1)
+	_, rowsUpdated, err := serversRepo.UpsertWorkerStatus(ctx, worker1.ReportedStatus)
 	require.NoError(err)
 	require.Equal(1, rowsUpdated)
 
@@ -121,7 +121,7 @@ func TestSessionConnectionCleanupJob(t *testing.T) {
 		require.True(ok)
 		require.Len(connIds, 6)
 		for _, connId := range connIds {
-			_, states, err := connectionRepo.LookupConnection(ctx, connId, nil)
+			_, states, err := connectionRepo.LookupConnection(ctx, connId)
 			require.NoError(err)
 			var foundClosed bool
 			for _, state := range states {
@@ -168,6 +168,61 @@ func TestSessionConnectionCleanupJobNewJobErr(t *testing.T) {
 	require.Nil(job)
 }
 
+func TestCloseConnections_ManuallyDeletedConfig(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	require := require.New(t)
+	conn, _ := db.TestSetup(t, "postgres")
+	rw := db.New(conn)
+	wrapper := db.TestWrapper(t)
+	iamRepo := iam.TestRepo(t, conn, wrapper)
+	kms := kms.TestKms(t, conn, wrapper)
+	repo, err := NewRepository(rw, rw, kms)
+	require.NoError(err)
+	connRepo, err := NewConnectionRepository(ctx, rw, rw, kms)
+	require.NoError(err)
+	serversRepo, err := servers.NewRepository(rw, rw, kms)
+	require.NoError(err)
+
+	gracePeriod := 1 * time.Second
+	job, err := newSessionConnectionCleanupJob(rw, deadWorkerConnCloseMinGrace)
+	require.NoError(err)
+
+	worker := servers.TestWorker(t, conn, wrapper)
+	sess := TestDefaultSession(t, conn, wrapper, iamRepo, WithDbOpts(db.WithSkipVetForWrite(true)))
+	sess, _, err = repo.ActivateSession(ctx, sess.GetPublicId(), sess.Version, []byte("foo"))
+	require.NoError(err)
+	c, cs, err := connRepo.AuthorizeConnection(ctx, sess.GetPublicId(), worker.GetPublicId())
+	require.NoError(err)
+	require.Len(cs, 1)
+	require.Equal(StatusAuthorized, cs[0].Status)
+
+	_, cs, err = connRepo.ConnectConnection(ctx, ConnectWith{
+		ConnectionId:       c.GetPublicId(),
+		ClientTcpAddress:   "127.0.0.1",
+		ClientTcpPort:      22,
+		EndpointTcpAddress: "127.0.0.1",
+		EndpointTcpPort:    22,
+		UserClientIp:       "127.0.0.1",
+	})
+
+	// The single worker's connections aren't cleaned up
+	_, _, err = serversRepo.UpsertWorkerStatus(ctx, worker.ReportedStatus)
+	assert.NoError(t, err)
+	result, err := job.closeConnectionsForDeadWorkers(ctx, gracePeriod)
+	require.NoError(err)
+	require.Empty(result)
+
+	// the manually deleted status causes the connections to be cleaned
+	_, _, err = serversRepo.UpsertWorkerStatus(ctx, worker.ReportedStatus)
+	assert.NoError(t, err)
+	_, err = rw.Delete(ctx, worker.ReportedStatus)
+	assert.NoError(t, err)
+	result, err = job.closeConnectionsForDeadWorkers(ctx, gracePeriod)
+	require.NoError(err)
+	require.Len(result, 1)
+}
+
 func TestCloseConnectionsForDeadWorkers(t *testing.T) {
 	t.Parallel()
 	ctx := context.Background()
@@ -290,7 +345,7 @@ func TestCloseConnectionsForDeadWorkers(t *testing.T) {
 	// the most up-to-date fields.
 	updateServer := func(t *testing.T, w *servers.Worker) *servers.Worker {
 		t.Helper()
-		_, rowsUpdated, err := serversRepo.UpsertWorker(ctx, w)
+		_, rowsUpdated, err := serversRepo.UpsertWorkerStatus(ctx, w.ReportedStatus)
 		require.NoError(err)
 		require.Equal(1, rowsUpdated)
 		servers, err := serversRepo.ListWorkers(ctx)
@@ -399,7 +454,7 @@ func TestCloseConnectionsForDeadWorkers(t *testing.T) {
 		require.Equal([]closeConnectionsForDeadWorkersResult{
 			{
 				WorkerId:                worker1.PublicId,
-				LastUpdateTime:          timestampPbAsUTC(t, worker1.UpdateTime.AsTime()),
+				LastUpdateTime:          timestampPbAsUTC(t, worker1.LastConnectionUpdate().AsTime()),
 				NumberConnectionsClosed: 12, // 18 per server, with 6 closed already
 			},
 		}, result)
@@ -423,12 +478,12 @@ func TestCloseConnectionsForDeadWorkers(t *testing.T) {
 		require.ElementsMatch([]closeConnectionsForDeadWorkersResult{
 			{
 				WorkerId:                worker2.PublicId,
-				LastUpdateTime:          timestampPbAsUTC(t, worker2.UpdateTime.AsTime()),
+				LastUpdateTime:          timestampPbAsUTC(t, worker2.LastConnectionUpdate().AsTime()),
 				NumberConnectionsClosed: 12, // 18 per server, with 6 closed already
 			},
 			{
 				WorkerId:                worker3.PublicId,
-				LastUpdateTime:          timestampPbAsUTC(t, worker3.UpdateTime.AsTime()),
+				LastUpdateTime:          timestampPbAsUTC(t, worker3.LastConnectionUpdate().AsTime()),
 				NumberConnectionsClosed: 12, // 18 per server, with 6 closed already
 			},
 		}, result)

testing/dbtest/session_list_benchmarks_dump_generation_test.go

@@ -22,7 +22,6 @@ import (
 	"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 +131,9 @@ func TestGenerateSessionBenchmarkTemplateDumps(t *testing.T) {
 			require.NoError(err)
 			serversRepo, err := servers.NewRepository(rw, rw, kms)
 			require.NoError(err)
-			worker := servers.NewWorker(scope.Global.String(),
-				servers.WithPublicId("test_worker_1"),
-				servers.WithAddress("127.0.0.1"))
-			_, _, err = serversRepo.UpsertWorker(ctx, worker)
+			_, _, err = serversRepo.UpsertWorkerStatus(ctx, servers.NewWorkerStatus(
+				"test_worker_1",
+				servers.WithAddress("127.0.0.1")))
 			require.NoError(err)
 
 			usersStart := time.Now()