From 2ecd550e276a09b84baf1a9436d75e685731f445 Mon Sep 17 00:00:00 2001 From: Todd Date: Wed, 18 May 2022 17:03:24 -0700 Subject: [PATCH] Add fields to server_worker to be a boundary resource (#2083) * Remove the server_id from session table, change it to worker_id on session_connection This is step 1 in updating the worker table. The worker is becoming a public resource so it must have a public id. We drop the server_id column on the session_connection table and add the worker_id column so we can use the wt_public_key type. Since db migrations are applied while the controller is offline, it should be safe to assume all session connections are not in a running state, thus allowing us to null out all the worker ids in the session_connection table. For now we don't have very good enforcement of the wt_public_id contraints, since the id is currently defined in the worker config, but this will soon change and the worker's id will be generated by the controller. * Updating server_id references to controller_id. * Adding worker tag db writing test. * Move the server table back to the top. * Update comment for the worker object. * Update comment for the worker table. * Update the newWorkerId call to take a context and fix its error. * Correct comment for the wt_network_address domain. --- .../server/worker_tags_reload_test.go | 10 +- .../handlers/targets/target_service.go | 12 +- .../handlers/workers/worker_service.go | 37 +-- .../workers/worker_service_status_test.go | 25 +- .../oss/postgres/99/01_domain_types.up.sql | 15 ++ ...sql => 02_worker_controller_tables.up.sql} | 51 +++- .../controller/storage/job/store/v1/job.proto | 14 +- .../storage/servers/store/v1/worker.proto | 43 ++- internal/scheduler/job/options.go | 8 +- internal/scheduler/job/options_test.go | 6 +- internal/scheduler/job/query.go | 2 +- internal/scheduler/job/repository_run.go | 12 +- internal/scheduler/job/repository_run_test.go | 244 +++++++++--------- internal/scheduler/job/store/job.pb.go | 75 +++--- internal/scheduler/job/testing.go | 4 +- internal/scheduler/job/testing_test.go | 2 +- internal/scheduler/scheduler.go | 2 +- internal/servers/gorm.go | 6 - internal/servers/options.go | 53 +++- internal/servers/options_test.go | 63 ++++- internal/servers/public_ids.go | 21 ++ internal/servers/repository.go | 74 +++--- internal/servers/repository_test.go | 125 ++++----- internal/servers/store/gorm.go | 10 +- internal/servers/store/worker.pb.go | 151 ++++++----- internal/servers/worker.go | 55 ++++ internal/servers/worker_tags_test.go | 78 ++++++ internal/servers/worker_test.go | 185 +++++++++++++ .../session/service_worker_status_report.go | 2 +- .../service_worker_status_report_test.go | 10 +- internal/session/session_cleanup_job_test.go | 3 +- internal/session/testing.go | 13 +- ...on_list_benchmarks_dump_generation_test.go | 13 +- 33 files changed, 967 insertions(+), 457 deletions(-) create mode 100644 internal/db/schema/migrations/oss/postgres/99/01_domain_types.up.sql rename internal/db/schema/migrations/oss/postgres/99/{01_worker_controller_tables.up.sql => 02_worker_controller_tables.up.sql} (70%) delete mode 100644 internal/servers/gorm.go create mode 100644 internal/servers/public_ids.go create mode 100644 internal/servers/worker.go create mode 100644 internal/servers/worker_tags_test.go create mode 100644 internal/servers/worker_test.go diff --git a/internal/cmd/commands/server/worker_tags_reload_test.go b/internal/cmd/commands/server/worker_tags_reload_test.go index f650e58fb..5229e8cd7 100644 --- a/internal/cmd/commands/server/worker_tags_reload_test.go +++ b/internal/cmd/commands/server/worker_tags_reload_test.go @@ -100,11 +100,11 @@ func TestServer_ReloadWorkerTags(t *testing.T) { require.NoError(err) tags, err := serversRepo.ListTagsForWorkers(testController.Context(), []string{name}) require.NoError(err) - require.Len(tags, 2) - require.Equal(key, tags[0].Key) - require.Equal(values[0], tags[0].Value) - require.Equal(key, tags[1].Key) - require.Equal(values[1], tags[1].Value) + require.Len(tags[name], 2) + require.Equal(key, tags[name][0].Key) + require.Equal(values[0], tags[name][0].Value) + require.Equal(key, tags[name][1].Key) + require.Equal(values[1], tags[name][1].Value) } // Give time to populate up to the controller diff --git a/internal/daemon/controller/handlers/targets/target_service.go b/internal/daemon/controller/handlers/targets/target_service.go index e1713c421..5427a0691 100644 --- a/internal/daemon/controller/handlers/targets/target_service.go +++ b/internal/daemon/controller/handlers/targets/target_service.go @@ -911,7 +911,7 @@ func (s Service) AuthorizeSession(ctx context.Context, req *pbs.AuthorizeSession if hasWorkerFilter && len(workerIds) > 0 { finalWorkers := make([]*pb.WorkerInfo, 0, len(workers)) // Fetch the tags for the given worker IDs - tags, err := serversRepo.ListTagsForWorkers(ctx, workerIds) + workerTags, err := serversRepo.ListTagsForWorkers(ctx, workerIds) if err != nil { return nil, err } @@ -919,13 +919,15 @@ func (s Service) AuthorizeSession(ctx context.Context, req *pbs.AuthorizeSession // built from the worker config, but with one extra level: a map of the // worker's ID to its filter map. tagMap := make(map[string]map[string][]string) - for _, tag := range tags { - currWorkerMap := tagMap[tag.WorkerId] + for wId, tags := range workerTags { + currWorkerMap := tagMap[wId] if currWorkerMap == nil { currWorkerMap = make(map[string][]string) - tagMap[tag.WorkerId] = currWorkerMap + tagMap[wId] = currWorkerMap + } + for _, tag := range tags { + currWorkerMap[tag.Key] = append(currWorkerMap[tag.Key], tag.Value) } - currWorkerMap[tag.Key] = append(currWorkerMap[tag.Key], tag.Value) // We don't need to reinsert after the fact because maps are // reference types, so we don't need to re-insert into tagMap } diff --git a/internal/daemon/controller/handlers/workers/worker_service.go b/internal/daemon/controller/handlers/workers/worker_service.go index f14fd004a..20f5235ec 100644 --- a/internal/daemon/controller/handlers/workers/worker_service.go +++ b/internal/daemon/controller/handlers/workers/worker_service.go @@ -7,8 +7,6 @@ import ( "sync" "time" - "github.com/hashicorp/boundary/internal/servers/store" - "github.com/hashicorp/boundary/internal/daemon/controller/common" "github.com/hashicorp/boundary/internal/daemon/controller/handlers" pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services" @@ -16,6 +14,7 @@ import ( "github.com/hashicorp/boundary/internal/observability/event" "github.com/hashicorp/boundary/internal/servers" "github.com/hashicorp/boundary/internal/session" + "github.com/hashicorp/boundary/internal/types/scope" "github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/targets" "github.com/hashicorp/go-bexpr" "google.golang.org/grpc/codes" @@ -74,20 +73,20 @@ func (ws *workerServiceServer) Status(ctx context.Context, req *pbs.StatusReques reqServer := req.GetWorker() // Convert API tags to storage tags - workerTags := make(map[string]*store.TagValues) + workerTags := make([]*servers.Tag, 0, len(reqServer.Tags)) for k, v := range reqServer.Tags { - thisTag := &store.TagValues{ - Values: v.GetValues(), + for _, val := range v.GetValues() { + workerTags = append(workerTags, &servers.Tag{ + Key: k, + Value: val, + }) } - workerTags[k] = thisTag - } - worker := &store.Worker{ - PublicId: reqServer.PrivateId, - Address: reqServer.Address, - CreateTime: reqServer.CreateTime, - UpdateTime: reqServer.UpdateTime, - Tags: workerTags, } + + worker := servers.NewWorker(scope.Global.String(), + servers.WithPublicId(reqServer.PrivateId), + servers.WithAddress(reqServer.Address), + servers.WithWorkerTags(workerTags...)) controllers, _, err := serverRepo.UpsertWorker(ctx, worker, servers.WithUpdateTags(req.GetUpdateTags())) if err != nil { event.WriteError(ctx, op, err, event.WithInfoMsg("error storing worker status")) @@ -199,17 +198,19 @@ func (ws *workerServiceServer) LookupSession(ctx context.Context, req *pbs.Looku event.WriteError(ctx, op, err, event.WithInfoMsg("error getting servers repo")) return &pbs.LookupSessionResponse{}, status.Errorf(codes.Internal, "Error acquiring server repo when looking up session: %v", err) } - tags, err := serversRepo.ListTagsForWorkers(ctx, []string{req.ServerId}) + workerTags, err := serversRepo.ListTagsForWorkers(ctx, []string{req.ServerId}) if err != nil { event.WriteError(ctx, op, err, event.WithInfoMsg("error looking up tags for server", "server_id", req.ServerId)) return &pbs.LookupSessionResponse{}, status.Errorf(codes.Internal, "Error looking up tags for server: %v", err) } // Build the map for filtering. tagMap := make(map[string][]string) - for _, tag := range tags { - tagMap[tag.Key] = append(tagMap[tag.Key], tag.Value) - // We don't need to reinsert after the fact because maps are - // reference types, so we don't need to re-insert into tagMap + for _, tags := range workerTags { + for _, tag := range tags { + tagMap[tag.Key] = append(tagMap[tag.Key], tag.Value) + // We don't need to reinsert after the fact because maps are + // reference types, so we don't need to re-insert into tagMap + } } // Create the evaluator 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 85b104e18..9c8854fa2 100644 --- a/internal/daemon/controller/handlers/workers/worker_service_status_test.go +++ b/internal/daemon/controller/handlers/workers/worker_service_status_test.go @@ -6,6 +6,7 @@ import ( "testing" "github.com/hashicorp/boundary/internal/servers/store" + "github.com/hashicorp/boundary/internal/types/scope" "github.com/google/go-cmp/cmp" "github.com/google/go-cmp/cmp/cmpopts" @@ -37,10 +38,10 @@ func TestStatus(t *testing.T) { PrivateId: "test_controller1", Address: "127.0.0.1", }) - serverRepo.UpsertWorker(ctx, &store.Worker{ - PublicId: "test_worker1", - Address: "127.0.0.1", - }) + serverRepo.UpsertWorker(ctx, servers.NewWorker( + scope.Global.String(), + servers.WithPublicId("test_worker1"), + servers.WithAddress("127.0.0.1"))) serversRepoFn := func() (*servers.Repository, error) { return serverRepo, nil @@ -205,10 +206,10 @@ func TestStatusSessionClosed(t *testing.T) { PrivateId: "test_controller1", Address: "127.0.0.1", }) - serverRepo.UpsertWorker(ctx, &store.Worker{ - PublicId: "test_worker1", - Address: "127.0.0.1", - }) + serverRepo.UpsertWorker(ctx, servers.NewWorker( + scope.Global.String(), + servers.WithPublicId("test_worker1"), + servers.WithAddress("127.0.0.1"))) serversRepoFn := func() (*servers.Repository, error) { return serverRepo, nil @@ -387,10 +388,10 @@ func TestStatusDeadConnection(t *testing.T) { PrivateId: "test_controller1", Address: "127.0.0.1", }) - serverRepo.UpsertWorker(ctx, &store.Worker{ - PublicId: "test_worker1", - Address: "127.0.0.1", - }) + serverRepo.UpsertWorker(ctx, servers.NewWorker( + scope.Global.String(), + servers.WithPublicId("test_worker1"), + servers.WithAddress("127.0.0.1"))) serversRepoFn := func() (*servers.Repository, error) { return serverRepo, nil diff --git a/internal/db/schema/migrations/oss/postgres/99/01_domain_types.up.sql b/internal/db/schema/migrations/oss/postgres/99/01_domain_types.up.sql new file mode 100644 index 000000000..504274797 --- /dev/null +++ b/internal/db/schema/migrations/oss/postgres/99/01_domain_types.up.sql @@ -0,0 +1,15 @@ +begin; + +-- wt_network_address defines a type for a network address. +-- While wt_dns_names does exist, a wt_network_address can be either a dns name +-- or an ip address. +create domain wt_network_address as text + constraint wt_network_address_too_short + check (length(trim(value)) > 0) + constraint wt_network_address_too_long + check (length(trim(value)) < 256); +comment on domain wt_network_address is + 'standard column for a network address.'; + + +commit; \ No newline at end of file diff --git a/internal/db/schema/migrations/oss/postgres/99/01_worker_controller_tables.up.sql b/internal/db/schema/migrations/oss/postgres/99/02_worker_controller_tables.up.sql similarity index 70% rename from internal/db/schema/migrations/oss/postgres/99/01_worker_controller_tables.up.sql rename to internal/db/schema/migrations/oss/postgres/99/02_worker_controller_tables.up.sql index 22e4b8437..88f97444b 100644 --- a/internal/db/schema/migrations/oss/postgres/99/01_worker_controller_tables.up.sql +++ b/internal/db/schema/migrations/oss/postgres/99/02_worker_controller_tables.up.sql @@ -1,12 +1,11 @@ begin; -- Split the server table into two new tables: controller and worker + create table server_controller ( private_id text primary key, description wt_description, - address text not null - constraint address_must_not_be_empty - check(length(trim(address)) > 0), + address wt_network_address not null, create_time wt_timestamp, update_time wt_timestamp ); @@ -30,19 +29,25 @@ create trigger controller_update_time_column before update on server_controller -- workers will now be exposed as resources in boundary. create table server_worker ( public_id wt_public_id primary key, + scope_id wt_scope_id not null + references iam_scope_global(scope_id) + on delete cascade + on update cascade, description wt_description, name wt_name unique, - address text not null - constraint address_must_not_be_empty - check(length(trim(address)) > 0), + -- The address can be null since it is an optional value from the API. + address wt_network_address, create_time wt_timestamp, - update_time wt_timestamp + update_time wt_timestamp, + version wt_version, + constraint server_worker_scope_id_name_uq + unique(scope_id, name) ); comment on table server_worker is 'server_worker is a table where each row represents a Boundary worker.'; create trigger immutable_columns before update on server_worker - for each row execute procedure immutable_columns('public_id','create_time'); + for each row execute procedure immutable_columns('public_id', 'scope_id', 'create_time'); create trigger default_create_time_column before insert on server_worker for each row execute procedure default_create_time(); @@ -96,15 +101,39 @@ alter table session_connection on update cascade; -- Update job run table so that server id references controller id +-- We are not migrating the values from server_id to controller_id. The fkey +-- constraint says that server_id can be set to null when the server is deleted +-- which is what this migration does (removing all records from the server table). +-- Not migrating values make it easier to change types in the server_worker and +-- server_controller tables (like from text to wt_public_id or text to wt_address) +-- without having to worry about old values being valid in the new types. +-- Finally, neither jobs nor servers are exposed out of boundary so the risk of +-- losing data that would be useful later on is diminished. +alter table job_run + add column controller_id wt_private_id, + drop column server_id; alter table job_run - drop constraint server_fkey, add constraint server_controller_fkey - foreign key (server_id) + foreign key (controller_id) references server_controller (private_id) on delete set null on update cascade; --- Replaces the view created in 9/01 to remove references to the server/worker id. +-- Since the above alter tables sets all controller_ids to null running jobs +-- can no longer be reclaimed by any controller and should be considered +-- interrupted. +update job_run +set + status = 'interrupted', + end_time = current_timestamp +where + status = 'running'; + + +-- Replaces the view created in 9/01. +-- Remove the worker id from this view. In actuality this is almost a no-op +-- because no server information was ever getting populated here due to a bug +-- in the update mask when updating a session at the time we activate a session. create view session_list as select s.public_id, s.user_id, s.host_id, s.target_id, diff --git a/internal/proto/controller/storage/job/store/v1/job.proto b/internal/proto/controller/storage/job/store/v1/job.proto index dd8c3caca..f5cdb1638 100644 --- a/internal/proto/controller/storage/job/store/v1/job.proto +++ b/internal/proto/controller/storage/job/store/v1/job.proto @@ -26,6 +26,12 @@ message Job { } message JobRun { + // server_id referenced an server table which was replaced with both a + // server_controller table and a server_worker table. For the purpose of + // jobs, only the controller table is relevant thus we replaced this column + // with controller_id. + reserved "server_id"; + // private_id is used to access the job run but not intended to be available via the API // @inject_tag: `gorm:"primary_key;default:null"` string private_id = 1; @@ -38,10 +44,6 @@ message JobRun { // @inject_tag: `gorm:"not_null"` string job_name = 3; - // The server_id of the server running the job and must be set. - // @inject_tag: `gorm:"not_null"` - string server_id = 4; - // The create_time is set by the database. // @inject_tag: `gorm:"default:current_timestamp"` timestamp.v1.Timestamp create_time = 5; @@ -65,4 +67,8 @@ message JobRun { // status of the job run (running, completed, failed or interrupted). // @inject_tag: `gorm:"not_null"` string status = 10; + + // The controller_id of the controller running the job and must be set. + // @inject_tag: `gorm:"not_null"` + string controller_id = 11; } diff --git a/internal/proto/controller/storage/servers/store/v1/worker.proto b/internal/proto/controller/storage/servers/store/v1/worker.proto index a5063074e..3898131d2 100644 --- a/internal/proto/controller/storage/servers/store/v1/worker.proto +++ b/internal/proto/controller/storage/servers/store/v1/worker.proto @@ -8,12 +8,11 @@ option go_package = "github.com/hashicorp/boundary/internal/servers/store;store" // Worker contains all fields related to a Worker resource message Worker { - // Public ID of the resource + // public_id is a surrogate key suitable for use in a public API // @inject_tag: `gorm:"primary_key"` string public_id = 10; // Name of the resource (optional) - // Description of the resource (optional) // @inject_tag: `gorm:"default:null"` string name = 20; @@ -21,21 +20,39 @@ message Worker { // @inject_tag: `gorm:"default:null"` string description = 30; - // Address for the worker + // Address for the worker. This must be set + // @inject_tag: `gorm:"default:null"` string address = 40; - // First seen time from the RDBMS - storage.timestamp.v1.Timestamp create_time = 50; + // The create_time is set by the database. + // @inject_tag: `gorm:"default:current_timestamp"` + timestamp.v1.Timestamp create_time = 50; - // Last time there was an update - storage.timestamp.v1.Timestamp update_time = 60; + // The update_time is set by the database. + // @inject_tag: `gorm:"default:current_timestamp"` + timestamp.v1.Timestamp update_time = 60; - // @gotags: `gorm:"-"` - map tags = 70; + // The scope_id of the owning scope and must be set. + // @inject_tag: `gorm:"not_null"` + string scope_id = 70; + + // version allows optimistic locking of the resource. + // @inject_tag: `gorm:"default:null"` + uint32 version = 80; } -// TagValues is used because map fields cannot be repeated but can be a -// message -message TagValues { - repeated string values = 10; +// WorkerTag is a tag for a worker. The primary key is comprised of the +// worker_id, key, value, and source. +message WorkerTag { + // worker_id is the public key that key of the worker this tag is for. + // @inject_tag: `gorm:"default:null"` + string worker_id = 10; + + // key is the key of the tag. This must be set. + // @inject_tag: `gorm:"default:null"` + string key = 20; + + // value is the value + // @inject_tag: `gorm:"default:null"` + string value = 30; } diff --git a/internal/scheduler/job/options.go b/internal/scheduler/job/options.go index 5b0cc64cf..493ae0837 100644 --- a/internal/scheduler/job/options.go +++ b/internal/scheduler/job/options.go @@ -27,7 +27,7 @@ type options struct { withRunJobsLimit uint withLimit int withName string - withServerId string + withControllerId string } func getDefaultOptions() options { @@ -72,9 +72,9 @@ func WithName(n string) Option { } } -// WithServerId provides an option to provide the server id to match when calling InterruptRuns -func WithServerId(id string) Option { +// WithControllerId provides an option to provide the server id to match when calling InterruptRuns +func WithControllerId(id string) Option { return func(o *options) { - o.withServerId = id + o.withControllerId = id } } diff --git a/internal/scheduler/job/options_test.go b/internal/scheduler/job/options_test.go index 1b7e4e78d..ff6c0484c 100644 --- a/internal/scheduler/job/options_test.go +++ b/internal/scheduler/job/options_test.go @@ -49,12 +49,12 @@ func Test_GetOpts(t *testing.T) { testOpts.withName = "name" assert.Equal(opts, testOpts) }) - t.Run("WithServerId", func(t *testing.T) { + t.Run("WithControllerId", func(t *testing.T) { assert := assert.New(t) - opts := getOpts(WithServerId("server_id")) + opts := getOpts(WithControllerId("controller_id")) testOpts := getDefaultOptions() assert.NotEqual(opts, testOpts) - testOpts.withServerId = "server_id" + testOpts.withControllerId = "controller_id" assert.Equal(opts, testOpts) }) } diff --git a/internal/scheduler/job/query.go b/internal/scheduler/job/query.go index 48f5e185e..4fd9c9aa4 100644 --- a/internal/scheduler/job/query.go +++ b/internal/scheduler/job/query.go @@ -2,7 +2,7 @@ package job const runJobsQuery = ` insert into job_run ( - job_plugin_id, job_name, server_id + job_plugin_id, job_name, controller_id ) select job_plugin_id, job_name, ? diff --git a/internal/scheduler/job/repository_run.go b/internal/scheduler/job/repository_run.go index 0c630b332..5288faf25 100644 --- a/internal/scheduler/job/repository_run.go +++ b/internal/scheduler/job/repository_run.go @@ -255,7 +255,7 @@ func (r *Repository) FailRun(ctx context.Context, runId string, completed, total // Once a run has been persisted with a final run status (completed, failed // or interrupted), any future calls to InterruptRuns will return an error with Code // errors.InvalidJobRunState. -// WithServerId is the only valid option +// WithControllerId is the only valid option func (r *Repository) InterruptRuns(ctx context.Context, interruptThreshold time.Duration, opt ...Option) ([]*Run, error) { const op = "job.(Repository).InterruptRuns" @@ -263,12 +263,12 @@ func (r *Repository) InterruptRuns(ctx context.Context, interruptThreshold time. // interruptThreshold is seconds in past so * -1 args := []interface{}{-1 * int(interruptThreshold.Round(time.Second).Seconds())} - var whereServerId string - if opts.withServerId != "" { - whereServerId = "and server_id = ?" - args = append(args, opts.withServerId) + var whereControllerId string + if opts.withControllerId != "" { + whereControllerId = "and controller_id = ?" + args = append(args, opts.withControllerId) } - query := fmt.Sprintf(interruptRunsQuery, whereServerId) + query := fmt.Sprintf(interruptRunsQuery, whereControllerId) var runs []*Run _, err := r.writer.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{}, diff --git a/internal/scheduler/job/repository_run_test.go b/internal/scheduler/job/repository_run_test.go index a89e24914..69b8e8808 100644 --- a/internal/scheduler/job/repository_run_test.go +++ b/internal/scheduler/job/repository_run_test.go @@ -27,30 +27,30 @@ func TestRepository_RunJobs(t *testing.T) { server := testController(t, conn, wrapper) tests := []struct { - name string - serverId string - job *Job - wantRun bool - wantErr bool - wantErrCode errors.Code - wantErrMsg string + name string + ControllerId string + job *Job + wantRun bool + wantErr bool + wantErrCode errors.Code + wantErrMsg string }{ { - name: "missing-server-id", - serverId: "", - wantErr: true, - wantErrCode: errors.InvalidParameter, - wantErrMsg: "job.(Repository).RunJobs: missing server id: parameter violation: error #100", + name: "missing-server-id", + ControllerId: "", + wantErr: true, + wantErrCode: errors.InvalidParameter, + wantErrMsg: "job.(Repository).RunJobs: missing server id: parameter violation: error #100", }, { - name: "no-work", - serverId: server.PrivateId, - wantRun: false, - wantErr: false, + name: "no-work", + ControllerId: server.PrivateId, + wantRun: false, + wantErr: false, }, { - name: "valid", - serverId: server.PrivateId, + name: "valid", + ControllerId: server.PrivateId, job: &Job{ Job: &store.Job{ Name: "valid-test", @@ -60,8 +60,8 @@ func TestRepository_RunJobs(t *testing.T) { wantRun: true, }, { - name: "fake-server-id", - serverId: "fake-server-id", + name: "fake-server-id", + ControllerId: "fake-server-id", job: &Job{ Job: &store.Job{ Name: "fake-server-id-test", @@ -84,7 +84,7 @@ func TestRepository_RunJobs(t *testing.T) { assert.NoError(err) require.NotNil(repo) - got, err := repo.RunJobs(context.Background(), tt.serverId) + got, err := repo.RunJobs(context.Background(), tt.ControllerId) if tt.wantErr { require.Error(err) assert.Truef(errors.Match(errors.T(tt.wantErrCode), err), "Unexpected error %s", err) @@ -253,10 +253,10 @@ func TestRepository_UpdateProgress(t *testing.T) { name: "status-already-interrupted", orig: &Run{ JobRun: &store.JobRun{ - JobName: job.Name, - JobPluginId: job.PluginId, - ServerId: server.PrivateId, - Status: Interrupted.string(), + JobName: job.Name, + JobPluginId: job.PluginId, + ControllerId: server.PrivateId, + Status: Interrupted.string(), }, }, wantErr: true, @@ -267,10 +267,10 @@ func TestRepository_UpdateProgress(t *testing.T) { name: "status-already-failed", orig: &Run{ JobRun: &store.JobRun{ - JobName: job.Name, - JobPluginId: job.PluginId, - ServerId: server.PrivateId, - Status: Failed.string(), + JobName: job.Name, + JobPluginId: job.PluginId, + ControllerId: server.PrivateId, + Status: Failed.string(), }, }, wantErr: true, @@ -281,10 +281,10 @@ func TestRepository_UpdateProgress(t *testing.T) { name: "status-already-completed", orig: &Run{ JobRun: &store.JobRun{ - JobName: job.Name, - JobPluginId: job.PluginId, - ServerId: server.PrivateId, - Status: Completed.string(), + JobName: job.Name, + JobPluginId: job.PluginId, + ControllerId: server.PrivateId, + Status: Completed.string(), }, }, wantErr: true, @@ -295,10 +295,10 @@ func TestRepository_UpdateProgress(t *testing.T) { name: "valid-no-changes", orig: &Run{ JobRun: &store.JobRun{ - JobName: job.Name, - JobPluginId: job.PluginId, - ServerId: server.PrivateId, - Status: Running.string(), + JobName: job.Name, + JobPluginId: job.PluginId, + ControllerId: server.PrivateId, + Status: Running.string(), }, }, }, @@ -306,10 +306,10 @@ func TestRepository_UpdateProgress(t *testing.T) { name: "valid-update-total", orig: &Run{ JobRun: &store.JobRun{ - JobName: job.Name, - JobPluginId: job.PluginId, - ServerId: server.PrivateId, - Status: Running.string(), + JobName: job.Name, + JobPluginId: job.PluginId, + ControllerId: server.PrivateId, + Status: Running.string(), }, }, args: args{ @@ -323,11 +323,11 @@ func TestRepository_UpdateProgress(t *testing.T) { name: "valid-update-completed", orig: &Run{ JobRun: &store.JobRun{ - JobName: job.Name, - JobPluginId: job.PluginId, - ServerId: server.PrivateId, - Status: Running.string(), - TotalCount: 10, + JobName: job.Name, + JobPluginId: job.PluginId, + ControllerId: server.PrivateId, + Status: Running.string(), + TotalCount: 10, }, }, args: args{ @@ -343,10 +343,10 @@ func TestRepository_UpdateProgress(t *testing.T) { name: "valid-update-completed-and-total", orig: &Run{ JobRun: &store.JobRun{ - JobName: job.Name, - JobPluginId: job.PluginId, - ServerId: server.PrivateId, - Status: Running.string(), + JobName: job.Name, + JobPluginId: job.PluginId, + ControllerId: server.PrivateId, + Status: Running.string(), }, }, args: args{ @@ -362,10 +362,10 @@ func TestRepository_UpdateProgress(t *testing.T) { name: "invalid-completed-greater-than-total", orig: &Run{ JobRun: &store.JobRun{ - JobName: job.Name, - JobPluginId: job.PluginId, - ServerId: server.PrivateId, - Status: Running.string(), + JobName: job.Name, + JobPluginId: job.PluginId, + ControllerId: server.PrivateId, + Status: Running.string(), }, }, args: args{ @@ -467,10 +467,10 @@ func TestRepository_CompleteRun(t *testing.T) { name: "status-already-interrupted", orig: &Run{ JobRun: &store.JobRun{ - JobName: job.Name, - JobPluginId: job.PluginId, - ServerId: server.PrivateId, - Status: Interrupted.string(), + JobName: job.Name, + JobPluginId: job.PluginId, + ControllerId: server.PrivateId, + Status: Interrupted.string(), }, }, wantErr: true, @@ -481,10 +481,10 @@ func TestRepository_CompleteRun(t *testing.T) { name: "status-already-failed", orig: &Run{ JobRun: &store.JobRun{ - JobName: job.Name, - JobPluginId: job.PluginId, - ServerId: server.PrivateId, - Status: Failed.string(), + JobName: job.Name, + JobPluginId: job.PluginId, + ControllerId: server.PrivateId, + Status: Failed.string(), }, }, wantErr: true, @@ -495,10 +495,10 @@ func TestRepository_CompleteRun(t *testing.T) { name: "status-already-completed", orig: &Run{ JobRun: &store.JobRun{ - JobName: job.Name, - JobPluginId: job.PluginId, - ServerId: server.PrivateId, - Status: Completed.string(), + JobName: job.Name, + JobPluginId: job.PluginId, + ControllerId: server.PrivateId, + Status: Completed.string(), }, }, wantErr: true, @@ -509,10 +509,10 @@ func TestRepository_CompleteRun(t *testing.T) { name: "valid", orig: &Run{ JobRun: &store.JobRun{ - JobName: job.Name, - JobPluginId: job.PluginId, - ServerId: server.PrivateId, - Status: Running.string(), + JobName: job.Name, + JobPluginId: job.PluginId, + ControllerId: server.PrivateId, + Status: Running.string(), }, }, nextRunIn: time.Hour, @@ -521,10 +521,10 @@ func TestRepository_CompleteRun(t *testing.T) { name: "valid-with-progress", orig: &Run{ JobRun: &store.JobRun{ - JobName: job.Name, - JobPluginId: job.PluginId, - ServerId: server.PrivateId, - Status: Running.string(), + JobName: job.Name, + JobPluginId: job.PluginId, + ControllerId: server.PrivateId, + Status: Running.string(), }, }, args: args{completed: 10, total: 20}, @@ -631,10 +631,10 @@ func TestRepository_FailRun(t *testing.T) { name: "status-already-interrupted", orig: &Run{ JobRun: &store.JobRun{ - JobName: job.Name, - JobPluginId: job.PluginId, - ServerId: server.PrivateId, - Status: Interrupted.string(), + JobName: job.Name, + JobPluginId: job.PluginId, + ControllerId: server.PrivateId, + Status: Interrupted.string(), }, }, wantErr: true, @@ -645,10 +645,10 @@ func TestRepository_FailRun(t *testing.T) { name: "status-already-failed", orig: &Run{ JobRun: &store.JobRun{ - JobName: job.Name, - JobPluginId: job.PluginId, - ServerId: server.PrivateId, - Status: Failed.string(), + JobName: job.Name, + JobPluginId: job.PluginId, + ControllerId: server.PrivateId, + Status: Failed.string(), }, }, wantErr: true, @@ -659,10 +659,10 @@ func TestRepository_FailRun(t *testing.T) { name: "status-already-completed", orig: &Run{ JobRun: &store.JobRun{ - JobName: job.Name, - JobPluginId: job.PluginId, - ServerId: server.PrivateId, - Status: Completed.string(), + JobName: job.Name, + JobPluginId: job.PluginId, + ControllerId: server.PrivateId, + Status: Completed.string(), }, }, wantErr: true, @@ -673,10 +673,10 @@ func TestRepository_FailRun(t *testing.T) { name: "valid", orig: &Run{ JobRun: &store.JobRun{ - JobName: job.Name, - JobPluginId: job.PluginId, - ServerId: server.PrivateId, - Status: Running.string(), + JobName: job.Name, + JobPluginId: job.PluginId, + ControllerId: server.PrivateId, + Status: Running.string(), }, }, }, @@ -684,10 +684,10 @@ func TestRepository_FailRun(t *testing.T) { name: "valid-with-progress", orig: &Run{ JobRun: &store.JobRun{ - JobName: job.Name, - JobPluginId: job.PluginId, - ServerId: server.PrivateId, - Status: Running.string(), + JobName: job.Name, + JobPluginId: job.PluginId, + ControllerId: server.PrivateId, + Status: Running.string(), }, }, args: args{completed: 10, total: 20}, @@ -859,7 +859,7 @@ func TestRepository_InterruptServerRuns(t *testing.T) { job3 := testJob(t, conn, "job3", "description", wrapper) type args struct { - serverId string + ControllerId string opts []Option expectedJobs []*Job } @@ -872,7 +872,7 @@ func TestRepository_InterruptServerRuns(t *testing.T) { name: "all-runs", runs: []args{ { - serverId: server1.PrivateId, + ControllerId: server1.PrivateId, opts: []Option{WithRunJobsLimit(3)}, expectedJobs: []*Job{job1, job2, job3}, }, @@ -887,22 +887,22 @@ func TestRepository_InterruptServerRuns(t *testing.T) { name: "all-runs-on-single-server-with-server-id", runs: []args{ { - serverId: server2.PrivateId, + ControllerId: server2.PrivateId, opts: []Option{WithRunJobsLimit(3)}, expectedJobs: []*Job{job1, job2, job3}, }, }, interrupts: []args{ { - opts: []Option{WithServerId(server1.PrivateId)}, + opts: []Option{WithControllerId(server1.PrivateId)}, expectedJobs: []*Job{}, }, { - opts: []Option{WithServerId(server2.PrivateId)}, + opts: []Option{WithControllerId(server2.PrivateId)}, expectedJobs: []*Job{job1, job2, job3}, }, { - opts: []Option{WithServerId(server3.PrivateId)}, + opts: []Option{WithControllerId(server3.PrivateId)}, expectedJobs: []*Job{}, }, }, @@ -921,15 +921,15 @@ func TestRepository_InterruptServerRuns(t *testing.T) { runs: []args{}, interrupts: []args{ { - opts: []Option{WithServerId(server1.PrivateId)}, + opts: []Option{WithControllerId(server1.PrivateId)}, expectedJobs: []*Job{}, }, { - opts: []Option{WithServerId(server2.PrivateId)}, + opts: []Option{WithControllerId(server2.PrivateId)}, expectedJobs: []*Job{}, }, { - opts: []Option{WithServerId(server3.PrivateId)}, + opts: []Option{WithControllerId(server3.PrivateId)}, expectedJobs: []*Job{}, }, }, @@ -938,17 +938,17 @@ func TestRepository_InterruptServerRuns(t *testing.T) { name: "multiple-servers-interrupt-all", runs: []args{ { - serverId: server1.PrivateId, + ControllerId: server1.PrivateId, opts: []Option{WithRunJobsLimit(1)}, expectedJobs: []*Job{job1}, }, { - serverId: server2.PrivateId, + ControllerId: server2.PrivateId, opts: []Option{WithRunJobsLimit(1)}, expectedJobs: []*Job{job2}, }, { - serverId: server3.PrivateId, + ControllerId: server3.PrivateId, opts: []Option{WithRunJobsLimit(1)}, expectedJobs: []*Job{job3}, }, @@ -963,32 +963,32 @@ func TestRepository_InterruptServerRuns(t *testing.T) { name: "multiple-servers-with-server-id", runs: []args{ { - serverId: server1.PrivateId, + ControllerId: server1.PrivateId, opts: []Option{WithRunJobsLimit(1)}, expectedJobs: []*Job{job1}, }, { - serverId: server2.PrivateId, + ControllerId: server2.PrivateId, opts: []Option{WithRunJobsLimit(1)}, expectedJobs: []*Job{job2}, }, { - serverId: server3.PrivateId, + ControllerId: server3.PrivateId, opts: []Option{WithRunJobsLimit(1)}, expectedJobs: []*Job{job3}, }, }, interrupts: []args{ { - opts: []Option{WithServerId(server1.PrivateId)}, + opts: []Option{WithControllerId(server1.PrivateId)}, expectedJobs: []*Job{job1}, }, { - opts: []Option{WithServerId(server2.PrivateId)}, + opts: []Option{WithControllerId(server2.PrivateId)}, expectedJobs: []*Job{job2}, }, { - opts: []Option{WithServerId(server3.PrivateId)}, + opts: []Option{WithControllerId(server3.PrivateId)}, expectedJobs: []*Job{job3}, }, }, @@ -997,32 +997,32 @@ func TestRepository_InterruptServerRuns(t *testing.T) { name: "multiple-servers-distributed-runs", runs: []args{ { - serverId: server1.PrivateId, + ControllerId: server1.PrivateId, opts: []Option{WithRunJobsLimit(2)}, expectedJobs: []*Job{job1, job2}, }, { - serverId: server2.PrivateId, + ControllerId: server2.PrivateId, opts: []Option{WithRunJobsLimit(1)}, expectedJobs: []*Job{job3}, }, { - serverId: server3.PrivateId, + ControllerId: server3.PrivateId, opts: []Option{WithRunJobsLimit(1)}, expectedJobs: []*Job{}, }, }, interrupts: []args{ { - opts: []Option{WithServerId(server1.PrivateId)}, + opts: []Option{WithControllerId(server1.PrivateId)}, expectedJobs: []*Job{job1, job2}, }, { - opts: []Option{WithServerId(server2.PrivateId)}, + opts: []Option{WithControllerId(server2.PrivateId)}, expectedJobs: []*Job{job3}, }, { - opts: []Option{WithServerId(server3.PrivateId)}, + opts: []Option{WithControllerId(server3.PrivateId)}, expectedJobs: []*Job{}, }, }, @@ -1031,17 +1031,17 @@ func TestRepository_InterruptServerRuns(t *testing.T) { name: "multiple-servers-distributed-runs-interrupt-all", runs: []args{ { - serverId: server1.PrivateId, + ControllerId: server1.PrivateId, opts: []Option{WithRunJobsLimit(2)}, expectedJobs: []*Job{job1, job2}, }, { - serverId: server2.PrivateId, + ControllerId: server2.PrivateId, opts: []Option{WithRunJobsLimit(1)}, expectedJobs: []*Job{job3}, }, { - serverId: server3.PrivateId, + ControllerId: server3.PrivateId, opts: []Option{WithRunJobsLimit(1)}, expectedJobs: []*Job{}, }, @@ -1060,7 +1060,7 @@ func TestRepository_InterruptServerRuns(t *testing.T) { require.NoError(err) for _, r := range tt.runs { - runs, err := repo.RunJobs(context.Background(), r.serverId, r.opts...) + runs, err := repo.RunJobs(context.Background(), r.ControllerId, r.opts...) require.NoError(err) assert.Len(runs, len(r.expectedJobs)) sort.Slice(runs, func(i, j int) bool { return runs[i].JobName < runs[j].JobName }) diff --git a/internal/scheduler/job/store/job.pb.go b/internal/scheduler/job/store/job.pb.go index 77c96a057..e04d18c12 100644 --- a/internal/scheduler/job/store/job.pb.go +++ b/internal/scheduler/job/store/job.pb.go @@ -116,9 +116,6 @@ type JobRun struct { // The job_name is the name of the job and must be set. // @inject_tag: `gorm:"not_null"` JobName string `protobuf:"bytes,3,opt,name=job_name,json=jobName,proto3" json:"job_name,omitempty" gorm:"not_null"` - // The server_id of the server running the job and must be set. - // @inject_tag: `gorm:"not_null"` - ServerId string `protobuf:"bytes,4,opt,name=server_id,json=serverId,proto3" json:"server_id,omitempty" gorm:"not_null"` // The create_time is set by the database. // @inject_tag: `gorm:"default:current_timestamp"` CreateTime *timestamp.Timestamp `protobuf:"bytes,5,opt,name=create_time,json=createTime,proto3" json:"create_time,omitempty" gorm:"default:current_timestamp"` @@ -137,6 +134,9 @@ type JobRun struct { // status of the job run (running, completed, failed or interrupted). // @inject_tag: `gorm:"not_null"` Status string `protobuf:"bytes,10,opt,name=status,proto3" json:"status,omitempty" gorm:"not_null"` + // The controller_id of the controller running the job and must be set. + // @inject_tag: `gorm:"not_null"` + ControllerId string `protobuf:"bytes,11,opt,name=controller_id,json=controllerId,proto3" json:"controller_id,omitempty" gorm:"not_null"` } func (x *JobRun) Reset() { @@ -192,13 +192,6 @@ func (x *JobRun) GetJobName() string { return "" } -func (x *JobRun) GetServerId() string { - if x != nil { - return x.ServerId - } - return "" -} - func (x *JobRun) GetCreateTime() *timestamp.Timestamp { if x != nil { return x.CreateTime @@ -241,6 +234,13 @@ func (x *JobRun) GetStatus() string { return "" } +func (x *JobRun) GetControllerId() string { + if x != nil { + return x.ControllerId + } + return "" +} + var File_controller_storage_job_store_v1_job_proto protoreflect.FileDescriptor var file_controller_storage_job_store_v1_job_proto_rawDesc = []byte{ @@ -263,40 +263,41 @@ var file_controller_storage_job_store_v1_job_proto_rawDesc = []byte{ 0x72, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e, 0x76, 0x31, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x10, 0x6e, 0x65, 0x78, 0x74, 0x53, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x64, 0x52, - 0x75, 0x6e, 0x22, 0xc6, 0x03, 0x0a, 0x06, 0x4a, 0x6f, 0x62, 0x52, 0x75, 0x6e, 0x12, 0x1d, 0x0a, + 0x75, 0x6e, 0x22, 0xd9, 0x03, 0x0a, 0x06, 0x4a, 0x6f, 0x62, 0x52, 0x75, 0x6e, 0x12, 0x1d, 0x0a, 0x0a, 0x70, 0x72, 0x69, 0x76, 0x61, 0x74, 0x65, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x70, 0x72, 0x69, 0x76, 0x61, 0x74, 0x65, 0x49, 0x64, 0x12, 0x22, 0x0a, 0x0d, 0x6a, 0x6f, 0x62, 0x5f, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x5f, 0x69, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0b, 0x6a, 0x6f, 0x62, 0x50, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x49, 0x64, 0x12, 0x19, 0x0a, 0x08, 0x6a, 0x6f, 0x62, 0x5f, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x03, 0x20, 0x01, - 0x28, 0x09, 0x52, 0x07, 0x6a, 0x6f, 0x62, 0x4e, 0x61, 0x6d, 0x65, 0x12, 0x1b, 0x0a, 0x09, 0x73, - 0x65, 0x72, 0x76, 0x65, 0x72, 0x5f, 0x69, 0x64, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, - 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x49, 0x64, 0x12, 0x4b, 0x0a, 0x0b, 0x63, 0x72, 0x65, 0x61, - 0x74, 0x65, 0x5f, 0x74, 0x69, 0x6d, 0x65, 0x18, 0x05, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x2a, 0x2e, + 0x28, 0x09, 0x52, 0x07, 0x6a, 0x6f, 0x62, 0x4e, 0x61, 0x6d, 0x65, 0x12, 0x4b, 0x0a, 0x0b, 0x63, + 0x72, 0x65, 0x61, 0x74, 0x65, 0x5f, 0x74, 0x69, 0x6d, 0x65, 0x18, 0x05, 0x20, 0x01, 0x28, 0x0b, + 0x32, 0x2a, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x74, + 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e, + 0x76, 0x31, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x0a, 0x63, 0x72, + 0x65, 0x61, 0x74, 0x65, 0x54, 0x69, 0x6d, 0x65, 0x12, 0x4b, 0x0a, 0x0b, 0x75, 0x70, 0x64, 0x61, + 0x74, 0x65, 0x5f, 0x74, 0x69, 0x6d, 0x65, 0x18, 0x06, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x2a, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e, 0x76, 0x31, 0x2e, - 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x0a, 0x63, 0x72, 0x65, 0x61, 0x74, - 0x65, 0x54, 0x69, 0x6d, 0x65, 0x12, 0x4b, 0x0a, 0x0b, 0x75, 0x70, 0x64, 0x61, 0x74, 0x65, 0x5f, - 0x74, 0x69, 0x6d, 0x65, 0x18, 0x06, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x2a, 0x2e, 0x63, 0x6f, 0x6e, - 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, - 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e, 0x76, 0x31, 0x2e, 0x54, 0x69, 0x6d, - 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x0a, 0x75, 0x70, 0x64, 0x61, 0x74, 0x65, 0x54, 0x69, - 0x6d, 0x65, 0x12, 0x45, 0x0a, 0x08, 0x65, 0x6e, 0x64, 0x5f, 0x74, 0x69, 0x6d, 0x65, 0x18, 0x07, - 0x20, 0x01, 0x28, 0x0b, 0x32, 0x2a, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, - 0x72, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, - 0x61, 0x6d, 0x70, 0x2e, 0x76, 0x31, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, - 0x52, 0x07, 0x65, 0x6e, 0x64, 0x54, 0x69, 0x6d, 0x65, 0x12, 0x27, 0x0a, 0x0f, 0x63, 0x6f, 0x6d, - 0x70, 0x6c, 0x65, 0x74, 0x65, 0x64, 0x5f, 0x63, 0x6f, 0x75, 0x6e, 0x74, 0x18, 0x08, 0x20, 0x01, - 0x28, 0x0d, 0x52, 0x0e, 0x63, 0x6f, 0x6d, 0x70, 0x6c, 0x65, 0x74, 0x65, 0x64, 0x43, 0x6f, 0x75, - 0x6e, 0x74, 0x12, 0x1f, 0x0a, 0x0b, 0x74, 0x6f, 0x74, 0x61, 0x6c, 0x5f, 0x63, 0x6f, 0x75, 0x6e, - 0x74, 0x18, 0x09, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0a, 0x74, 0x6f, 0x74, 0x61, 0x6c, 0x43, 0x6f, - 0x75, 0x6e, 0x74, 0x12, 0x16, 0x0a, 0x06, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x18, 0x0a, 0x20, - 0x01, 0x28, 0x09, 0x52, 0x06, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x42, 0x42, 0x5a, 0x40, 0x67, - 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, - 0x6f, 0x72, 0x70, 0x2f, 0x62, 0x6f, 0x75, 0x6e, 0x64, 0x61, 0x72, 0x79, 0x2f, 0x69, 0x6e, 0x74, - 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x73, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x2f, - 0x6a, 0x6f, 0x62, 0x2f, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x3b, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x62, - 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, + 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x0a, 0x75, 0x70, 0x64, 0x61, 0x74, + 0x65, 0x54, 0x69, 0x6d, 0x65, 0x12, 0x45, 0x0a, 0x08, 0x65, 0x6e, 0x64, 0x5f, 0x74, 0x69, 0x6d, + 0x65, 0x18, 0x07, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x2a, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, + 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, 0x74, 0x69, 0x6d, + 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e, 0x76, 0x31, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, + 0x61, 0x6d, 0x70, 0x52, 0x07, 0x65, 0x6e, 0x64, 0x54, 0x69, 0x6d, 0x65, 0x12, 0x27, 0x0a, 0x0f, + 0x63, 0x6f, 0x6d, 0x70, 0x6c, 0x65, 0x74, 0x65, 0x64, 0x5f, 0x63, 0x6f, 0x75, 0x6e, 0x74, 0x18, + 0x08, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0e, 0x63, 0x6f, 0x6d, 0x70, 0x6c, 0x65, 0x74, 0x65, 0x64, + 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x12, 0x1f, 0x0a, 0x0b, 0x74, 0x6f, 0x74, 0x61, 0x6c, 0x5f, 0x63, + 0x6f, 0x75, 0x6e, 0x74, 0x18, 0x09, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0a, 0x74, 0x6f, 0x74, 0x61, + 0x6c, 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x12, 0x16, 0x0a, 0x06, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, + 0x18, 0x0a, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x12, 0x23, + 0x0a, 0x0d, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x5f, 0x69, 0x64, 0x18, + 0x0b, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0c, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, + 0x72, 0x49, 0x64, 0x52, 0x09, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x5f, 0x69, 0x64, 0x42, 0x42, + 0x5a, 0x40, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x68, 0x61, 0x73, + 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2f, 0x62, 0x6f, 0x75, 0x6e, 0x64, 0x61, 0x72, 0x79, 0x2f, + 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x73, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, + 0x65, 0x72, 0x2f, 0x6a, 0x6f, 0x62, 0x2f, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x3b, 0x73, 0x74, 0x6f, + 0x72, 0x65, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, } var ( diff --git a/internal/scheduler/job/testing.go b/internal/scheduler/job/testing.go index c1c43450b..9d00670e9 100644 --- a/internal/scheduler/job/testing.go +++ b/internal/scheduler/job/testing.go @@ -35,7 +35,7 @@ func testJob(t testing.TB, conn *db.DB, name, description string, wrapper wrappi func testRun(conn *db.DB, pluginId, name, cId string) (*Run, error) { query := ` insert into job_run ( - job_plugin_id, job_name, server_id + job_plugin_id, job_name, controller_id ) values (?,?,?) on conflict (job_plugin_id, job_name) where status = 'running' @@ -65,7 +65,7 @@ func testRun(conn *db.DB, pluginId, name, cId string) (*Run, error) { func testRunWithUpdateTime(conn *db.DB, pluginId, name, cId string, updateTime time.Time) (*Run, error) { query := ` insert into job_run ( - job_plugin_id, job_name, server_id, update_time + job_plugin_id, job_name, controller_id, update_time ) values (?,?,?,?) returning *; diff --git a/internal/scheduler/job/testing_test.go b/internal/scheduler/job/testing_test.go index cd3e6ddd4..4abd3f74a 100644 --- a/internal/scheduler/job/testing_test.go +++ b/internal/scheduler/job/testing_test.go @@ -68,7 +68,7 @@ func Test_TestRun(t *testing.T) { var got Run err = rw.LookupWhere(context.Background(), &got, "private_id = ?", []interface{}{run.PrivateId}) require.NoError(err) - assert.Equal(server.PrivateId, got.ServerId) + assert.Equal(server.PrivateId, got.ControllerId) assert.Equal(job.Name, got.JobName) assert.Equal(job.PluginId, got.JobPluginId) assert.NotEmpty(got.CreateTime) diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go index 313a2ce55..7b6d8371d 100644 --- a/internal/scheduler/scheduler.go +++ b/internal/scheduler/scheduler.go @@ -154,7 +154,7 @@ func (s *Scheduler) Start(ctx context.Context, wg *sync.WaitGroup) error { } // Interrupt all runs previously allocated to this server - _, err = repo.InterruptRuns(ctx, 0, job.WithServerId(s.serverId)) + _, err = repo.InterruptRuns(ctx, 0, job.WithControllerId(s.serverId)) if err != nil { return errors.Wrap(ctx, err, op) } diff --git a/internal/servers/gorm.go b/internal/servers/gorm.go deleted file mode 100644 index 7588948f6..000000000 --- a/internal/servers/gorm.go +++ /dev/null @@ -1,6 +0,0 @@ -package servers - -// TableName overrides the table name used by WorkerTag to `worker_tag` -func (WorkerTag) TableName() string { - return "server_worker_tag" -} diff --git a/internal/servers/options.go b/internal/servers/options.go index f82f5a52d..d58d9dd84 100644 --- a/internal/servers/options.go +++ b/internal/servers/options.go @@ -1,6 +1,8 @@ package servers -import "time" +import ( + "time" +) // getOpts - iterate the inbound Options and return a struct func getOpts(opt ...Option) options { @@ -16,15 +18,45 @@ type Option func(*options) // options = how options are represented type options struct { - withLimit int - withLiveness time.Duration - withUpdateTags bool + withName string + withDescription string + withAddress string + withPublicId string + withLimit int + withLiveness time.Duration + withUpdateTags bool + withWorkerTags []*Tag } func getDefaultOptions() options { - return options{ - withLimit: 0, - withLiveness: 0, + return options{} +} + +// WithDescription provides an optional description. +func WithDescription(desc string) Option { + return func(o *options) { + o.withDescription = desc + } +} + +// WithName provides an optional name. +func WithName(name string) Option { + return func(o *options) { + o.withName = name + } +} + +// WithAddress provides an optional address. +func WithAddress(address string) Option { + return func(o *options) { + o.withAddress = address + } +} + +// WithPublicId provides an optional public id +func WithPublicId(id string) Option { + return func(o *options) { + o.withPublicId = id } } @@ -54,3 +86,10 @@ func WithUpdateTags(updateTags bool) Option { o.withUpdateTags = updateTags } } + +// WithWorkerTags provides worker tags. +func WithWorkerTags(tags ...*Tag) Option { + return func(o *options) { + o.withWorkerTags = tags + } +} diff --git a/internal/servers/options_test.go b/internal/servers/options_test.go index 6ceedfc52..05b9d4070 100644 --- a/internal/servers/options_test.go +++ b/internal/servers/options_test.go @@ -13,15 +13,56 @@ func Test_GetOpts(t *testing.T) { opts := getOpts() assert.Equal(t, options{}, opts) - opts = getOpts( - WithLimit(5), - WithLiveness(time.Hour), - WithUpdateTags(true), - ) - exp := options{ - withLimit: 5, - withLiveness: time.Hour, - withUpdateTags: true, - } - assert.Equal(t, exp, opts) + t.Run("WithName", func(t *testing.T) { + opts := getOpts(WithName("test")) + testOpts := getDefaultOptions() + testOpts.withName = "test" + assert.Equal(t, opts, testOpts) + }) + t.Run("WithDescription", func(t *testing.T) { + opts := getOpts(WithDescription("test desc")) + testOpts := getDefaultOptions() + testOpts.withDescription = "test desc" + assert.Equal(t, opts, testOpts) + }) + t.Run("WithLimit", func(t *testing.T) { + opts := getOpts(WithLimit(5)) + testOpts := getDefaultOptions() + testOpts.withLimit = 5 + assert.Equal(t, opts, testOpts) + }) + t.Run("WithAddress", func(t *testing.T) { + opts := getOpts(WithAddress("test")) + testOpts := getDefaultOptions() + testOpts.withAddress = "test" + assert.Equal(t, opts, testOpts) + }) + t.Run("WithPublicId", func(t *testing.T) { + opts := getOpts(WithPublicId("test")) + testOpts := getDefaultOptions() + testOpts.withPublicId = "test" + assert.Equal(t, opts, testOpts) + }) + t.Run("WithLiveness", func(t *testing.T) { + opts := getOpts(WithLiveness(time.Hour)) + testOpts := getDefaultOptions() + testOpts.withLiveness = time.Hour + assert.Equal(t, opts, testOpts) + }) + t.Run("WithUpdateTags", func(t *testing.T) { + opts := getOpts(WithUpdateTags(true)) + testOpts := getDefaultOptions() + testOpts.withUpdateTags = true + assert.Equal(t, opts, testOpts) + }) + t.Run("WithWorkerTags", func(t *testing.T) { + tags := []*Tag{ + {Key: "key1", Value: "val1"}, + {Key: "key2", Value: "val2"}, + } + opts := getOpts(WithWorkerTags(tags...)) + testOpts := getDefaultOptions() + testOpts.withWorkerTags = tags + assert.Equal(t, opts, testOpts) + }) } diff --git a/internal/servers/public_ids.go b/internal/servers/public_ids.go new file mode 100644 index 000000000..16d9a7829 --- /dev/null +++ b/internal/servers/public_ids.go @@ -0,0 +1,21 @@ +package servers + +import ( + "context" + + "github.com/hashicorp/boundary/internal/db" + "github.com/hashicorp/boundary/internal/errors" +) + +// PublicId prefixes for the resources in the servers package. +const ( + WorkerPrefix = "w" +) + +func newWorkerId(ctx context.Context) (string, error) { + id, err := db.NewPublicId(WorkerPrefix) + if err != nil { + return "", errors.Wrap(ctx, err, "servers.newWorkerId") + } + return id, nil +} diff --git a/internal/servers/repository.go b/internal/servers/repository.go index b089d3b16..908a6beee 100644 --- a/internal/servers/repository.go +++ b/internal/servers/repository.go @@ -24,17 +24,6 @@ const ( DefaultLiveness = 15 * time.Second ) -type ServerType string - -const ( - ServerTypeController ServerType = "controller" - ServerTypeWorker ServerType = "worker" -) - -func (s ServerType) String() string { - return string(s) -} - // Repository is the servers database repository type Repository struct { reader db.Reader @@ -63,13 +52,13 @@ func NewRepository(r db.Reader, w db.Writer, kms *kms.Kms) (*Repository, error) } // ListWorkers is a passthrough to listWorkersWithReader that uses the repo's normal reader. -func (r *Repository) ListWorkers(ctx context.Context, opt ...Option) ([]*store.Worker, error) { +func (r *Repository) ListWorkers(ctx context.Context, opt ...Option) ([]*Worker, error) { return r.listWorkersWithReader(ctx, r.reader, opt...) } // listWorkersWithReader will return a listing of resources and honor the WithLimit option or the repo // defaultLimit. It accepts a reader, allowing it to be used within a transaction or without. -func (r *Repository) listWorkersWithReader(ctx context.Context, reader db.Reader, opt ...Option) ([]*store.Worker, error) { +func (r *Repository) listWorkersWithReader(ctx context.Context, reader db.Reader, opt ...Option) ([]*Worker, error) { opts := getOpts(opt...) liveness := opts.withLiveness if liveness == 0 { @@ -81,7 +70,7 @@ func (r *Repository) listWorkersWithReader(ctx context.Context, reader db.Reader where = fmt.Sprintf("update_time > now() - interval '%d seconds'", uint32(liveness.Seconds())) } - var workers []*store.Worker + var workers []*Worker if err := reader.SearchWhere( ctx, &workers, @@ -128,16 +117,10 @@ func (r *Repository) listControllersWithReader(ctx context.Context, reader db.Re return controllers, nil } -// WorkerTag holds the information for the worker_tag table for Gorm. -type WorkerTag struct { - WorkerId string - Key string - Value string -} - -// ListTagsForWorkers pulls out tag tuples into WorkerTag structs for the given worker -func (r *Repository) ListTagsForWorkers(ctx context.Context, workerIds []string, opt ...Option) ([]*WorkerTag, error) { - var workerTags []*WorkerTag +// ListTagsForWorkers returns a map from the worker's id to the list of Tags +// that are for that worker. All options are ignored. +func (r *Repository) ListTagsForWorkers(ctx context.Context, workerIds []string, opt ...Option) (map[string][]*Tag, error) { + var workerTags []*store.WorkerTag if err := r.reader.SearchWhere( ctx, &workerTags, @@ -147,14 +130,29 @@ func (r *Repository) ListTagsForWorkers(ctx context.Context, workerIds []string, ); err != nil { return nil, errors.Wrap(ctx, err, "servers.ListTagsForWorkers", errors.WithMsg(fmt.Sprintf("worker IDs %v", workerIds))) } - return workerTags, nil + + ret := make(map[string][]*Tag, len(workerIds)) + for _, t := range workerTags { + ret[t.WorkerId] = append(ret[t.WorkerId], &Tag{Key: t.Key, Value: t.Value}) + } + return ret, nil } -func (r *Repository) UpsertWorker(ctx context.Context, worker *store.Worker, opt ...Option) ([]*store.Controller, int, error) { +// UpsertWorker creates a new worker if one with the provided public id doesn't +// already exist. If it does, UpsertWorker updates the worker. The +// WithUpdateTags option is the only one used. All others are ignored. +func (r *Repository) UpsertWorker(ctx context.Context, worker *Worker, opt ...Option) ([]*store.Controller, int, error) { const op = "servers.UpsertWorker" - if worker == nil { + switch { + case worker == nil: return nil, db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "worker is nil") + case worker.Address == "": + return nil, db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "address is empty") + case worker.ScopeId == "": + return nil, db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "scope id is empty") + case worker.PublicId == "": + return nil, db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "public id is empty") } opts := getOpts(opt...) @@ -187,9 +185,9 @@ func (r *Repository) UpsertWorker(ctx context.Context, worker *store.Worker, opt // delete all tags for the given worker, then add the new ones // we've been sent. if opts.withUpdateTags { - _, err = w.Delete(ctx, &WorkerTag{}, db.WithWhere(deleteTagsSql, worker.PublicId)) + _, err = w.Delete(ctx, &store.WorkerTag{}, db.WithWhere(deleteTagsSql, worker.GetPublicId())) if err != nil { - return errors.Wrap(ctx, err, op+":DeleteTags", errors.WithMsg(worker.PublicId)) + return errors.Wrap(ctx, err, op+":DeleteTags", errors.WithMsg(worker.GetPublicId())) } // If tags were cleared out entirely, then we'll have nothing @@ -198,20 +196,18 @@ func (r *Repository) UpsertWorker(ctx context.Context, worker *store.Worker, opt // below. if len(worker.Tags) > 0 { tags := make([]interface{}, 0, len(worker.Tags)) - for k, v := range worker.Tags { + for _, v := range worker.Tags { if v == nil { - return errors.New(ctx, errors.InvalidParameter, op+":RangeTags", fmt.Sprintf("found nil tag value for worker %s and key %s", worker.PublicId, k)) - } - for _, val := range v.Values { - tags = append(tags, WorkerTag{ - WorkerId: worker.PublicId, - Key: k, - Value: val, - }) + return errors.New(ctx, errors.InvalidParameter, op+":RangeTags", fmt.Sprintf("found nil tag value for worker %s", worker.GetPublicId())) } + tags = append(tags, &store.WorkerTag{ + WorkerId: worker.GetPublicId(), + Key: v.Key, + Value: v.Value, + }) } if err = w.CreateItems(ctx, tags); err != nil { - return errors.Wrap(ctx, err, op+":CreateTags", errors.WithMsg(worker.PublicId)) + return errors.Wrap(ctx, err, op+":CreateTags", errors.WithMsg(worker.GetPublicId())) } } } diff --git a/internal/servers/repository_test.go b/internal/servers/repository_test.go index 173b2169e..8f46ebe7e 100644 --- a/internal/servers/repository_test.go +++ b/internal/servers/repository_test.go @@ -5,8 +5,6 @@ import ( "testing" "time" - "github.com/hashicorp/boundary/internal/servers/store" - "github.com/hashicorp/boundary/api/recovery" "github.com/hashicorp/boundary/api/roles" "github.com/hashicorp/boundary/globals" @@ -99,28 +97,34 @@ func TestTagUpdatingListing(t *testing.T) { defer tc.Shutdown() repo := tc.ServersRepo() - srv := &store.Worker{ - PublicId: "test_worker_1", - Address: "127.0.0.1", - Tags: map[string]*store.TagValues{ - "tag1": { - Values: []string{"value1", "value2"}, + srv := servers.NewWorker(scope.Global.String(), + servers.WithPublicId("test_worker_1"), + servers.WithAddress("127.0.0.1"), + servers.WithWorkerTags( + &servers.Tag{ + Key: "tag1", + Value: "value1", }, - }, - } + &servers.Tag{ + Key: "tag1", + Value: "value2", + })) _, _, err := repo.UpsertWorker(tc.Context(), srv, servers.WithUpdateTags(true)) require.NoError(err) - srv = &store.Worker{ - PublicId: "test_worker_2", - Address: "127.0.0.1", - Tags: map[string]*store.TagValues{ - "tag2": { - Values: []string{"value1", "value2"}, + srv = servers.NewWorker(scope.Global.String(), + servers.WithPublicId("test_worker_2"), + servers.WithAddress("127.0.0.1"), + servers.WithWorkerTags( + &servers.Tag{ + Key: "tag2", + Value: "value1", }, - }, - } + &servers.Tag{ + Key: "tag2", + Value: "value2", + })) _, _, err = repo.UpsertWorker(tc.Context(), srv, servers.WithUpdateTags(true)) require.NoError(err) @@ -128,40 +132,42 @@ func TestTagUpdatingListing(t *testing.T) { require.NoError(err) // Base case - exp := []*servers.WorkerTag{ - { - WorkerId: "test_worker_1", - Key: "tag1", - Value: "value1", + exp := map[string][]*servers.Tag{ + "test_worker_1": { + { + Key: "tag1", + Value: "value1", + }, { + Key: "tag1", + Value: "value2", + }, }, - { - WorkerId: "test_worker_1", - Key: "tag1", - Value: "value2", - }, - { - WorkerId: "test_worker_2", - Key: "tag2", - Value: "value1", - }, - { - WorkerId: "test_worker_2", - Key: "tag2", - Value: "value2", + "test_worker_2": { + { + Key: "tag2", + Value: "value1", + }, + { + Key: "tag2", + Value: "value2", + }, }, } require.Equal(exp, tags) // Update without saying to update tags - srv = &store.Worker{ - PublicId: "test_worker_2", - Address: "192.168.1.1", - Tags: map[string]*store.TagValues{ - "tag22": { - Values: []string{"value21", "value22"}, + srv = servers.NewWorker(scope.Global.String(), + servers.WithPublicId("test_worker_2"), + servers.WithAddress("192.168.1.1"), + servers.WithWorkerTags( + &servers.Tag{ + Key: "tag22", + Value: "value21", }, - }, - } + &servers.Tag{ + Key: "tag22", + Value: "value22", + })) _, _, err = repo.UpsertWorker(tc.Context(), srv) require.NoError(err) tags, err = repo.ListTagsForWorkers(tc.Context(), []string{"test_worker_1", "test_worker_2"}) @@ -175,15 +181,15 @@ func TestTagUpdatingListing(t *testing.T) { require.NoError(err) require.NotEqual(exp, tags) // Update and try again - exp[2] = &servers.WorkerTag{ - WorkerId: "test_worker_2", - Key: "tag22", - Value: "value21", - } - exp[3] = &servers.WorkerTag{ - WorkerId: "test_worker_2", - Key: "tag22", - Value: "value22", + exp["test_worker_2"] = []*servers.Tag{ + { + Key: "tag22", + Value: "value21", + }, + { + Key: "tag22", + Value: "value22", + }, } require.Equal(exp, tags) } @@ -199,11 +205,10 @@ func TestListServersWithLiveness(t *testing.T) { require.NoError(err) ctx := context.Background() - newWorker := func(publicId string) *store.Worker { - result := &store.Worker{ - PublicId: publicId, - Address: "127.0.0.1", - } + newWorker := func(publicId string) *servers.Worker { + result := servers.NewWorker(scope.Global.String(), + servers.WithPublicId(publicId), + servers.WithAddress("127.0.0.1")) _, rowsUpdated, err := serversRepo.UpsertWorker(ctx, result) require.NoError(err) require.Equal(1, rowsUpdated) @@ -224,7 +229,7 @@ func TestListServersWithLiveness(t *testing.T) { require.NoError(err) require.Equal(1, rowsUpdated) - requireIds := func(expected []string, actual []*store.Worker) { + requireIds := func(expected []string, actual []*servers.Worker) { require.Len(expected, len(actual)) want := make(map[string]struct{}) for _, v := range expected { diff --git a/internal/servers/store/gorm.go b/internal/servers/store/gorm.go index 7ab3fc709..2da1ceae9 100644 --- a/internal/servers/store/gorm.go +++ b/internal/servers/store/gorm.go @@ -1,11 +1,11 @@ package store -// TableName overrides the table name used by Worker to `server_worker` -func (Worker) TableName() string { - return "server_worker" -} - // TableName overrides the table name used by Controller to `server_controller` func (Controller) TableName() string { return "server_controller" } + +// TableName overrides the table name used by WorkerTag to `worker_tag` +func (WorkerTag) TableName() string { + return "server_worker_tag" +} diff --git a/internal/servers/store/worker.pb.go b/internal/servers/store/worker.pb.go index 4f78c2685..3ad103fc6 100644 --- a/internal/servers/store/worker.pb.go +++ b/internal/servers/store/worker.pb.go @@ -27,24 +27,30 @@ type Worker struct { sizeCache protoimpl.SizeCache unknownFields protoimpl.UnknownFields - // Public ID of the resource + // public_id is a surrogate key suitable for use in a public API // @inject_tag: `gorm:"primary_key"` PublicId string `protobuf:"bytes,10,opt,name=public_id,json=publicId,proto3" json:"public_id,omitempty" gorm:"primary_key"` // Name of the resource (optional) - // Description of the resource (optional) // @inject_tag: `gorm:"default:null"` Name string `protobuf:"bytes,20,opt,name=name,proto3" json:"name,omitempty" gorm:"default:null"` // Description of the resource (optional) // @inject_tag: `gorm:"default:null"` Description string `protobuf:"bytes,30,opt,name=description,proto3" json:"description,omitempty" gorm:"default:null"` - // Address for the worker - Address string `protobuf:"bytes,40,opt,name=address,proto3" json:"address,omitempty"` - // First seen time from the RDBMS - CreateTime *timestamp.Timestamp `protobuf:"bytes,50,opt,name=create_time,json=createTime,proto3" json:"create_time,omitempty"` - // Last time there was an update - UpdateTime *timestamp.Timestamp `protobuf:"bytes,60,opt,name=update_time,json=updateTime,proto3" json:"update_time,omitempty"` - // @gotags: `gorm:"-"` - Tags map[string]*TagValues `protobuf:"bytes,70,rep,name=tags,proto3" json:"tags,omitempty" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3" gorm:"-"` + // Address for the worker. This must be set + // @inject_tag: `gorm:"default:null"` + Address string `protobuf:"bytes,40,opt,name=address,proto3" json:"address,omitempty" gorm:"default:null"` + // The create_time is set by the database. + // @inject_tag: `gorm:"default:current_timestamp"` + CreateTime *timestamp.Timestamp `protobuf:"bytes,50,opt,name=create_time,json=createTime,proto3" json:"create_time,omitempty" gorm:"default:current_timestamp"` + // The update_time is set by the database. + // @inject_tag: `gorm:"default:current_timestamp"` + UpdateTime *timestamp.Timestamp `protobuf:"bytes,60,opt,name=update_time,json=updateTime,proto3" json:"update_time,omitempty" gorm:"default:current_timestamp"` + // The scope_id of the owning scope and must be set. + // @inject_tag: `gorm:"not_null"` + ScopeId string `protobuf:"bytes,70,opt,name=scope_id,json=scopeId,proto3" json:"scope_id,omitempty" gorm:"not_null"` + // version allows optimistic locking of the resource. + // @inject_tag: `gorm:"default:null"` + Version uint32 `protobuf:"varint,80,opt,name=version,proto3" json:"version,omitempty" gorm:"default:null"` } func (x *Worker) Reset() { @@ -121,25 +127,40 @@ func (x *Worker) GetUpdateTime() *timestamp.Timestamp { return nil } -func (x *Worker) GetTags() map[string]*TagValues { +func (x *Worker) GetScopeId() string { if x != nil { - return x.Tags + return x.ScopeId } - return nil + return "" } -// TagValues is used because map fields cannot be repeated but can be a -// message -type TagValues struct { +func (x *Worker) GetVersion() uint32 { + if x != nil { + return x.Version + } + return 0 +} + +// WorkerTag is a tag for a worker. The primary key is comprised of the +// worker_id, key, value, and source. +type WorkerTag struct { state protoimpl.MessageState sizeCache protoimpl.SizeCache unknownFields protoimpl.UnknownFields - Values []string `protobuf:"bytes,10,rep,name=values,proto3" json:"values,omitempty"` + // worker_id is the public key that key of the worker this tag is for. + // @inject_tag: `gorm:"default:null"` + WorkerId string `protobuf:"bytes,10,opt,name=worker_id,json=workerId,proto3" json:"worker_id,omitempty" gorm:"default:null"` + // key is the key of the tag. This must be set. + // @inject_tag: `gorm:"default:null"` + Key string `protobuf:"bytes,20,opt,name=key,proto3" json:"key,omitempty" gorm:"default:null"` + // value is the value + // @inject_tag: `gorm:"default:null"` + Value string `protobuf:"bytes,30,opt,name=value,proto3" json:"value,omitempty" gorm:"default:null"` } -func (x *TagValues) Reset() { - *x = TagValues{} +func (x *WorkerTag) Reset() { + *x = WorkerTag{} if protoimpl.UnsafeEnabled { mi := &file_controller_storage_servers_store_v1_worker_proto_msgTypes[1] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) @@ -147,13 +168,13 @@ func (x *TagValues) Reset() { } } -func (x *TagValues) String() string { +func (x *WorkerTag) String() string { return protoimpl.X.MessageStringOf(x) } -func (*TagValues) ProtoMessage() {} +func (*WorkerTag) ProtoMessage() {} -func (x *TagValues) ProtoReflect() protoreflect.Message { +func (x *WorkerTag) ProtoReflect() protoreflect.Message { mi := &file_controller_storage_servers_store_v1_worker_proto_msgTypes[1] if protoimpl.UnsafeEnabled && x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) @@ -165,16 +186,30 @@ func (x *TagValues) ProtoReflect() protoreflect.Message { return mi.MessageOf(x) } -// Deprecated: Use TagValues.ProtoReflect.Descriptor instead. -func (*TagValues) Descriptor() ([]byte, []int) { +// Deprecated: Use WorkerTag.ProtoReflect.Descriptor instead. +func (*WorkerTag) Descriptor() ([]byte, []int) { return file_controller_storage_servers_store_v1_worker_proto_rawDescGZIP(), []int{1} } -func (x *TagValues) GetValues() []string { +func (x *WorkerTag) GetWorkerId() string { if x != nil { - return x.Values + return x.WorkerId } - return nil + return "" +} + +func (x *WorkerTag) GetKey() string { + if x != nil { + return x.Key + } + return "" +} + +func (x *WorkerTag) GetValue() string { + if x != nil { + return x.Value + } + return "" } var File_controller_storage_servers_store_v1_worker_proto protoreflect.FileDescriptor @@ -188,7 +223,7 @@ var file_controller_storage_servers_store_v1_worker_proto_rawDesc = []byte{ 0x74, 0x6f, 0x72, 0x65, 0x2e, 0x76, 0x31, 0x1a, 0x2f, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2f, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2f, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2f, 0x76, 0x31, 0x2f, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, - 0x6d, 0x70, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0xc3, 0x03, 0x0a, 0x06, 0x57, 0x6f, 0x72, + 0x6d, 0x70, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0xc4, 0x02, 0x0a, 0x06, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x12, 0x1b, 0x0a, 0x09, 0x70, 0x75, 0x62, 0x6c, 0x69, 0x63, 0x5f, 0x69, 0x64, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x70, 0x75, 0x62, 0x6c, 0x69, 0x63, 0x49, 0x64, 0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x14, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, @@ -205,25 +240,20 @@ var file_controller_storage_servers_store_v1_worker_proto_rawDesc = []byte{ 0x28, 0x0b, 0x32, 0x2a, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e, 0x76, 0x31, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x0a, - 0x75, 0x70, 0x64, 0x61, 0x74, 0x65, 0x54, 0x69, 0x6d, 0x65, 0x12, 0x49, 0x0a, 0x04, 0x74, 0x61, - 0x67, 0x73, 0x18, 0x46, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x35, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, - 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, 0x73, 0x65, - 0x72, 0x76, 0x65, 0x72, 0x73, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x2e, 0x76, 0x31, 0x2e, 0x57, - 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x2e, 0x54, 0x61, 0x67, 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, - 0x04, 0x74, 0x61, 0x67, 0x73, 0x1a, 0x67, 0x0a, 0x09, 0x54, 0x61, 0x67, 0x73, 0x45, 0x6e, 0x74, - 0x72, 0x79, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, - 0x03, 0x6b, 0x65, 0x79, 0x12, 0x44, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, - 0x01, 0x28, 0x0b, 0x32, 0x2e, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, - 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, - 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x2e, 0x76, 0x31, 0x2e, 0x54, 0x61, 0x67, 0x56, 0x61, 0x6c, - 0x75, 0x65, 0x73, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x3a, 0x02, 0x38, 0x01, 0x22, 0x23, - 0x0a, 0x09, 0x54, 0x61, 0x67, 0x56, 0x61, 0x6c, 0x75, 0x65, 0x73, 0x12, 0x16, 0x0a, 0x06, 0x76, - 0x61, 0x6c, 0x75, 0x65, 0x73, 0x18, 0x0a, 0x20, 0x03, 0x28, 0x09, 0x52, 0x06, 0x76, 0x61, 0x6c, - 0x75, 0x65, 0x73, 0x42, 0x3c, 0x5a, 0x3a, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, - 0x6d, 0x2f, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2f, 0x62, 0x6f, 0x75, 0x6e, - 0x64, 0x61, 0x72, 0x79, 0x2f, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x73, 0x65, - 0x72, 0x76, 0x65, 0x72, 0x73, 0x2f, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x3b, 0x73, 0x74, 0x6f, 0x72, - 0x65, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, + 0x75, 0x70, 0x64, 0x61, 0x74, 0x65, 0x54, 0x69, 0x6d, 0x65, 0x12, 0x19, 0x0a, 0x08, 0x73, 0x63, + 0x6f, 0x70, 0x65, 0x5f, 0x69, 0x64, 0x18, 0x46, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x73, 0x63, + 0x6f, 0x70, 0x65, 0x49, 0x64, 0x12, 0x18, 0x0a, 0x07, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, + 0x18, 0x50, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x07, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x22, + 0x50, 0x0a, 0x09, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x54, 0x61, 0x67, 0x12, 0x1b, 0x0a, 0x09, + 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x5f, 0x69, 0x64, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x09, 0x52, + 0x08, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x49, 0x64, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, + 0x18, 0x14, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x14, 0x0a, 0x05, 0x76, + 0x61, 0x6c, 0x75, 0x65, 0x18, 0x1e, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, + 0x65, 0x42, 0x3c, 0x5a, 0x3a, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, + 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2f, 0x62, 0x6f, 0x75, 0x6e, 0x64, 0x61, + 0x72, 0x79, 0x2f, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x73, 0x65, 0x72, 0x76, + 0x65, 0x72, 0x73, 0x2f, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x3b, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x62, + 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, } var ( @@ -238,23 +268,20 @@ func file_controller_storage_servers_store_v1_worker_proto_rawDescGZIP() []byte return file_controller_storage_servers_store_v1_worker_proto_rawDescData } -var file_controller_storage_servers_store_v1_worker_proto_msgTypes = make([]protoimpl.MessageInfo, 3) +var file_controller_storage_servers_store_v1_worker_proto_msgTypes = make([]protoimpl.MessageInfo, 2) var file_controller_storage_servers_store_v1_worker_proto_goTypes = []interface{}{ (*Worker)(nil), // 0: controller.storage.servers.store.v1.Worker - (*TagValues)(nil), // 1: controller.storage.servers.store.v1.TagValues - nil, // 2: controller.storage.servers.store.v1.Worker.TagsEntry - (*timestamp.Timestamp)(nil), // 3: controller.storage.timestamp.v1.Timestamp + (*WorkerTag)(nil), // 1: controller.storage.servers.store.v1.WorkerTag + (*timestamp.Timestamp)(nil), // 2: controller.storage.timestamp.v1.Timestamp } var file_controller_storage_servers_store_v1_worker_proto_depIdxs = []int32{ - 3, // 0: controller.storage.servers.store.v1.Worker.create_time:type_name -> controller.storage.timestamp.v1.Timestamp - 3, // 1: controller.storage.servers.store.v1.Worker.update_time:type_name -> controller.storage.timestamp.v1.Timestamp - 2, // 2: controller.storage.servers.store.v1.Worker.tags:type_name -> controller.storage.servers.store.v1.Worker.TagsEntry - 1, // 3: controller.storage.servers.store.v1.Worker.TagsEntry.value:type_name -> controller.storage.servers.store.v1.TagValues - 4, // [4:4] is the sub-list for method output_type - 4, // [4:4] is the sub-list for method input_type - 4, // [4:4] is the sub-list for extension type_name - 4, // [4:4] is the sub-list for extension extendee - 0, // [0:4] is the sub-list for field type_name + 2, // 0: controller.storage.servers.store.v1.Worker.create_time:type_name -> controller.storage.timestamp.v1.Timestamp + 2, // 1: controller.storage.servers.store.v1.Worker.update_time:type_name -> controller.storage.timestamp.v1.Timestamp + 2, // [2:2] is the sub-list for method output_type + 2, // [2:2] is the sub-list for method input_type + 2, // [2:2] is the sub-list for extension type_name + 2, // [2:2] is the sub-list for extension extendee + 0, // [0:2] is the sub-list for field type_name } func init() { file_controller_storage_servers_store_v1_worker_proto_init() } @@ -276,7 +303,7 @@ func file_controller_storage_servers_store_v1_worker_proto_init() { } } file_controller_storage_servers_store_v1_worker_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} { - switch v := v.(*TagValues); i { + switch v := v.(*WorkerTag); i { case 0: return &v.state case 1: @@ -294,7 +321,7 @@ func file_controller_storage_servers_store_v1_worker_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: file_controller_storage_servers_store_v1_worker_proto_rawDesc, NumEnums: 0, - NumMessages: 3, + NumMessages: 2, NumExtensions: 0, NumServices: 0, }, diff --git a/internal/servers/worker.go b/internal/servers/worker.go new file mode 100644 index 000000000..2b1a68af4 --- /dev/null +++ b/internal/servers/worker.go @@ -0,0 +1,55 @@ +package servers + +import ( + "github.com/hashicorp/boundary/internal/servers/store" + "google.golang.org/protobuf/proto" +) + +// A Worker is a server that provides an address which can be used to proxy +// session connections. It can be tagged with custom tags and is used when +// authorizing and establishing a session. It is owned by a scope. +type Worker struct { + *store.Worker + Tags []*Tag `gorm:"-"` +} + +// A Tag is a custom key/value pair which can be attached to a Worker. +// Multiple Tags may contain the same key and different values in which +// case both key/value pairs are valid. +type Tag struct { + Key string + Value string +} + +// NewWorker returns a new Worker. Valid options are WithName, WithDescription +// WithAddress, and WithWorkerTags. All other options are ignored. +func NewWorker(scopeId string, opt ...Option) *Worker { + opts := getOpts(opt...) + return &Worker{ + Worker: &store.Worker{ + ScopeId: scopeId, + PublicId: opts.withPublicId, + Name: opts.withName, + Description: opts.withDescription, + Address: opts.withAddress, + }, + Tags: opts.withWorkerTags, + } +} + +func (w *Worker) clone() *Worker { + tags := make([]*Tag, 0, len(w.Tags)) + for _, t := range w.Tags { + tags = append(tags, &Tag{Key: t.Key, Value: t.Value}) + } + cw := proto.Clone(w.Worker) + return &Worker{ + Worker: cw.(*store.Worker), + Tags: tags, + } +} + +// TableName overrides the table name used by Worker to `server_worker` +func (Worker) TableName() string { + return "server_worker" +} diff --git a/internal/servers/worker_tags_test.go b/internal/servers/worker_tags_test.go new file mode 100644 index 000000000..bd7e16019 --- /dev/null +++ b/internal/servers/worker_tags_test.go @@ -0,0 +1,78 @@ +package servers + +import ( + "context" + "testing" + + "github.com/hashicorp/boundary/internal/db" + "github.com/hashicorp/boundary/internal/servers/store" + "github.com/hashicorp/boundary/internal/types/scope" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestWorkerTags_Create(t *testing.T) { + t.Parallel() + conn, _ := db.TestSetup(t, "postgres") + rw := db.New(conn) + + workerPublicid, err := newWorkerId(context.Background()) + require.NoError(t, err) + worker := NewWorker(scope.Global.String(), WithPublicId(workerPublicid)) + + tests := []struct { + name string + want *store.WorkerTag + wantCreateErr bool + }{ + { + name: "success", + want: &store.WorkerTag{ + WorkerId: worker.GetPublicId(), + Key: "key", + Value: "value", + }, + }, + { + name: "bad worker id", + want: &store.WorkerTag{ + WorkerId: "w_badworkeridthatdoesntexist", + Key: "key", + Value: "value", + }, + wantCreateErr: true, + }, + { + name: "missing worker id", + want: &store.WorkerTag{ + Key: "key", + Value: "value", + }, + wantCreateErr: true, + }, + { + name: "missing key", + want: &store.WorkerTag{ + WorkerId: worker.GetPublicId(), + Value: "value", + }, + }, + { + name: "missing value", + want: &store.WorkerTag{ + WorkerId: worker.GetPublicId(), + Key: "key", + }, + }, + } + + for _, tt := range tests { + tt := tt + t.Run(tt.name, func(t *testing.T) { + err := rw.Create(context.Background(), tt.want) + if tt.wantCreateErr { + assert.Error(t, err) + } + }) + } +} diff --git a/internal/servers/worker_test.go b/internal/servers/worker_test.go new file mode 100644 index 000000000..4b4bf2acd --- /dev/null +++ b/internal/servers/worker_test.go @@ -0,0 +1,185 @@ +package servers + +import ( + "context" + "testing" + + "github.com/google/go-cmp/cmp" + "github.com/hashicorp/boundary/internal/db" + "github.com/hashicorp/boundary/internal/iam" + "github.com/hashicorp/boundary/internal/servers/store" + "github.com/hashicorp/boundary/internal/types/scope" + "github.com/stretchr/testify/assert" + "google.golang.org/protobuf/testing/protocmp" +) + +func TestWorker_New(t *testing.T) { + t.Parallel() + conn, _ := db.TestSetup(t, "postgres") + wrapper := db.TestWrapper(t) + rw := db.New(conn) + + org, prj := iam.TestScopes(t, iam.TestRepo(t, conn, wrapper)) + + type args struct { + scopeId string + opts []Option + } + + tests := []struct { + name string + args args + want *Worker + wantCreateErr bool + }{ + { + name: "missing-scope-id", + args: args{}, + want: &Worker{ + Worker: &store.Worker{}, + }, + wantCreateErr: true, + }, + { + name: "global-scope-id", + args: args{ + scopeId: scope.Global.String(), + }, + want: &Worker{ + Worker: &store.Worker{ + ScopeId: scope.Global.String(), + }, + }, + }, + { + name: "org-scope-id", + args: args{ + scopeId: org.GetPublicId(), + }, + want: &Worker{ + Worker: &store.Worker{ + ScopeId: org.GetPublicId(), + }, + }, + wantCreateErr: true, + }, + { + name: "project-scope-id", + args: args{ + scopeId: prj.GetPublicId(), + }, + want: &Worker{ + Worker: &store.Worker{ + ScopeId: prj.GetPublicId(), + }, + }, + wantCreateErr: true, + }, + { + name: "with public id", + args: args{ + scopeId: scope.Global.String(), + opts: []Option{ + WithPublicId("w_test_public_id"), + }, + }, + want: &Worker{ + Worker: &store.Worker{ + PublicId: "w_test_public_id", + ScopeId: scope.Global.String(), + }, + }, + }, + { + name: "with name", + args: args{ + scopeId: scope.Global.String(), + opts: []Option{ + WithName("foo"), + }, + }, + want: &Worker{ + Worker: &store.Worker{ + ScopeId: scope.Global.String(), + Name: "foo", + }, + }, + }, + { + name: "with description", + args: args{ + scopeId: scope.Global.String(), + opts: []Option{ + WithDescription("foo"), + }, + }, + want: &Worker{ + Worker: &store.Worker{ + ScopeId: scope.Global.String(), + Description: "foo", + }, + }, + }, + { + name: "with address", + args: args{ + scopeId: scope.Global.String(), + opts: []Option{ + WithAddress("foo"), + }, + }, + want: &Worker{ + Worker: &store.Worker{ + ScopeId: scope.Global.String(), + Address: "foo", + }, + }, + }, + { + name: "with Tags", + args: args{ + scopeId: scope.Global.String(), + opts: []Option{ + WithWorkerTags(&Tag{ + Key: "key", + Value: "val", + }), + }, + }, + want: &Worker{ + Worker: &store.Worker{ + ScopeId: scope.Global.String(), + }, + Tags: []*Tag{ + { + Key: "key", + Value: "val", + }, + }, + }, + }, + } + + for _, tt := range tests { + tt := tt + t.Run(tt.name, func(t *testing.T) { + assert := assert.New(t) + got := NewWorker(tt.args.scopeId, tt.args.opts...) + assert.Empty(cmp.Diff(tt.want, got.clone(), protocmp.Transform())) + + id, err := newWorkerId(context.Background()) + assert.NoError(err) + + tt.want.PublicId = id + got.PublicId = id + + err2 := rw.Create(context.Background(), got) + if tt.wantCreateErr { + assert.Error(err2) + } else { + assert.NoError(err2) + assert.Equal(uint32(1), got.Version) + } + }) + } +} diff --git a/internal/session/service_worker_status_report.go b/internal/session/service_worker_status_report.go index 0cc62a426..0ed4820a3 100644 --- a/internal/session/service_worker_status_report.go +++ b/internal/session/service_worker_status_report.go @@ -42,7 +42,7 @@ func WorkerStatusReport(ctx context.Context, repo *Repository, connRepo *Connect return notActive, errors.New(ctx, errors.Internal, op, fmt.Sprintf("Error closing orphaned connections for worker %s: %v", workerId, err)) } if len(closed) > 0 { - event.WriteSysEvent(ctx, op, "marked unclaimed connections as closed", "server_id", workerId, "count", len(closed)) + event.WriteSysEvent(ctx, op, "marked unclaimed connections as closed", "controller_id", workerId, "count", len(closed)) } return notActive, err } diff --git a/internal/session/service_worker_status_report_test.go b/internal/session/service_worker_status_report_test.go index 52ed0bb64..ff5b2b6d7 100644 --- a/internal/session/service_worker_status_report_test.go +++ b/internal/session/service_worker_status_report_test.go @@ -5,6 +5,7 @@ import ( "testing" "github.com/hashicorp/boundary/internal/servers/store" + "github.com/hashicorp/boundary/internal/types/scope" "github.com/hashicorp/boundary/internal/authtoken" "github.com/hashicorp/boundary/internal/db" @@ -32,10 +33,9 @@ func TestWorkerStatusReport(t *testing.T) { PrivateId: "test_controller1", Address: "127.0.0.1", }) - serverRepo.UpsertWorker(ctx, &store.Worker{ - PublicId: "test_worker1", - Address: "127.0.0.1", - }) + serverRepo.UpsertWorker(ctx, servers.NewWorker(scope.Global.String(), + servers.WithPublicId("test_worker1"), + servers.WithAddress("127.0.0.1"))) repo, err := session.NewRepository(rw, rw, kms) require.NoError(t, err) @@ -56,7 +56,7 @@ func TestWorkerStatusReport(t *testing.T) { ) type testCase struct { - worker *store.Worker + worker *servers.Worker req []session.StateReport want []session.StateReport orphanedConnections []string diff --git a/internal/session/session_cleanup_job_test.go b/internal/session/session_cleanup_job_test.go index 5b63c37ce..fb8d819b3 100644 --- a/internal/session/session_cleanup_job_test.go +++ b/internal/session/session_cleanup_job_test.go @@ -12,7 +12,6 @@ import ( "github.com/hashicorp/boundary/internal/kms" "github.com/hashicorp/boundary/internal/scheduler" "github.com/hashicorp/boundary/internal/servers" - "github.com/hashicorp/boundary/internal/servers/store" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -289,7 +288,7 @@ func TestCloseConnectionsForDeadWorkers(t *testing.T) { // updateServer is a helper for updating the update time for our // servers. The controller is read back so that we can reference // the most up-to-date fields. - updateServer := func(t *testing.T, w *store.Worker) *store.Worker { + updateServer := func(t *testing.T, w *servers.Worker) *servers.Worker { t.Helper() _, rowsUpdated, err := serversRepo.UpsertWorker(ctx, w) require.NoError(err) diff --git a/internal/session/testing.go b/internal/session/testing.go index 21e04af94..dfc783025 100644 --- a/internal/session/testing.go +++ b/internal/session/testing.go @@ -6,7 +6,7 @@ import ( "testing" "time" - "github.com/hashicorp/boundary/internal/servers/store" + "github.com/hashicorp/boundary/internal/types/scope" "github.com/hashicorp/boundary/internal/auth/password" "github.com/hashicorp/boundary/internal/authtoken" @@ -181,7 +181,7 @@ func TestTofu(t testing.TB) []byte { } // TestWorker inserts a worker into the db to satisfy foreign key constraints. -func TestWorker(t *testing.T, conn *db.DB, wrapper wrapping.Wrapper, opt ...Option) *store.Worker { +func TestWorker(t *testing.T, conn *db.DB, wrapper wrapping.Wrapper, opt ...Option) *servers.Worker { t.Helper() rw := db.New(conn) kms := kms.TestKms(t, conn, wrapper) @@ -193,11 +193,10 @@ func TestWorker(t *testing.T, conn *db.DB, wrapper wrapping.Wrapper, opt ...Opti id = "test_session_worker-" + id name := "test-worker-" + id - worker := &store.Worker{ - PublicId: id, - Name: name, - Address: "127.0.0.1", - } + worker := servers.NewWorker(scope.Global.String(), + servers.WithPublicId(id), + servers.WithName(name), + servers.WithAddress("127.0.0.1")) _, _, err = serversRepo.UpsertWorker(context.Background(), worker) require.NoError(t, err) return worker diff --git a/testing/dbtest/session_list_benchmarks_dump_generation_test.go b/testing/dbtest/session_list_benchmarks_dump_generation_test.go index 939f794f8..5adf48a79 100644 --- a/testing/dbtest/session_list_benchmarks_dump_generation_test.go +++ b/testing/dbtest/session_list_benchmarks_dump_generation_test.go @@ -19,10 +19,10 @@ import ( "github.com/hashicorp/boundary/internal/iam" "github.com/hashicorp/boundary/internal/kms" "github.com/hashicorp/boundary/internal/servers" - "github.com/hashicorp/boundary/internal/servers/store" "github.com/hashicorp/boundary/internal/session" "github.com/hashicorp/boundary/internal/target" "github.com/hashicorp/boundary/internal/target/tcp" + "github.com/hashicorp/boundary/internal/types/scope" "github.com/hashicorp/boundary/testing/dbtest" "github.com/stretchr/testify/require" "golang.org/x/sync/errgroup" @@ -132,10 +132,9 @@ func TestGenerateSessionBenchmarkTemplateDumps(t *testing.T) { require.NoError(err) serversRepo, err := servers.NewRepository(rw, rw, kms) require.NoError(err) - worker := &store.Worker{ - PublicId: "test_worker_1", - Address: "127.0.0.1", - } + worker := servers.NewWorker(scope.Global.String(), + servers.WithPublicId("test_worker_1"), + servers.WithAddress("127.0.0.1")) _, _, err = serversRepo.UpsertWorker(ctx, worker) require.NoError(err) @@ -196,7 +195,7 @@ func TestGenerateSessionBenchmarkTemplateDumps(t *testing.T) { ScopeId: users[userIndex].scopeId, Endpoint: "tcp://127.0.0.1:22", }) - cycleSessionStates(t, ctx, sess, sessRepo, connRepo, conn, worker, scenario.connsPerSession) + cycleSessionStates(t, ctx, sess, sessRepo, connRepo, conn, scenario.connsPerSession) return nil }) } @@ -220,7 +219,7 @@ func TestGenerateSessionBenchmarkTemplateDumps(t *testing.T) { } } -func cycleSessionStates(t testing.TB, ctx context.Context, sess *session.Session, sessRepo *session.Repository, connRepo *session.ConnectionRepository, conn *db.DB, worker *store.Worker, numConns int) { +func cycleSessionStates(t testing.TB, ctx context.Context, sess *session.Session, sessRepo *session.Repository, connRepo *session.ConnectionRepository, conn *db.DB, numConns int) { sess, _, err := sessRepo.ActivateSession(ctx, sess.PublicId, sess.Version, []byte(`tofu`)) require.NoError(t, err) var closeWiths []session.CloseWith