diff --git a/internal/daemon/controller/handlers/targets/target_service.go b/internal/daemon/controller/handlers/targets/target_service.go index 5427a0691..251962b32 100644 --- a/internal/daemon/controller/handlers/targets/target_service.go +++ b/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 +} diff --git a/internal/daemon/controller/handlers/targets/target_service_test.go b/internal/daemon/controller/handlers/targets/target_service_test.go new file mode 100644 index 000000000..2644b4d78 --- /dev/null +++ b/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) + } +} diff --git a/internal/daemon/controller/handlers/workers/worker_service.go b/internal/daemon/controller/handlers/workers/worker_service.go index 20f5235ec..cedc1b213 100644 --- a/internal/daemon/controller/handlers/workers/worker_service.go +++ b/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) diff --git a/internal/daemon/controller/handlers/workers/worker_service_status_test.go b/internal/daemon/controller/handlers/workers/worker_service_status_test.go index bc3842897..8e1da4997 100644 --- a/internal/daemon/controller/handlers/workers/worker_service_status_test.go +++ b/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, }, diff --git a/internal/db/schema/migrations/oss/postgres/99/02_worker_controller_tables.up.sql b/internal/db/schema/migrations/oss/postgres/99/02_worker_controller_tables.up.sql index aa1ed8142..ef48224ea 100644 --- a/internal/db/schema/migrations/oss/postgres/99/02_worker_controller_tables.up.sql +++ b/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; diff --git a/internal/proto/controller/storage/servers/store/v1/worker.proto b/internal/proto/controller/storage/servers/store/v1/worker.proto index aa743f32d..281c9497a 100644 --- a/internal/proto/controller/storage/servers/store/v1/worker.proto +++ b/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; +} diff --git a/internal/servers/query.go b/internal/servers/query.go index 61e07b293..3d51150ee 100644 --- a/internal/servers/query.go +++ b/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 = ?` ) diff --git a/internal/servers/repository.go b/internal/servers/repository.go index c8dcab618..d482364d4 100644 --- a/internal/servers/repository.go +++ b/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 +} diff --git a/internal/servers/repository_worker.go b/internal/servers/repository_worker.go index 76fcda59d..e2e828362 100644 --- a/internal/servers/repository_worker.go +++ b/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 +} diff --git a/internal/servers/repository_worker_test.go b/internal/servers/repository_worker_test.go index 1ab6410b8..a69d474af 100644 --- a/internal/servers/repository_worker_test.go +++ b/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) } diff --git a/internal/servers/store/worker.pb.go b/internal/servers/store/worker.pb.go index 7dec56b04..fc7a0b554 100644 --- a/internal/servers/store/worker.pb.go +++ b/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, }, diff --git a/internal/servers/testing.go b/internal/servers/testing.go index 79e5c42b1..3a4ec7d28 100644 --- a/internal/servers/testing.go +++ b/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 } diff --git a/internal/servers/worker.go b/internal/servers/worker.go index eb1d9b0eb..b232e07a7 100644 --- a/internal/servers/worker.go +++ b/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" +} diff --git a/internal/servers/worker_status.go b/internal/servers/worker_status.go new file mode 100644 index 000000000..c8918618e --- /dev/null +++ b/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" +} diff --git a/internal/servers/worker_test.go b/internal/servers/worker_test.go index 4b4bf2acd..d01eeeca2 100644 --- a/internal/servers/worker_test.go +++ b/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") diff --git a/internal/session/query.go b/internal/session/query.go index 8ede00ad8..ae68dcece 100644 --- a/internal/session/query.go +++ b/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 diff --git a/internal/session/service_worker_status_report_test.go b/internal/session/service_worker_status_report_test.go index 330da5202..e524a7458 100644 --- a/internal/session/service_worker_status_report_test.go +++ b/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) diff --git a/internal/session/session_cleanup_job_test.go b/internal/session/session_cleanup_job_test.go index 75a75fd03..d7bb5887f 100644 --- a/internal/session/session_cleanup_job_test.go +++ b/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) diff --git a/testing/dbtest/session_list_benchmarks_dump_generation_test.go b/testing/dbtest/session_list_benchmarks_dump_generation_test.go index 5adf48a79..d09986b9e 100644 --- a/testing/dbtest/session_list_benchmarks_dump_generation_test.go +++ b/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()