diff --git a/Makefile b/Makefile index c4bd3d0cc..c4abd4552 100644 --- a/Makefile +++ b/Makefile @@ -146,7 +146,9 @@ protobuild: @protoc-go-inject-tag -input=./internal/kms/store/oplog_key.pb.go @protoc-go-inject-tag -input=./internal/kms/store/token_key.pb.go @protoc-go-inject-tag -input=./internal/kms/store/session_key.pb.go - @protoc-go-inject-tag -input=./internal/kms/store/oidc_key.pb.go + @protoc-go-inject-tag -input=./internal/kms/store/oidc_key.pb.go + @protoc-go-inject-tag -input=./internal/servers/store/controller.pb.go + @protoc-go-inject-tag -input=./internal/servers/store/worker.pb.go @protoc-go-inject-tag -input=./internal/target/store/target.pb.go @protoc-go-inject-tag -input=./internal/target/targettest/store/target.pb.go @protoc-go-inject-tag -input=./internal/target/tcp/store/target.pb.go diff --git a/internal/cmd/commands/server/worker_tags_reload_test.go b/internal/cmd/commands/server/worker_tags_reload_test.go index dc278af51..ceeb17d1d 100644 --- a/internal/cmd/commands/server/worker_tags_reload_test.go +++ b/internal/cmd/commands/server/worker_tags_reload_test.go @@ -140,7 +140,7 @@ func TestServer_ReloadWorkerTags(t *testing.T) { t.Helper() serversRepo, err := cmd.controller.ServersRepoFn() require.NoError(err) - tags, err := serversRepo.ListTagsForServers(cmd.Context, []string{name}) + tags, err := serversRepo.ListTagsForWorkers(cmd.Context, []string{name}) require.NoError(err) require.Len(tags, 2) require.Equal(key, tags[0].Key) diff --git a/internal/db/schema/migrations/oss/postgres/99/01_worker_controller_tables.up.sql b/internal/db/schema/migrations/oss/postgres/99/01_worker_controller_tables.up.sql new file mode 100644 index 000000000..cddd38dd5 --- /dev/null +++ b/internal/db/schema/migrations/oss/postgres/99/01_worker_controller_tables.up.sql @@ -0,0 +1,130 @@ +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, + create_time wt_timestamp, + update_time wt_timestamp +); + +create trigger immutable_columns before update on server_controller + for each row execute procedure immutable_columns('private_id','create_time'); + +create trigger default_create_time_column before insert on server_controller + for each row execute procedure default_create_time(); + +create trigger controller_insert_time_column before insert on server_controller + for each row execute procedure update_time_column(); + +create trigger controller_update_time_column before update on server_controller + for each row execute procedure update_time_column(); + +-- Worker table adds the field: name +create table server_worker ( + private_id text primary key, + description wt_description, + name wt_name unique, + address text not null, + create_time wt_timestamp, + update_time wt_timestamp +); + +create trigger immutable_columns before update on server_worker + for each row execute procedure immutable_columns('private_id','create_time'); + +create trigger default_create_time_column before insert on server_worker + for each row execute procedure default_create_time(); + +create trigger worker_insert_time_column before insert on server_worker + for each row execute procedure update_time_column(); + +create trigger worker_update_time_column before update on server_worker + for each row execute procedure update_time_column(); + +-- TODO: Migrate server entries to worker and controller tables, if necessary + +-- Create table worker tag +create table server_worker_tag ( + worker_id text + references server_worker(private_id) + on delete cascade + on update cascade, + key wt_tagpair, + value wt_tagpair, + primary key(worker_id, key, value) +); + +-- Aaand drop server_tag +drop table server_tag; + +-- Update session table to use worker_id instead of server_id +alter table session + drop constraint session_server_id_fkey; +drop trigger update_version_column + on session; +alter table session + drop column server_type cascade; + +alter table session + rename column server_id to worker_id; +alter table session + add constraint session_worker_id_fkey + foreign key (worker_id) + references server_worker(private_id) + on delete set null + on update cascade; + +create trigger + update_version_column + after update of version, termination_reason, key_id, tofu_token, worker_id on session + for each row execute procedure update_version_column(); + +-- Update session_connection table to use worker_id instead of server_id +alter table session_connection + drop constraint server_fkey; + +alter table session_connection + rename column server_id to worker_id; +alter table session_connection + add constraint worker_fkey + foreign key (worker_id) + references server_worker (private_id) + on delete set null + on update cascade; + +-- Update job run table so that server id references controller id +alter table job_run + drop constraint server_fkey; +alter table job_run + add constraint server_fkey + foreign key (server_id) + references server_controller (private_id) + on delete set null + on update cascade; + +-- Replaces the view created in 9/01 to include worker id instead of server id and server type +drop view if exists session_list; + +create view session_list as + + select + s.public_id, s.user_id, s.host_id, s.worker_id, s.target_id, + s.host_set_id, s.auth_token_id, s.scope_id, s.certificate,s.expiration_time, + s.connection_limit, s.tofu_token, s.key_id, s.termination_reason, s.version, + s.create_time, s.update_time, s.endpoint, s.worker_filter, + ss.state, ss.previous_end_time, ss.start_time, ss.end_time, sc.public_id as connection_id, + sc.client_tcp_address, sc.client_tcp_port, sc.endpoint_tcp_address, sc.endpoint_tcp_port, + sc.bytes_up, sc.bytes_down, sc.closed_reason + from session s + join session_state ss on + s.public_id = ss.session_id + left join session_connection sc on + s.public_id = sc.session_id; + +-- Drop the server and server_type_enm tables +drop table server; +drop table server_type_enm; + +commit; \ No newline at end of file diff --git a/internal/db/schema/migrations/oss/postgres_24_01_test.go b/internal/db/schema/migrations/oss/postgres_24_01_test.go deleted file mode 100644 index bb1893d4b..000000000 --- a/internal/db/schema/migrations/oss/postgres_24_01_test.go +++ /dev/null @@ -1,166 +0,0 @@ -package oss_test - -import ( - "context" - "testing" - - "github.com/hashicorp/boundary/internal/authtoken" - "github.com/hashicorp/boundary/internal/db" - "github.com/hashicorp/boundary/internal/db/common" - "github.com/hashicorp/boundary/internal/db/schema" - "github.com/hashicorp/boundary/internal/host/static" - "github.com/hashicorp/boundary/internal/iam" - "github.com/hashicorp/boundary/internal/kms" - "github.com/hashicorp/boundary/internal/session" - "github.com/hashicorp/boundary/internal/target" - "github.com/hashicorp/boundary/internal/target/tcp" - "github.com/hashicorp/boundary/testing/dbtest" - "github.com/stretchr/testify/require" -) - -func TestMigrations_SessionState(t *testing.T) { - t.Parallel() - require := require.New(t) - - const ( - priorMigration = 21002 - currentMigration = 24001 - ) - dialect := dbtest.Postgres - ctx := context.Background() - - c, u, _, err := dbtest.StartUsingTemplate(dialect, dbtest.WithTemplate(dbtest.Template1)) - require.NoError(err) - t.Cleanup(func() { - require.NoError(c()) - }) - d, err := common.SqlOpen(dialect, u) - require.NoError(err) - - // migration to the prior migration (before the one we want to test) - m, err := schema.NewManager(ctx, schema.Dialect(dialect), d, schema.WithEditions( - schema.TestCreatePartialEditions(schema.Dialect(dialect), schema.PartialEditions{"oss": priorMigration}), - )) - require.NoError(err) - - require.NoError(m.ApplyMigrations(ctx)) - state, err := m.CurrentState(ctx) - require.NoError(err) - want := &schema.State{ - Initialized: true, - Editions: []schema.EditionState{ - { - Name: "oss", - BinarySchemaVersion: priorMigration, - DatabaseSchemaVersion: priorMigration, - DatabaseSchemaState: schema.Equal, - }, - }, - } - require.Equal(want, state) - - // Seed the database with test data - dbType, err := db.StringToDbType(dialect) - require.NoError(err) - - conn, err := db.Open(dbType, u) - require.NoError(err) - - rw := db.New(conn) - wrapper := db.TestWrapper(t) - - org, prj := iam.TestScopes(t, iam.TestRepo(t, conn, wrapper)) - require.NotNil(prj) - - hc := static.TestCatalogs(t, conn, prj.GetPublicId(), 1)[0] - hs := static.TestSets(t, conn, hc.GetPublicId(), 1)[0] - h := static.TestHosts(t, conn, hc.GetPublicId(), 1)[0] - static.TestSetMembers(t, conn, hs.GetPublicId(), []*static.Host{h}) - - tar := tcp.TestTarget(ctx, t, conn, prj.GetPublicId(), "test", target.WithHostSources([]string{hs.GetPublicId()})) - - kmsCache := kms.TestKms(t, conn, wrapper) - - { - at := authtoken.TestAuthToken(t, conn, kmsCache, org.GetPublicId()) - uId := at.GetIamUserId() - - sess := session.TestSession(t, conn, wrapper, session.ComposedOf{ - UserId: uId, - HostId: h.GetPublicId(), - TargetId: tar.GetPublicId(), - HostSetId: hs.GetPublicId(), - AuthTokenId: at.GetPublicId(), - ScopeId: prj.GetPublicId(), - Endpoint: "tcp://127.0.0.1:22", - ConnectionLimit: -1, - }) - - session.TestConnection(t, conn, sess.PublicId, "127.0.0.1", 22, - "127.0.0.2", 23, "127.0.0.1") - } - - // now we're ready for the migration we want to test. - m, err = schema.NewManager(ctx, schema.Dialect(dialect), d, schema.WithEditions( - schema.TestCreatePartialEditions(schema.Dialect(dialect), schema.PartialEditions{"oss": currentMigration}), - )) - require.NoError(err) - - require.NoError(m.ApplyMigrations(ctx)) - state, err = m.CurrentState(ctx) - require.NoError(err) - want = &schema.State{ - Initialized: true, - Editions: []schema.EditionState{ - { - Name: "oss", - BinarySchemaVersion: currentMigration, - DatabaseSchemaVersion: currentMigration, - DatabaseSchemaState: schema.Equal, - }, - }, - } - require.Equal(want, state) - - sessionRepo, err := session.NewRepository(rw, rw, kmsCache) - connectionRepo, err := session.NewConnectionRepository(ctx, rw, rw, kmsCache) - require.NoError(err) - - // Ensure session is cancelled - repoSessions, err := sessionRepo.ListSessions(ctx) - require.NoError(err) - var sessionTermReason []string - var repoSessionId string - for i := 0; i < len(repoSessions); i++ { - s := repoSessions[i] - repoSessionId = s.PublicId - sessionTermReason = append(sessionTermReason, s.TerminationReason) - } - require.Equal([]string{"canceled"}, sessionTermReason) - - // Ensure connection is also cancelled - connections, err := connectionRepo.ListConnectionsBySessionId(ctx, repoSessionId) - require.NoError(err) - - var connTermReason []string - for i := 0; i < len(connections); i++ { - c := connections[i] - connTermReason = append(connTermReason, c.ClosedReason) - } - require.Equal([]string{"canceled"}, connTermReason) - - // Validate new table contents - rows, err := d.QueryContext(ctx, "select * from session_valid_state") - require.NoError(err) - - var validStates []string - for rows.Next() { - var a, b string - require.NoError(rows.Scan(&a, &b)) - validStates = append(validStates, []string{a, b}...) - } - require.Equal([]string{ - "pending", "pending", "pending", "active", "pending", "terminated", - "pending", "canceling", "active", "canceling", "active", "terminated", "canceling", "terminated", - }, validStates) -} diff --git a/internal/gen/controller/servers/services/server_coordination_service.pb.go b/internal/gen/controller/servers/services/server_coordination_service.pb.go index bf6698ec1..c893c7c2b 100644 --- a/internal/gen/controller/servers/services/server_coordination_service.pb.go +++ b/internal/gen/controller/servers/services/server_coordination_service.pb.go @@ -468,7 +468,7 @@ type StatusRequest struct { unknownFields protoimpl.UnknownFields // The worker info. We could use information from the TLS connection but this - // is easier and going the other route doesn't provijde much benefit -- if you + // is easier and going the other route doesn't provide much benefit -- if you // get access to the key and spoof the connection, you're already compromised. Worker *servers.Server `protobuf:"bytes,10,opt,name=worker,proto3" json:"worker,omitempty"` // Jobs which this worker wants to report the status. @@ -593,8 +593,7 @@ type StatusResponse struct { sizeCache protoimpl.SizeCache unknownFields protoimpl.UnknownFields - // Active controllers. This can be used (eventually) for connection - // management. + // Active controllers. This can be used for connection management. Controllers []*servers.Server `protobuf:"bytes,10,rep,name=controllers,proto3" json:"controllers,omitempty"` // List of jobs and the expected state changes. For example, this will // include jobs witch change type of canceled for jobs which are active on a diff --git a/internal/proto/controller/servers/services/v1/server_coordination_service.proto b/internal/proto/controller/servers/services/v1/server_coordination_service.proto index 098b2d7fc..6f8a0b009 100644 --- a/internal/proto/controller/servers/services/v1/server_coordination_service.proto +++ b/internal/proto/controller/servers/services/v1/server_coordination_service.proto @@ -56,9 +56,11 @@ message JobStatus { Job job = 1; } + + message StatusRequest { // The worker info. We could use information from the TLS connection but this - // is easier and going the other route doesn't provijde much benefit -- if you + // is easier and going the other route doesn't provide much benefit -- if you // get access to the key and spoof the connection, you're already compromised. servers.v1.Server worker = 10; @@ -85,8 +87,7 @@ message JobChangeRequest { } message StatusResponse { - // Active controllers. This can be used (eventually) for connection - // management. + // Active controllers. This can be used for connection management. repeated servers.v1.Server controllers = 10; // List of jobs and the expected state changes. For example, this will diff --git a/internal/proto/controller/storage/servers/store/v1/controller.proto b/internal/proto/controller/storage/servers/store/v1/controller.proto new file mode 100644 index 000000000..5d2be86f6 --- /dev/null +++ b/internal/proto/controller/storage/servers/store/v1/controller.proto @@ -0,0 +1,25 @@ +syntax = "proto3"; + +package controller.storage.servers.store.v1; +option go_package = "github.com/hashicorp/boundary/internal/servers/store;store"; + +import "controller/storage/timestamp/v1/timestamp.proto"; + +// Controller contains all fields related to a Controller resource +message Controller { + // Private ID of the resource + string private_id = 10; + + // Address for the controller + string address = 20; + + // Description of the resource (optional) + // @inject_tag: `gorm:"default:null"` + string description = 30; + + // First seen time from the RDBMS + storage.timestamp.v1.Timestamp create_time = 40; + + // Last time there was an update + storage.timestamp.v1.Timestamp update_time = 50; +} \ No newline at end of file diff --git a/internal/proto/controller/storage/servers/store/v1/worker.proto b/internal/proto/controller/storage/servers/store/v1/worker.proto new file mode 100644 index 000000000..6203a5f6a --- /dev/null +++ b/internal/proto/controller/storage/servers/store/v1/worker.proto @@ -0,0 +1,39 @@ +syntax = "proto3"; + +package controller.storage.servers.store.v1; +option go_package = "github.com/hashicorp/boundary/internal/servers/store;store"; + +import "controller/storage/timestamp/v1/timestamp.proto"; + +// Worker contains all fields related to a Worker resource +message Worker { + // Private ID of the resource + string private_id = 10; + + // Name of the resource (optional) + // Description of the resource (optional) + // @inject_tag: `gorm:"default:null"` + string name = 20; + + // Description of the resource (optional) + // @inject_tag: `gorm:"default:null"` + string description = 30; + + // Address for the worker + string address = 40; + + // First seen time from the RDBMS + storage.timestamp.v1.Timestamp create_time = 50; + + // Last time there was an update + storage.timestamp.v1.Timestamp update_time = 60; + + // @gotags: `gorm:"-"` + map tags = 70; +} + +// TagValues is used because map fields cannot be repeated but can be a +// message +message TagValues { + repeated string values = 10; +} diff --git a/internal/scheduler/job/testing.go b/internal/scheduler/job/testing.go index 8a2fa6852..c1c43450b 100644 --- a/internal/scheduler/job/testing.go +++ b/internal/scheduler/job/testing.go @@ -6,6 +6,8 @@ import ( "testing" "time" + "github.com/hashicorp/boundary/internal/servers/store" + "github.com/hashicorp/boundary/internal/db" "github.com/hashicorp/boundary/internal/kms" "github.com/hashicorp/boundary/internal/servers" @@ -88,7 +90,7 @@ func testRunWithUpdateTime(conn *db.DB, pluginId, name, cId string, updateTime t return run, nil } -func testController(t testing.TB, conn *db.DB, wrapper wrapping.Wrapper) *servers.Server { +func testController(t *testing.T, conn *db.DB, wrapper wrapping.Wrapper) *store.Controller { t.Helper() rw := db.New(conn) kms := kms.TestKms(t, conn, wrapper) @@ -97,13 +99,11 @@ func testController(t testing.TB, conn *db.DB, wrapper wrapping.Wrapper) *server id, err := uuid.GenerateUUID() require.NoError(t, err) - controller := &servers.Server{ - PrivateId: "test-job-server-" + id, - Type: servers.ServerTypeController.String(), - Description: "Test Job Controller", - Address: "127.0.0.1", + controller := &store.Controller{ + PrivateId: "test-job-server-" + id, + Address: "127.0.0.1", } - _, _, err = serversRepo.UpsertServer(context.Background(), controller) + _, err = serversRepo.UpsertController(context.Background(), controller) require.NoError(t, err) return controller } diff --git a/internal/scheduler/testing.go b/internal/scheduler/testing.go index e11672b0b..93c6a4752 100644 --- a/internal/scheduler/testing.go +++ b/internal/scheduler/testing.go @@ -6,6 +6,8 @@ import ( "testing" "time" + "github.com/hashicorp/boundary/internal/servers/store" + "github.com/hashicorp/boundary/internal/db" "github.com/hashicorp/boundary/internal/iam" "github.com/hashicorp/boundary/internal/kms" @@ -32,13 +34,11 @@ func TestScheduler(t testing.TB, conn *db.DB, wrapper wrapping.Wrapper, opt ...O id, err := uuid.GenerateUUID() require.NoError(t, err) - controller := &servers.Server{ - PrivateId: "test-job-server-" + id, - Type: servers.ServerTypeController.String(), - Description: "Test Job Controller", - Address: "127.0.0.1", + controller := &store.Controller{ + PrivateId: "test-job-server-" + id, + Address: "127.0.0.1", } - _, _, err = serversRepo.UpsertServer(context.Background(), controller) + _, err = serversRepo.UpsertController(context.Background(), controller) require.NoError(t, err) jobRepoFn := func() (*job.Repository, error) { diff --git a/internal/servers/controller/controller.go b/internal/servers/controller/controller.go index aeeb2b60f..a089694fc 100644 --- a/internal/servers/controller/controller.go +++ b/internal/servers/controller/controller.go @@ -334,9 +334,9 @@ func (c *Controller) Start() error { return fmt.Errorf("error starting controller listeners: %w", err) } - // Upsert server before starting tickers and scheduler to ensure the server exists - if err := c.upsertServer(c.baseContext); err != nil { - return fmt.Errorf("error upserting server: %w", err) + // Upsert controller before starting tickers and scheduler to ensure the controller exists + if err := c.upsertController(c.baseContext); err != nil { + return fmt.Errorf("error upserting controller: %w", err) } if err := c.scheduler.Start(c.baseContext, c.schedulerWg); err != nil { return fmt.Errorf("error starting scheduler: %w", err) diff --git a/internal/servers/controller/handlers/targets/target_service.go b/internal/servers/controller/handlers/targets/target_service.go index d90528da0..19798d438 100644 --- a/internal/servers/controller/handlers/targets/target_service.go +++ b/internal/servers/controller/handlers/targets/target_service.go @@ -23,7 +23,6 @@ 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/servers/controller/auth" "github.com/hashicorp/boundary/internal/servers/controller/common" "github.com/hashicorp/boundary/internal/servers/controller/common/scopeids" @@ -898,7 +897,7 @@ func (s Service) AuthorizeSession(ctx context.Context, req *pbs.AuthorizeSession var workers []*pb.WorkerInfo var workerIds []string hasWorkerFilter := len(t.GetWorkerFilter()) > 0 - servers, err := serversRepo.ListServers(ctx, servers.ServerTypeWorker) + servers, err := serversRepo.ListWorkers(ctx) if err != nil { return nil, err } @@ -912,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.ListTagsForServers(ctx, workerIds) + tags, err := serversRepo.ListTagsForWorkers(ctx, workerIds) if err != nil { return nil, err } @@ -921,10 +920,10 @@ func (s Service) AuthorizeSession(ctx context.Context, req *pbs.AuthorizeSession // worker's ID to its filter map. tagMap := make(map[string]map[string][]string) for _, tag := range tags { - currWorkerMap := tagMap[tag.ServerId] + currWorkerMap := tagMap[tag.WorkerId] if currWorkerMap == nil { currWorkerMap = make(map[string][]string) - tagMap[tag.ServerId] = currWorkerMap + tagMap[tag.WorkerId] = currWorkerMap } currWorkerMap[tag.Key] = append(currWorkerMap[tag.Key], tag.Value) // We don't need to reinsert after the fact because maps are diff --git a/internal/servers/controller/handlers/workers/worker_service.go b/internal/servers/controller/handlers/workers/worker_service.go index 29df3393c..45fbbb473 100644 --- a/internal/servers/controller/handlers/workers/worker_service.go +++ b/internal/servers/controller/handlers/workers/worker_service.go @@ -7,6 +7,8 @@ import ( "sync" "time" + "github.com/hashicorp/boundary/internal/servers/store" + pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services" "github.com/hashicorp/boundary/internal/kms" "github.com/hashicorp/boundary/internal/observability/event" @@ -14,7 +16,6 @@ import ( "github.com/hashicorp/boundary/internal/servers/controller/common" "github.com/hashicorp/boundary/internal/servers/controller/handlers" "github.com/hashicorp/boundary/internal/session" - "github.com/hashicorp/boundary/internal/types/resource" "github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/targets" "github.com/hashicorp/go-bexpr" "google.golang.org/grpc/codes" @@ -70,14 +71,41 @@ func (ws *workerServiceServer) Status(ctx context.Context, req *pbs.StatusReques event.WriteError(ctx, op, err, event.WithInfoMsg("error getting sessions repo")) return &pbs.StatusResponse{}, status.Errorf(codes.Internal, "Error acquiring repo to query session status: %v", err) } - req.Worker.Type = resource.Worker.String() - controllers, _, err := serverRepo.UpsertServer(ctx, req.Worker, servers.WithUpdateTags(req.GetUpdateTags())) + + reqServer := req.GetWorker() + // Convert API tags to storage tags + workerTags := make(map[string]*store.TagValues) + for k, v := range reqServer.Tags { + thisTag := &store.TagValues{ + Values: v.GetValues(), + } + workerTags[k] = thisTag + } + worker := &store.Worker{ + PrivateId: reqServer.PrivateId, + Address: reqServer.Address, + CreateTime: reqServer.CreateTime, + UpdateTime: reqServer.UpdateTime, + Tags: 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")) return &pbs.StatusResponse{}, status.Errorf(codes.Internal, "Error storing worker status: %v", err) } + + responseControllers := []*servers.Server{} + for _, c := range controllers { + thisController := &servers.Server{ + PrivateId: c.PrivateId, + Address: c.Address, + CreateTime: c.CreateTime, + UpdateTime: c.UpdateTime, + } + responseControllers = append(responseControllers, thisController) + } ret := &pbs.StatusResponse{ - Controllers: controllers, + Controllers: responseControllers, } stateReport := make([]session.StateReport, 0, len(req.GetJobs())) @@ -171,7 +199,7 @@ 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.ListTagsForServers(ctx, []string{req.ServerId}) + tags, 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) @@ -295,7 +323,6 @@ func (ws *workerServiceServer) ActivateSession(ctx context.Context, req *pbs.Act req.GetSessionId(), req.GetVersion(), req.GetWorkerId(), - resource.Worker.String(), []byte(req.GetTofuToken())) if err != nil { return nil, status.Errorf(codes.Internal, "error looking up session: %v", err) diff --git a/internal/servers/controller/handlers/workers/worker_service_status_test.go b/internal/servers/controller/handlers/workers/worker_service_status_test.go index c1efb3ba7..b66dc3351 100644 --- a/internal/servers/controller/handlers/workers/worker_service_status_test.go +++ b/internal/servers/controller/handlers/workers/worker_service_status_test.go @@ -5,6 +5,8 @@ import ( "sync" "testing" + "github.com/hashicorp/boundary/internal/servers/store" + "github.com/google/go-cmp/cmp" "github.com/google/go-cmp/cmp/cmpopts" "github.com/hashicorp/boundary/internal/authtoken" @@ -31,14 +33,12 @@ func TestStatus(t *testing.T) { org, prj := iam.TestScopes(t, iam.TestRepo(t, conn, wrapper)) serverRepo, _ := servers.NewRepository(rw, rw, kms) - serverRepo.UpsertServer(ctx, &servers.Server{ + serverRepo.UpsertController(ctx, &store.Controller{ PrivateId: "test_controller1", - Type: "controller", Address: "127.0.0.1", }) - serverRepo.UpsertServer(ctx, &servers.Server{ + serverRepo.UpsertWorker(ctx, &store.Worker{ PrivateId: "test_worker1", - Type: "worker", Address: "127.0.0.1", }) @@ -83,7 +83,7 @@ func TestStatus(t *testing.T) { ConnectionLimit: 10, }) tofu := session.TestTofu(t) - sess, _, err = repo.ActivateSession(ctx, sess.PublicId, sess.Version, worker1.PrivateId, worker1.Type, tofu) + sess, _, err = repo.ActivateSession(ctx, sess.PublicId, sess.Version, worker1.PrivateId, tofu) require.NoError(t, err) require.NoError(t, err) @@ -104,13 +104,17 @@ func TestStatus(t *testing.T) { name: "No Sessions", wantErr: false, req: &pbs.StatusRequest{ - Worker: worker1, + Worker: &servers.Server{ + PrivateId: worker1.PrivateId, + Address: worker1.Address, + CreateTime: worker1.CreateTime, + UpdateTime: worker1.UpdateTime, + }, }, want: &pbs.StatusResponse{ Controllers: []*servers.Server{ { PrivateId: "test_controller1", - Type: "controller", Address: "127.0.0.1", }, }, @@ -120,7 +124,12 @@ func TestStatus(t *testing.T) { name: "Still Active", wantErr: false, req: &pbs.StatusRequest{ - Worker: worker1, + Worker: &servers.Server{ + PrivateId: worker1.PrivateId, + Address: worker1.Address, + CreateTime: worker1.CreateTime, + UpdateTime: worker1.UpdateTime, + }, Jobs: []*pbs.JobStatus{ { Job: &pbs.Job{ @@ -145,7 +154,6 @@ func TestStatus(t *testing.T) { Controllers: []*servers.Server{ { PrivateId: "test_controller1", - Type: "controller", Address: "127.0.0.1", }, }, @@ -193,14 +201,12 @@ func TestStatusSessionClosed(t *testing.T) { org, prj := iam.TestScopes(t, iam.TestRepo(t, conn, wrapper)) serverRepo, _ := servers.NewRepository(rw, rw, kms) - serverRepo.UpsertServer(ctx, &servers.Server{ + serverRepo.UpsertController(ctx, &store.Controller{ PrivateId: "test_controller1", - Type: "controller", Address: "127.0.0.1", }) - serverRepo.UpsertServer(ctx, &servers.Server{ + serverRepo.UpsertWorker(ctx, &store.Worker{ PrivateId: "test_worker1", - Type: "worker", Address: "127.0.0.1", }) @@ -245,7 +251,7 @@ func TestStatusSessionClosed(t *testing.T) { ConnectionLimit: 10, }) tofu := session.TestTofu(t) - sess, _, err = repo.ActivateSession(ctx, sess.PublicId, sess.Version, worker1.PrivateId, worker1.Type, tofu) + sess, _, err = repo.ActivateSession(ctx, sess.PublicId, sess.Version, worker1.PrivateId, tofu) require.NoError(t, err) sess2 := session.TestSession(t, conn, wrapper, session.ComposedOf{ UserId: uId, @@ -258,7 +264,7 @@ func TestStatusSessionClosed(t *testing.T) { ConnectionLimit: 10, }) tofu2 := session.TestTofu(t) - sess2, _, err = repo.ActivateSession(ctx, sess2.PublicId, sess2.Version, worker1.PrivateId, worker1.Type, tofu2) + sess2, _, err = repo.ActivateSession(ctx, sess2.PublicId, sess2.Version, worker1.PrivateId, tofu2) require.NoError(t, err) s := workers.NewWorkerServiceServer(serversRepoFn, sessionRepoFn, connRepoFn, new(sync.Map), kms) @@ -283,7 +289,12 @@ func TestStatusSessionClosed(t *testing.T) { require.NoError(t, err) }, req: &pbs.StatusRequest{ - Worker: worker1, + Worker: &servers.Server{ + PrivateId: worker1.PrivateId, + Address: worker1.Address, + CreateTime: worker1.CreateTime, + UpdateTime: worker1.UpdateTime, + }, Jobs: []*pbs.JobStatus{ { Job: &pbs.Job{ @@ -308,7 +319,6 @@ func TestStatusSessionClosed(t *testing.T) { Controllers: []*servers.Server{ { PrivateId: "test_controller1", - Type: "controller", Address: "127.0.0.1", }, }, @@ -373,14 +383,12 @@ func TestStatusDeadConnection(t *testing.T) { org, prj := iam.TestScopes(t, iam.TestRepo(t, conn, wrapper)) serverRepo, _ := servers.NewRepository(rw, rw, kms) - serverRepo.UpsertServer(ctx, &servers.Server{ + serverRepo.UpsertController(ctx, &store.Controller{ PrivateId: "test_controller1", - Type: "controller", Address: "127.0.0.1", }) - serverRepo.UpsertServer(ctx, &servers.Server{ + serverRepo.UpsertWorker(ctx, &store.Worker{ PrivateId: "test_worker1", - Type: "worker", Address: "127.0.0.1", }) @@ -425,7 +433,7 @@ func TestStatusDeadConnection(t *testing.T) { ConnectionLimit: 10, }) tofu := session.TestTofu(t) - sess, _, err = repo.ActivateSession(ctx, sess.PublicId, sess.Version, worker1.PrivateId, worker1.Type, tofu) + sess, _, err = repo.ActivateSession(ctx, sess.PublicId, sess.Version, worker1.PrivateId, tofu) require.NoError(t, err) sess2 := session.TestSession(t, conn, wrapper, session.ComposedOf{ UserId: uId, @@ -438,7 +446,7 @@ func TestStatusDeadConnection(t *testing.T) { ConnectionLimit: 10, }) tofu2 := session.TestTofu(t) - sess2, _, err = repo.ActivateSession(ctx, sess2.PublicId, sess2.Version, worker1.PrivateId, worker1.Type, tofu2) + sess2, _, err = repo.ActivateSession(ctx, sess2.PublicId, sess2.Version, worker1.PrivateId, tofu2) require.NoError(t, err) s := workers.NewWorkerServiceServer(serversRepoFn, sessionRepoFn, connRepoFn, new(sync.Map), kms) @@ -451,7 +459,12 @@ func TestStatusDeadConnection(t *testing.T) { require.NotEqual(t, deadConn.PublicId, connection.PublicId) req := &pbs.StatusRequest{ - Worker: worker1, + Worker: &servers.Server{ + PrivateId: worker1.PrivateId, + Address: worker1.Address, + CreateTime: worker1.CreateTime, + UpdateTime: worker1.UpdateTime, + }, Jobs: []*pbs.JobStatus{ { Job: &pbs.Job{ @@ -476,7 +489,6 @@ func TestStatusDeadConnection(t *testing.T) { Controllers: []*servers.Server{ { PrivateId: "test_controller1", - Type: "controller", Address: "127.0.0.1", }, }, diff --git a/internal/servers/controller/tickers.go b/internal/servers/controller/tickers.go index 749195f50..3de6f52fd 100644 --- a/internal/servers/controller/tickers.go +++ b/internal/servers/controller/tickers.go @@ -5,10 +5,10 @@ import ( "math/rand" "time" + "github.com/hashicorp/boundary/internal/servers/store" + "github.com/hashicorp/boundary/internal/errors" "github.com/hashicorp/boundary/internal/observability/event" - "github.com/hashicorp/boundary/internal/servers" - "github.com/hashicorp/boundary/internal/types/resource" ) // In the future we could make this configurable @@ -30,7 +30,7 @@ func (c *Controller) startStatusTicking(cancelCtx context.Context) { return case <-timer.C: - if err := c.upsertServer(cancelCtx); err != nil { + if err := c.upsertController(cancelCtx); err != nil { event.WriteError(cancelCtx, op, err, event.WithInfoMsg("error fetching repository for status update")) } timer.Reset(statusInterval) @@ -38,21 +38,18 @@ func (c *Controller) startStatusTicking(cancelCtx context.Context) { } } -func (c *Controller) upsertServer(ctx context.Context) error { - const op = "controller.(Controller).upsertServer" - server := &servers.Server{ - PrivateId: c.conf.RawConfig.Controller.Name, - Name: c.conf.RawConfig.Controller.Name, - Type: resource.Controller.String(), - Description: c.conf.RawConfig.Controller.Description, - Address: c.conf.RawConfig.Controller.PublicClusterAddr, +func (c *Controller) upsertController(ctx context.Context) error { + const op = "controller.(Controller).upsertController" + controller := &store.Controller{ + PrivateId: c.conf.RawConfig.Controller.Name, + Address: c.conf.RawConfig.Controller.PublicClusterAddr, } repo, err := c.ServersRepoFn() if err != nil { return errors.Wrap(ctx, err, op, errors.WithMsg("error fetching repository for status update")) } - _, _, err = repo.UpsertServer(ctx, server) + _, err = repo.UpsertController(ctx, controller) if err != nil { return errors.Wrap(ctx, err, op, errors.WithMsg("error performing status update")) } diff --git a/internal/servers/gorm.go b/internal/servers/gorm.go index 4ec277ef7..7588948f6 100644 --- a/internal/servers/gorm.go +++ b/internal/servers/gorm.go @@ -1,5 +1,6 @@ package servers -func (s *Server) TableName() string { - return "server" +// TableName overrides the table name used by WorkerTag to `worker_tag` +func (WorkerTag) TableName() string { + return "server_worker_tag" } diff --git a/internal/servers/query.go b/internal/servers/query.go index c0f865a17..81d52677e 100644 --- a/internal/servers/query.go +++ b/internal/servers/query.go @@ -2,5 +2,5 @@ package servers const ( deleteWhereCreateTimeSql = `create_time < ?` - deleteTagsSql = `server_id = ?` + deleteTagsSql = `worker_id = ?` ) diff --git a/internal/servers/repository.go b/internal/servers/repository.go index 4d71ae8b8..508968870 100644 --- a/internal/servers/repository.go +++ b/internal/servers/repository.go @@ -9,7 +9,7 @@ import ( "github.com/hashicorp/boundary/internal/db" "github.com/hashicorp/boundary/internal/errors" "github.com/hashicorp/boundary/internal/kms" - "github.com/hashicorp/boundary/internal/types/resource" + "github.com/hashicorp/boundary/internal/servers/store" ) const ( @@ -62,16 +62,14 @@ func NewRepository(r db.Reader, w db.Writer, kms *kms.Kms) (*Repository, error) }, nil } -// ListServers is a passthrough to listServersWithReader that uses the repo's -// normal reader. -func (r *Repository) ListServers(ctx context.Context, serverType ServerType, opt ...Option) ([]*Server, error) { - return r.listServersWithReader(ctx, r.reader, serverType, opt...) +// 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) { + return r.listWorkersWithReader(ctx, r.reader, opt...) } -// listServersWithReader 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) listServersWithReader(ctx context.Context, reader db.Reader, serverType ServerType, opt ...Option) ([]*Server, error) { +// 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) { opts := getOpts(opt...) liveness := opts.withLiveness if liveness == 0 { @@ -80,65 +78,89 @@ func (r *Repository) listServersWithReader(ctx context.Context, reader db.Reader var where string if liveness > 0 { - where = fmt.Sprintf("type = ? and update_time > now() - interval '%d seconds'", uint32(liveness.Seconds())) - } else { - where = "type = ?" + where = fmt.Sprintf("update_time > now() - interval '%d seconds'", uint32(liveness.Seconds())) } - var servers []*Server + var workers []*store.Worker if err := reader.SearchWhere( ctx, - &servers, + &workers, where, - []interface{}{serverType}, + []interface{}{}, db.WithLimit(-1), ); err != nil { - return nil, errors.Wrap(ctx, err, "servers.listServersWithReader") + return nil, errors.Wrap(ctx, err, "workers.listWorkersWithReader") } - return servers, nil + return workers, nil } -// ServerTag holds the information for the server_tag table for Gorm. -type ServerTag struct { - ServerId string +func (r *Repository) ListControllers(ctx context.Context, opt ...Option) ([]*store.Controller, error) { + return r.listControllersWithReader(ctx, r.reader, opt...) +} + +// listControllersWithReader 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) listControllersWithReader(ctx context.Context, reader db.Reader, opt ...Option) ([]*store.Controller, error) { + opts := getOpts(opt...) + liveness := opts.withLiveness + if liveness == 0 { + liveness = DefaultLiveness + } + + var where string + if liveness > 0 { + where = fmt.Sprintf("update_time > now() - interval '%d seconds'", uint32(liveness.Seconds())) + } + + var controllers []*store.Controller + if err := reader.SearchWhere( + ctx, + &controllers, + where, + []interface{}{}, + db.WithLimit(-1), + ); err != nil { + return nil, errors.Wrap(ctx, err, "workers.listControllersWithReader") + } + + return controllers, nil +} + +// WorkerTag holds the information for the worker_tag table for Gorm. +type WorkerTag struct { + WorkerId string Key string Value string } -// TableName overrides the table name used by ServerTag to `server_tag` -func (ServerTag) TableName() string { - return "server_tag" -} - -// ListTagsForServers pulls out tag tuples into ServerTag structs for the -// given server ID values. -func (r *Repository) ListTagsForServers(ctx context.Context, serverIds []string, opt ...Option) ([]*ServerTag, error) { - var serverTags []*ServerTag +// 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 if err := r.reader.SearchWhere( ctx, - &serverTags, - "server_id in (?)", - []interface{}{serverIds}, + &workerTags, + "worker_id in (?)", + []interface{}{workerIds}, db.WithLimit(-1), ); err != nil { - return nil, errors.Wrap(ctx, err, "servers.ListTagsForServers", errors.WithMsg(fmt.Sprintf("server IDs %v", serverIds))) + return nil, errors.Wrap(ctx, err, "servers.ListTagsForWorkers", errors.WithMsg(fmt.Sprintf("worker IDs %v", workerIds))) } - return serverTags, nil + return workerTags, nil } -// UpsertServer adds or updates a server in the DB -func (r *Repository) UpsertServer(ctx context.Context, server *Server, opt ...Option) ([]*Server, int, error) { - const op = "servers.UpsertServer" +func (r *Repository) UpsertWorker(ctx context.Context, worker *store.Worker, opt ...Option) ([]*store.Controller, int, error) { + const op = "servers.UpsertWorker" - if server == nil { - return nil, db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "server is nil") + if worker == nil { + return nil, db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "worker is nil") } opts := getOpts(opt...) var rowsUpdated int64 - var controllers []*Server + var controllers []*store.Controller _, err := r.writer.DoTx( ctx, db.StdRetryCnt, @@ -146,21 +168,18 @@ func (r *Repository) UpsertServer(ctx context.Context, server *Server, opt ...Op func(read db.Reader, w db.Writer) error { var err error onConflict := &db.OnConflict{ - Target: db.Constraint("server_pkey"), - Action: append(db.SetColumns([]string{"type", "description", "address"}), db.SetColumnValues(map[string]interface{}{"update_time": "now()"})...), + Target: db.Columns{"private_id"}, + Action: append(db.SetColumns([]string{"description", "address"}), db.SetColumnValues(map[string]interface{}{"update_time": "now()"})...), } - err = w.Create(ctx, server, db.WithOnConflict(onConflict), db.WithReturnRowsAffected(&rowsUpdated)) + err = w.Create(ctx, worker, db.WithOnConflict(onConflict), db.WithReturnRowsAffected(&rowsUpdated)) if err != nil { return errors.Wrap(ctx, err, op+":Upsert") } - // If it's a worker, fetch the current controllers to feed to them - if server.Type == resource.Worker.String() { - // Fetch current controllers to feed to the workers - controllers, err = r.listServersWithReader(ctx, read, ServerTypeController) - if err != nil { - return errors.Wrap(ctx, err, op+":ListServer") - } + // Fetch current controllers to feed to the workers + controllers, err = r.listControllersWithReader(ctx, read) + if err != nil { + return errors.Wrap(ctx, err, op+":ListController") } // If we've been told to update tags, we need to clean out old @@ -168,31 +187,31 @@ func (r *Repository) UpsertServer(ctx context.Context, server *Server, 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, &ServerTag{}, db.WithWhere(deleteTagsSql, server.PrivateId)) + _, err = w.Delete(ctx, &WorkerTag{}, db.WithWhere(deleteTagsSql, worker.PrivateId)) if err != nil { - return errors.Wrap(ctx, err, op+":DeleteTags", errors.WithMsg(server.PrivateId)) + return errors.Wrap(ctx, err, op+":DeleteTags", errors.WithMsg(worker.PrivateId)) } // 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(server.Tags) > 0 { - tags := make([]interface{}, 0, len(server.Tags)) - for k, v := range server.Tags { + if len(worker.Tags) > 0 { + tags := make([]interface{}, 0, len(worker.Tags)) + for k, 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", server.PrivateId, k)) + return errors.New(ctx, errors.InvalidParameter, op+":RangeTags", fmt.Sprintf("found nil tag value for worker %s and key %s", worker.PrivateId, k)) } for _, val := range v.Values { - tags = append(tags, ServerTag{ - ServerId: server.PrivateId, + tags = append(tags, WorkerTag{ + WorkerId: worker.PrivateId, Key: k, Value: val, }) } } if err = w.CreateItems(ctx, tags); err != nil { - return errors.Wrap(ctx, err, op+":CreateTags", errors.WithMsg(server.PrivateId)) + return errors.Wrap(ctx, err, op+":CreateTags", errors.WithMsg(worker.PrivateId)) } } } @@ -207,6 +226,39 @@ func (r *Repository) UpsertServer(ctx context.Context, server *Server, opt ...Op return controllers, int(rowsUpdated), nil } +func (r *Repository) UpsertController(ctx context.Context, controller *store.Controller) (int, error) { + const op = "servers.UpsertController" + + if controller == nil { + return db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "controller is nil") + } + + var rowsUpdated int64 + _, err := r.writer.DoTx( + ctx, + db.StdRetryCnt, + db.ExpBackoff{}, + func(read db.Reader, w db.Writer) error { + var err error + onConflict := &db.OnConflict{ + Target: db.Columns{"private_id"}, + Action: append(db.SetColumns([]string{"description", "address"}), db.SetColumnValues(map[string]interface{}{"update_time": "now()"})...), + } + err = w.Create(ctx, controller, db.WithOnConflict(onConflict), db.WithReturnRowsAffected(&rowsUpdated)) + if err != nil { + return errors.Wrap(ctx, err, op+":Upsert") + } + + return nil + }, + ) + if err != nil { + return db.NoRowsAffected, err + } + + return int(rowsUpdated), nil +} + type Nonce struct { Nonce string Purpose string diff --git a/internal/servers/repository_test.go b/internal/servers/repository_test.go index 3ff57b655..74eba1f17 100644 --- a/internal/servers/repository_test.go +++ b/internal/servers/repository_test.go @@ -5,6 +5,8 @@ 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" @@ -93,55 +95,53 @@ func TestTagUpdatingListing(t *testing.T) { defer tc.Shutdown() repo := tc.ServersRepo() - srv := &servers.Server{ + srv := &store.Worker{ PrivateId: "test1", - Type: "worker", Address: "127.0.0.1", - Tags: map[string]*servers.TagValues{ + Tags: map[string]*store.TagValues{ "tag1": { Values: []string{"value1", "value2"}, }, }, } - _, _, err := repo.UpsertServer(tc.Context(), srv, servers.WithUpdateTags(true)) + _, _, err := repo.UpsertWorker(tc.Context(), srv, servers.WithUpdateTags(true)) require.NoError(err) - srv = &servers.Server{ + srv = &store.Worker{ PrivateId: "test2", - Type: "worker", Address: "127.0.0.1", - Tags: map[string]*servers.TagValues{ + Tags: map[string]*store.TagValues{ "tag2": { Values: []string{"value1", "value2"}, }, }, } - _, _, err = repo.UpsertServer(tc.Context(), srv, servers.WithUpdateTags(true)) + _, _, err = repo.UpsertWorker(tc.Context(), srv, servers.WithUpdateTags(true)) require.NoError(err) - tags, err := repo.ListTagsForServers(tc.Context(), []string{"test1", "test2"}) + tags, err := repo.ListTagsForWorkers(tc.Context(), []string{"test1", "test2"}) require.NoError(err) // Base case - exp := []*servers.ServerTag{ + exp := []*servers.WorkerTag{ { - ServerId: "test1", + WorkerId: "test1", Key: "tag1", Value: "value1", }, { - ServerId: "test1", + WorkerId: "test1", Key: "tag1", Value: "value2", }, { - ServerId: "test2", + WorkerId: "test2", Key: "tag2", Value: "value1", }, { - ServerId: "test2", + WorkerId: "test2", Key: "tag2", Value: "value2", }, @@ -149,36 +149,35 @@ func TestTagUpdatingListing(t *testing.T) { require.Equal(exp, tags) // Update without saying to update tags - srv = &servers.Server{ + srv = &store.Worker{ PrivateId: "test2", - Type: "worker", Address: "192.168.1.1", - Tags: map[string]*servers.TagValues{ + Tags: map[string]*store.TagValues{ "tag22": { Values: []string{"value21", "value22"}, }, }, } - _, _, err = repo.UpsertServer(tc.Context(), srv) + _, _, err = repo.UpsertWorker(tc.Context(), srv) require.NoError(err) - tags, err = repo.ListTagsForServers(tc.Context(), []string{"test1", "test2"}) + tags, err = repo.ListTagsForWorkers(tc.Context(), []string{"test1", "test2"}) require.NoError(err) require.Equal(exp, tags) // Update tags and test again - _, _, err = repo.UpsertServer(tc.Context(), srv, servers.WithUpdateTags(true)) + _, _, err = repo.UpsertWorker(tc.Context(), srv, servers.WithUpdateTags(true)) require.NoError(err) - tags, err = repo.ListTagsForServers(tc.Context(), []string{"test1", "test2"}) + tags, err = repo.ListTagsForWorkers(tc.Context(), []string{"test1", "test2"}) require.NoError(err) require.NotEqual(exp, tags) // Update and try again - exp[2] = &servers.ServerTag{ - ServerId: "test2", + exp[2] = &servers.WorkerTag{ + WorkerId: "test2", Key: "tag22", Value: "value21", } - exp[3] = &servers.ServerTag{ - ServerId: "test2", + exp[3] = &servers.WorkerTag{ + WorkerId: "test2", Key: "tag22", Value: "value22", } @@ -196,33 +195,32 @@ func TestListServersWithLiveness(t *testing.T) { require.NoError(err) ctx := context.Background() - newServer := func(privateId string) *servers.Server { - result := &servers.Server{ + newWorker := func(privateId string) *store.Worker { + result := &store.Worker{ PrivateId: privateId, - Type: "worker", Address: "127.0.0.1", } - _, rowsUpdated, err := serversRepo.UpsertServer(ctx, result) + _, rowsUpdated, err := serversRepo.UpsertWorker(ctx, result) require.NoError(err) require.Equal(1, rowsUpdated) return result } - server1 := newServer("test1") - server2 := newServer("test2") - server3 := newServer("test3") + server1 := newWorker("test1") + server2 := newWorker("test2") + server3 := newWorker("test3") // 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.UpsertServer(ctx, server1) + _, rowsUpdated, err := serversRepo.UpsertWorker(ctx, server1) require.NoError(err) require.Equal(1, rowsUpdated) - requireIds := func(expected []string, actual []*servers.Server) { + requireIds := func(expected []string, actual []*store.Worker) { require.Len(expected, len(actual)) want := make(map[string]struct{}) for _, v := range expected { @@ -238,25 +236,25 @@ func TestListServersWithLiveness(t *testing.T) { } // Default liveness, should only list 1 - result, err := serversRepo.ListServers(ctx, servers.ServerTypeWorker) + result, err := serversRepo.ListWorkers(ctx) require.NoError(err) require.Len(result, 1) requireIds([]string{server1.PrivateId}, result) // Upsert second server. - _, rowsUpdated, err = serversRepo.UpsertServer(ctx, server2) + _, rowsUpdated, err = serversRepo.UpsertWorker(ctx, server2) require.NoError(err) require.Equal(1, rowsUpdated) // Static liveness. Should get two, so long as this did not take // more than 5s to execute. - result, err = serversRepo.ListServers(ctx, servers.ServerTypeWorker, servers.WithLiveness(time.Second*5)) + result, err = serversRepo.ListWorkers(ctx, servers.WithLiveness(time.Second*5)) require.NoError(err) require.Len(result, 2) requireIds([]string{server1.PrivateId, server2.PrivateId}, result) - // Liveness disabled, should get all three servers. - result, err = serversRepo.ListServers(ctx, servers.ServerTypeWorker, servers.WithLiveness(-1)) + // 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.PrivateId, server2.PrivateId, server3.PrivateId}, result) diff --git a/internal/servers/store/controller.pb.go b/internal/servers/store/controller.pb.go new file mode 100644 index 000000000..c0d684a89 --- /dev/null +++ b/internal/servers/store/controller.pb.go @@ -0,0 +1,209 @@ +// Code generated by protoc-gen-go. DO NOT EDIT. +// versions: +// protoc-gen-go v1.27.1 +// protoc (unknown) +// source: controller/storage/servers/store/v1/controller.proto + +package store + +import ( + timestamp "github.com/hashicorp/boundary/internal/db/timestamp" + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" + reflect "reflect" + sync "sync" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +// Controller contains all fields related to a Controller resource +type Controller struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + // Private ID of the resource + PrivateId string `protobuf:"bytes,10,opt,name=private_id,json=privateId,proto3" json:"private_id,omitempty"` + // Address for the controller + Address string `protobuf:"bytes,20,opt,name=address,proto3" json:"address,omitempty"` + // 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"` + // First seen time from the RDBMS + CreateTime *timestamp.Timestamp `protobuf:"bytes,40,opt,name=create_time,json=createTime,proto3" json:"create_time,omitempty"` + // Last time there was an update + UpdateTime *timestamp.Timestamp `protobuf:"bytes,50,opt,name=update_time,json=updateTime,proto3" json:"update_time,omitempty"` +} + +func (x *Controller) Reset() { + *x = Controller{} + if protoimpl.UnsafeEnabled { + mi := &file_controller_storage_servers_store_v1_controller_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *Controller) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*Controller) ProtoMessage() {} + +func (x *Controller) ProtoReflect() protoreflect.Message { + mi := &file_controller_storage_servers_store_v1_controller_proto_msgTypes[0] + 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 Controller.ProtoReflect.Descriptor instead. +func (*Controller) Descriptor() ([]byte, []int) { + return file_controller_storage_servers_store_v1_controller_proto_rawDescGZIP(), []int{0} +} + +func (x *Controller) GetPrivateId() string { + if x != nil { + return x.PrivateId + } + return "" +} + +func (x *Controller) GetAddress() string { + if x != nil { + return x.Address + } + return "" +} + +func (x *Controller) GetDescription() string { + if x != nil { + return x.Description + } + return "" +} + +func (x *Controller) GetCreateTime() *timestamp.Timestamp { + if x != nil { + return x.CreateTime + } + return nil +} + +func (x *Controller) GetUpdateTime() *timestamp.Timestamp { + if x != nil { + return x.UpdateTime + } + return nil +} + +var File_controller_storage_servers_store_v1_controller_proto protoreflect.FileDescriptor + +var file_controller_storage_servers_store_v1_controller_proto_rawDesc = []byte{ + 0x0a, 0x34, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2f, 0x73, 0x74, 0x6f, + 0x72, 0x61, 0x67, 0x65, 0x2f, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2f, 0x73, 0x74, 0x6f, + 0x72, 0x65, 0x2f, 0x76, 0x31, 0x2f, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, + 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x23, 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, 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, 0x81, 0x02, 0x0a, + 0x0a, 0x43, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x12, 0x1d, 0x0a, 0x0a, 0x70, + 0x72, 0x69, 0x76, 0x61, 0x74, 0x65, 0x5f, 0x69, 0x64, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x09, 0x52, + 0x09, 0x70, 0x72, 0x69, 0x76, 0x61, 0x74, 0x65, 0x49, 0x64, 0x12, 0x18, 0x0a, 0x07, 0x61, 0x64, + 0x64, 0x72, 0x65, 0x73, 0x73, 0x18, 0x14, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x61, 0x64, 0x64, + 0x72, 0x65, 0x73, 0x73, 0x12, 0x20, 0x0a, 0x0b, 0x64, 0x65, 0x73, 0x63, 0x72, 0x69, 0x70, 0x74, + 0x69, 0x6f, 0x6e, 0x18, 0x1e, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0b, 0x64, 0x65, 0x73, 0x63, 0x72, + 0x69, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x4b, 0x0a, 0x0b, 0x63, 0x72, 0x65, 0x61, 0x74, 0x65, + 0x5f, 0x74, 0x69, 0x6d, 0x65, 0x18, 0x28, 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, 0x32, 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, + 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 ( + file_controller_storage_servers_store_v1_controller_proto_rawDescOnce sync.Once + file_controller_storage_servers_store_v1_controller_proto_rawDescData = file_controller_storage_servers_store_v1_controller_proto_rawDesc +) + +func file_controller_storage_servers_store_v1_controller_proto_rawDescGZIP() []byte { + file_controller_storage_servers_store_v1_controller_proto_rawDescOnce.Do(func() { + file_controller_storage_servers_store_v1_controller_proto_rawDescData = protoimpl.X.CompressGZIP(file_controller_storage_servers_store_v1_controller_proto_rawDescData) + }) + return file_controller_storage_servers_store_v1_controller_proto_rawDescData +} + +var file_controller_storage_servers_store_v1_controller_proto_msgTypes = make([]protoimpl.MessageInfo, 1) +var file_controller_storage_servers_store_v1_controller_proto_goTypes = []interface{}{ + (*Controller)(nil), // 0: controller.storage.servers.store.v1.Controller + (*timestamp.Timestamp)(nil), // 1: controller.storage.timestamp.v1.Timestamp +} +var file_controller_storage_servers_store_v1_controller_proto_depIdxs = []int32{ + 1, // 0: controller.storage.servers.store.v1.Controller.create_time:type_name -> controller.storage.timestamp.v1.Timestamp + 1, // 1: controller.storage.servers.store.v1.Controller.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_controller_proto_init() } +func file_controller_storage_servers_store_v1_controller_proto_init() { + if File_controller_storage_servers_store_v1_controller_proto != nil { + return + } + if !protoimpl.UnsafeEnabled { + file_controller_storage_servers_store_v1_controller_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*Controller); 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{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: file_controller_storage_servers_store_v1_controller_proto_rawDesc, + NumEnums: 0, + NumMessages: 1, + NumExtensions: 0, + NumServices: 0, + }, + GoTypes: file_controller_storage_servers_store_v1_controller_proto_goTypes, + DependencyIndexes: file_controller_storage_servers_store_v1_controller_proto_depIdxs, + MessageInfos: file_controller_storage_servers_store_v1_controller_proto_msgTypes, + }.Build() + File_controller_storage_servers_store_v1_controller_proto = out.File + file_controller_storage_servers_store_v1_controller_proto_rawDesc = nil + file_controller_storage_servers_store_v1_controller_proto_goTypes = nil + file_controller_storage_servers_store_v1_controller_proto_depIdxs = nil +} diff --git a/internal/servers/store/gorm.go b/internal/servers/store/gorm.go new file mode 100644 index 000000000..7ab3fc709 --- /dev/null +++ b/internal/servers/store/gorm.go @@ -0,0 +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" +} diff --git a/internal/servers/store/worker.pb.go b/internal/servers/store/worker.pb.go new file mode 100644 index 000000000..fd5504fb1 --- /dev/null +++ b/internal/servers/store/worker.pb.go @@ -0,0 +1,308 @@ +// Code generated by protoc-gen-go. DO NOT EDIT. +// versions: +// protoc-gen-go v1.27.1 +// protoc (unknown) +// source: controller/storage/servers/store/v1/worker.proto + +package store + +import ( + timestamp "github.com/hashicorp/boundary/internal/db/timestamp" + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" + reflect "reflect" + sync "sync" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +// Worker contains all fields related to a Worker resource +type Worker struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + // Private ID of the resource + PrivateId string `protobuf:"bytes,10,opt,name=private_id,json=privateId,proto3" json:"private_id,omitempty"` + // 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:"-"` +} + +func (x *Worker) Reset() { + *x = Worker{} + if protoimpl.UnsafeEnabled { + mi := &file_controller_storage_servers_store_v1_worker_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *Worker) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*Worker) ProtoMessage() {} + +func (x *Worker) ProtoReflect() protoreflect.Message { + mi := &file_controller_storage_servers_store_v1_worker_proto_msgTypes[0] + 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 Worker.ProtoReflect.Descriptor instead. +func (*Worker) Descriptor() ([]byte, []int) { + return file_controller_storage_servers_store_v1_worker_proto_rawDescGZIP(), []int{0} +} + +func (x *Worker) GetPrivateId() string { + if x != nil { + return x.PrivateId + } + return "" +} + +func (x *Worker) GetName() string { + if x != nil { + return x.Name + } + return "" +} + +func (x *Worker) GetDescription() string { + if x != nil { + return x.Description + } + return "" +} + +func (x *Worker) GetAddress() string { + if x != nil { + return x.Address + } + return "" +} + +func (x *Worker) GetCreateTime() *timestamp.Timestamp { + if x != nil { + return x.CreateTime + } + return nil +} + +func (x *Worker) GetUpdateTime() *timestamp.Timestamp { + if x != nil { + return x.UpdateTime + } + return nil +} + +func (x *Worker) GetTags() map[string]*TagValues { + if x != nil { + return x.Tags + } + return nil +} + +// TagValues is used because map fields cannot be repeated but can be a +// message +type TagValues struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + Values []string `protobuf:"bytes,10,rep,name=values,proto3" json:"values,omitempty"` +} + +func (x *TagValues) Reset() { + *x = TagValues{} + if protoimpl.UnsafeEnabled { + mi := &file_controller_storage_servers_store_v1_worker_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *TagValues) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*TagValues) ProtoMessage() {} + +func (x *TagValues) 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)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use TagValues.ProtoReflect.Descriptor instead. +func (*TagValues) Descriptor() ([]byte, []int) { + return file_controller_storage_servers_store_v1_worker_proto_rawDescGZIP(), []int{1} +} + +func (x *TagValues) GetValues() []string { + if x != nil { + return x.Values + } + return nil +} + +var File_controller_storage_servers_store_v1_worker_proto protoreflect.FileDescriptor + +var file_controller_storage_servers_store_v1_worker_proto_rawDesc = []byte{ + 0x0a, 0x30, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2f, 0x73, 0x74, 0x6f, + 0x72, 0x61, 0x67, 0x65, 0x2f, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2f, 0x73, 0x74, 0x6f, + 0x72, 0x65, 0x2f, 0x76, 0x31, 0x2f, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x2e, 0x70, 0x72, 0x6f, + 0x74, 0x6f, 0x12, 0x23, 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, 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, 0xc5, 0x03, 0x0a, 0x06, 0x57, 0x6f, 0x72, + 0x6b, 0x65, 0x72, 0x12, 0x1d, 0x0a, 0x0a, 0x70, 0x72, 0x69, 0x76, 0x61, 0x74, 0x65, 0x5f, 0x69, + 0x64, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x70, 0x72, 0x69, 0x76, 0x61, 0x74, 0x65, + 0x49, 0x64, 0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x14, 0x20, 0x01, 0x28, 0x09, + 0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x12, 0x20, 0x0a, 0x0b, 0x64, 0x65, 0x73, 0x63, 0x72, 0x69, + 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x1e, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0b, 0x64, 0x65, 0x73, + 0x63, 0x72, 0x69, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x18, 0x0a, 0x07, 0x61, 0x64, 0x64, 0x72, + 0x65, 0x73, 0x73, 0x18, 0x28, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x61, 0x64, 0x64, 0x72, 0x65, + 0x73, 0x73, 0x12, 0x4b, 0x0a, 0x0b, 0x63, 0x72, 0x65, 0x61, 0x74, 0x65, 0x5f, 0x74, 0x69, 0x6d, + 0x65, 0x18, 0x32, 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, 0x3c, + 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, 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, +} + +var ( + file_controller_storage_servers_store_v1_worker_proto_rawDescOnce sync.Once + file_controller_storage_servers_store_v1_worker_proto_rawDescData = file_controller_storage_servers_store_v1_worker_proto_rawDesc +) + +func file_controller_storage_servers_store_v1_worker_proto_rawDescGZIP() []byte { + file_controller_storage_servers_store_v1_worker_proto_rawDescOnce.Do(func() { + file_controller_storage_servers_store_v1_worker_proto_rawDescData = protoimpl.X.CompressGZIP(file_controller_storage_servers_store_v1_worker_proto_rawDescData) + }) + 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_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 +} +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 +} + +func init() { file_controller_storage_servers_store_v1_worker_proto_init() } +func file_controller_storage_servers_store_v1_worker_proto_init() { + if File_controller_storage_servers_store_v1_worker_proto != nil { + return + } + if !protoimpl.UnsafeEnabled { + file_controller_storage_servers_store_v1_worker_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*Worker); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + file_controller_storage_servers_store_v1_worker_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*TagValues); 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{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: file_controller_storage_servers_store_v1_worker_proto_rawDesc, + NumEnums: 0, + NumMessages: 3, + NumExtensions: 0, + NumServices: 0, + }, + GoTypes: file_controller_storage_servers_store_v1_worker_proto_goTypes, + DependencyIndexes: file_controller_storage_servers_store_v1_worker_proto_depIdxs, + MessageInfos: file_controller_storage_servers_store_v1_worker_proto_msgTypes, + }.Build() + File_controller_storage_servers_store_v1_worker_proto = out.File + file_controller_storage_servers_store_v1_worker_proto_rawDesc = nil + file_controller_storage_servers_store_v1_worker_proto_goTypes = nil + file_controller_storage_servers_store_v1_worker_proto_depIdxs = nil +} diff --git a/internal/servers/worker/status.go b/internal/servers/worker/status.go index 88b349f2e..4bb9481c5 100644 --- a/internal/servers/worker/status.go +++ b/internal/servers/worker/status.go @@ -6,12 +6,12 @@ import ( "math/rand" "time" + "github.com/hashicorp/boundary/internal/servers" + pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services" "github.com/hashicorp/boundary/internal/observability/event" - "github.com/hashicorp/boundary/internal/servers" "github.com/hashicorp/boundary/internal/servers/worker/common" "github.com/hashicorp/boundary/internal/servers/worker/session" - "github.com/hashicorp/boundary/internal/types/resource" "google.golang.org/grpc/resolver" ) @@ -143,14 +143,13 @@ func (w *Worker) sendWorkerStatus(cancelCtx context.Context) { } statusCtx, statusCancel := context.WithTimeout(cancelCtx, common.StatusTimeout) defer statusCancel() + result, err := client.Status(statusCtx, &pbs.StatusRequest{ Jobs: activeJobs, Worker: &servers.Server{ - PrivateId: w.conf.RawConfig.Worker.Name, - Type: resource.Worker.String(), - Description: w.conf.RawConfig.Worker.Description, - Address: w.conf.RawConfig.Worker.PublicAddr, - Tags: tags, + PrivateId: w.conf.RawConfig.Worker.Name, + Address: w.conf.RawConfig.Worker.PublicAddr, + Tags: tags, }, UpdateTags: w.updateTags.Load(), }) diff --git a/internal/servers/worker/worker.go b/internal/servers/worker/worker.go index 8f87cf731..7a7a95c14 100644 --- a/internal/servers/worker/worker.go +++ b/internal/servers/worker/worker.go @@ -14,12 +14,13 @@ import ( "sync/atomic" "time" + "github.com/hashicorp/boundary/internal/servers" + "github.com/hashicorp/boundary/globals" "github.com/hashicorp/boundary/internal/cmd/base" "github.com/hashicorp/boundary/internal/cmd/config" pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services" "github.com/hashicorp/boundary/internal/observability/event" - "github.com/hashicorp/boundary/internal/servers" "github.com/hashicorp/boundary/internal/servers/worker/internal/metric" "github.com/hashicorp/boundary/internal/servers/worker/session" "github.com/hashicorp/go-hclog" diff --git a/internal/session/options.go b/internal/session/options.go index a4e5a335b..1ae61bbd9 100644 --- a/internal/session/options.go +++ b/internal/session/options.go @@ -21,17 +21,18 @@ type Option func(*options) // options = how options are represented type options struct { - withLimit int - withOrderByCreateTime db.OrderBy - withScopeIds []string - withUserId string - withExpirationTime *timestamp.Timestamp - withTestTofu []byte - withListingConvert bool - withSessionIds []string - withServerId string - withDbOpts []db.Option - withWorkerStateDelay time.Duration + withLimit int + withOrderByCreateTime db.OrderBy + withScopeIds []string + withUserId string + withExpirationTime *timestamp.Timestamp + withTestTofu []byte + withListingConvert bool + withSessionIds []string + withWorkerId string + withDbOpts []db.Option + withWorkerStateDelay time.Duration + withDeadWorkerConnCloseMinGrace time.Duration } func getDefaultOptions() options { @@ -99,11 +100,10 @@ func withListingConvert(withListingConvert bool) Option { } } -// WithServerId allows the specification of the server id to use for the -// operation. -func WithServerId(id string) Option { +// WithWorkerId allows the specification of the worker id to use +func WithWorkerId(id string) Option { return func(o *options) { - o.withServerId = id + o.withWorkerId = id } } diff --git a/internal/session/options_test.go b/internal/session/options_test.go index 975a2f93f..890d4c390 100644 --- a/internal/session/options_test.go +++ b/internal/session/options_test.go @@ -75,11 +75,11 @@ func Test_GetOpts(t *testing.T) { testOpts.withSessionIds = []string{"s_1", "s_2", "s_3"} assert.Equal(opts, testOpts) }) - t.Run("WithServerId", func(t *testing.T) { + t.Run("WithWorkerId", func(t *testing.T) { assert := assert.New(t) - opts := getOpts(WithServerId("worker1")) + opts := getOpts(WithWorkerId("worker1")) testOpts := getDefaultOptions() - testOpts.withServerId = "worker1" + testOpts.withWorkerId = "worker1" assert.Equal(opts, testOpts) }) } diff --git a/internal/session/query.go b/internal/session/query.go index 7a10af70f..919980c55 100644 --- a/internal/session/query.go +++ b/internal/session/query.go @@ -81,7 +81,7 @@ active_session as ( select ss.session_id as session_id, @public_id as public_id, - @worker_id as server_id + @worker_id as worker_id from session_state ss where @@ -92,7 +92,7 @@ active_session as ( insert into session_connection ( session_id, public_id, - server_id + worker_id ) select * from active_session; ` @@ -323,26 +323,26 @@ where // each. closeConnectionsForDeadServersCte = ` with - dead_servers (server_id, last_update_time) as ( + dead_workers (worker_id, last_update_time) as ( select private_id, update_time - from server + from server_worker where update_time < wt_sub_seconds_from_now(@grace_period_seconds) ), - closed_connections (connection_id, server_id) as ( + closed_connections (connection_id, worker_id) as ( update session_connection set closed_reason = 'system error' - where server_id in (select server_id from dead_servers) + where worker_id in (select worker_id from dead_workers) and closed_reason is null - returning public_id, server_id + returning public_id, worker_id ) - select closed_connections.server_id, - dead_servers.last_update_time, + select closed_connections.worker_id, + dead_workers.last_update_time, count(closed_connections.connection_id) as number_connections_closed from closed_connections - join dead_servers - on closed_connections.server_id = dead_servers.server_id - group by closed_connections.server_id, dead_servers.last_update_time - order by closed_connections.server_id; + join dead_workers + on closed_connections.worker_id = dead_workers.worker_id + group by closed_connections.worker_id, dead_workers.last_update_time + order by closed_connections.worker_id; ` orphanedConnectionsCte = ` @@ -366,7 +366,7 @@ with from session_connection where -- Related to the worker that just reported to us - server_id = @server_id + worker_id = @worker_id -- Only unclosed ones and public_id in (select connection_id from unclosed_connections) -- These are connection IDs that just got reported to us by the given diff --git a/internal/session/repository.go b/internal/session/repository.go index 8735c8d7f..3e0948c30 100644 --- a/internal/session/repository.go +++ b/internal/session/repository.go @@ -110,8 +110,7 @@ func (r *Repository) convertToSessions(ctx context.Context, sessionList []*sessi PublicId: sv.PublicId, UserId: sv.UserId, HostId: sv.HostId, - ServerId: sv.ServerId, - ServerType: sv.ServerType, + WorkerId: sv.WorkerId, TargetId: sv.TargetId, HostSetId: sv.HostSetId, AuthTokenId: sv.AuthTokenId, diff --git a/internal/session/repository_connection.go b/internal/session/repository_connection.go index 808a2333a..6b3845b10 100644 --- a/internal/session/repository_connection.go +++ b/internal/session/repository_connection.go @@ -345,6 +345,59 @@ func (r *ConnectionRepository) DeleteConnection(ctx context.Context, publicId st return rowsDeleted, nil } +type CloseConnectionsForDeadWorkersResult struct { + WorkerId string + LastUpdateTime time.Time + NumberConnectionsClosed int +} + +// CloseConnectionsForDeadWorkers will run +// closeConnectionsForDeadServersCte to look for connections that +// should be marked because they are on a worker that is no longer +// sending status updates to the controller(s). +// +// The only input to the method is the grace period, in seconds. +func (r *ConnectionRepository) CloseConnectionsForDeadWorkers(ctx context.Context, gracePeriod time.Duration) ([]CloseConnectionsForDeadWorkersResult, error) { + const op = "session.(ConnectionRepository).CloseConnectionsForDeadWorkers" + if gracePeriod < r.deadWorkerConnCloseMinGrace { + return nil, errors.New(ctx, + errors.InvalidParameter, op, fmt.Sprintf("gracePeriod must be at least %s", r.deadWorkerConnCloseMinGrace)) + } + + args := []interface{}{ + sql.Named("grace_period_seconds", gracePeriod.Seconds()), + } + results := make([]CloseConnectionsForDeadWorkersResult, 0) + _, err := r.writer.DoTx( + ctx, + db.StdRetryCnt, + db.ExpBackoff{}, + func(reader db.Reader, w db.Writer) error { + rows, err := w.Query(ctx, closeConnectionsForDeadServersCte, args) + if err != nil { + return errors.Wrap(ctx, err, op) + } + defer rows.Close() + + for rows.Next() { + var result CloseConnectionsForDeadWorkersResult + if err := w.ScanRows(ctx, rows, &result); err != nil { + return errors.Wrap(ctx, err, op) + } + + results = append(results, result) + } + + return nil + }, + ) + if err != nil { + return nil, errors.Wrap(ctx, err, op) + } + + return results, nil +} + // closeOrphanedConnections looks for connections that are still active, but where not reported by the worker. func (r *ConnectionRepository) closeOrphanedConnections(ctx context.Context, workerId string, reportedConnections []string) ([]string, error) { const op = "session.(ConnectionRepository).closeOrphanedConnections" @@ -352,7 +405,7 @@ func (r *ConnectionRepository) closeOrphanedConnections(ctx context.Context, wor var orphanedConns []string args := make([]interface{}, 0, len(reportedConnections)+2) - args = append(args, sql.Named("server_id", workerId)) + args = append(args, sql.Named("worker_id", workerId)) args = append(args, sql.Named("worker_state_delay_seconds", r.workerStateDelay.Seconds())) var notInClause string diff --git a/internal/session/repository_connection_test.go b/internal/session/repository_connection_test.go index 164d0155c..4bb5cbbd4 100644 --- a/internal/session/repository_connection_test.go +++ b/internal/session/repository_connection_test.go @@ -6,6 +6,9 @@ import ( "testing" "time" + "github.com/hashicorp/boundary/internal/servers" + "github.com/hashicorp/boundary/internal/servers/store" + "github.com/hashicorp/boundary/internal/db" "github.com/hashicorp/boundary/internal/errors" "github.com/hashicorp/boundary/internal/iam" @@ -149,7 +152,7 @@ func TestRepository_ConnectConnection(t *testing.T) { s := TestDefaultSession(t, conn, wrapper, iamRepo) srv := TestWorker(t, conn, wrapper) tofu := TestTofu(t) - _, _, err := repo.ActivateSession(context.Background(), s.PublicId, s.Version, srv.PrivateId, srv.Type, tofu) + _, _, err := repo.ActivateSession(context.Background(), s.PublicId, s.Version, srv.PrivateId, tofu) require.NoError(t, err) c := TestConnection(t, conn, s.PublicId, "127.0.0.1", 22, "127.0.0.1", 2222, "127.0.0.1") return ConnectWith{ @@ -359,8 +362,8 @@ func TestRepository_orphanedConnections(t *testing.T) { // Create two "workers". One will remain untouched while the other "goes // away and comes back" (worker 2). - worker1 := TestWorker(t, conn, wrapper, WithServerId("worker1")) - worker2 := TestWorker(t, conn, wrapper, WithServerId("worker2")) + worker1 := TestWorker(t, conn, wrapper, WithWorkerId("worker1")) + worker2 := TestWorker(t, conn, wrapper, WithWorkerId("worker2")) // Create a few sessions on each, activate, and authorize a connection var connIds []string @@ -371,8 +374,8 @@ func TestRepository_orphanedConnections(t *testing.T) { if i%2 == 0 { serverId = worker2.PrivateId } - sess := TestDefaultSession(t, conn, wrapper, iamRepo, WithServerId(serverId), WithDbOpts(db.WithSkipVetForWrite(true))) - sess, _, err = repo.ActivateSession(ctx, sess.GetPublicId(), sess.Version, serverId, "worker", []byte("foo")) + sess := TestDefaultSession(t, conn, wrapper, iamRepo, WithWorkerId(serverId), WithDbOpts(db.WithSkipVetForWrite(true))) + sess, _, err = repo.ActivateSession(ctx, sess.GetPublicId(), sess.Version, serverId, []byte("foo")) require.NoError(err) c, cs, err := connRepo.AuthorizeConnection(ctx, sess.GetPublicId(), serverId) require.NoError(err) @@ -438,6 +441,287 @@ func TestRepository_orphanedConnections(t *testing.T) { assert.ElementsMatch(found, worker1ConnIds) } +func TestRepository_CloseConnectionsForDeadWorkers(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) + deadWorkerConnCloseMinGrace := 1 * time.Second + connRepo, err := NewConnectionRepository(ctx, rw, rw, kms, WithDeadWorkerConnCloseMinGrace(deadWorkerConnCloseMinGrace)) + require.NoError(err) + serversRepo, err := servers.NewRepository(rw, rw, kms) + require.NoError(err) + + // connection count = 6 * states(authorized, connected, closed = 3) * servers_with_open_connections(3) + numConns := 54 + + // Create four "workers". This is similar to the setup in + // TestRepository_CloseDeadConnectionsOnWorker, but a bit more complex; + // firstly, the last worker will have no connections at all, and we will be + // closing the others in stages to test multiple servers being closed at + // once. + worker1 := TestWorker(t, conn, wrapper, WithWorkerId("worker1")) + worker2 := TestWorker(t, conn, wrapper, WithWorkerId("worker2")) + worker3 := TestWorker(t, conn, wrapper, WithWorkerId("worker3")) + worker4 := TestWorker(t, conn, wrapper, WithWorkerId("worker4")) + + // Create sessions on the first three, activate, and authorize connections + var worker1ConnIds, worker2ConnIds, worker3ConnIds []string + for i := 0; i < numConns; i++ { + var serverId string + if i%3 == 0 { + serverId = worker1.PrivateId + } else if i%3 == 1 { + serverId = worker2.PrivateId + } else { + serverId = worker3.PrivateId + } + sess := TestDefaultSession(t, conn, wrapper, iamRepo, WithWorkerId(serverId), WithDbOpts(db.WithSkipVetForWrite(true))) + sess, _, err = repo.ActivateSession(ctx, sess.GetPublicId(), sess.Version, serverId, []byte("foo")) + require.NoError(err) + c, cs, err := connRepo.AuthorizeConnection(ctx, sess.GetPublicId(), serverId) + require.NoError(err) + require.Len(cs, 1) + require.Equal(StatusAuthorized, cs[0].Status) + if i%3 == 0 { + worker1ConnIds = append(worker1ConnIds, c.GetPublicId()) + } else if i%3 == 1 { + worker2ConnIds = append(worker2ConnIds, c.GetPublicId()) + } else { + worker3ConnIds = append(worker3ConnIds, c.GetPublicId()) + } + } + + // Mark a third of the connections connected, a third closed, and leave the + // others authorized. This is just to ensure we have a spread when we test it + // out. + for i, connId := range func() []string { + var s []string + s = append(s, worker1ConnIds...) + s = append(s, worker2ConnIds...) + s = append(s, worker3ConnIds...) + return s + }() { + if i%3 == 0 { + _, cs, err := connRepo.ConnectConnection(ctx, ConnectWith{ + ConnectionId: connId, + ClientTcpAddress: "127.0.0.1", + ClientTcpPort: 22, + EndpointTcpAddress: "127.0.0.1", + EndpointTcpPort: 22, + UserClientIp: "127.0.0.1", + }) + require.NoError(err) + require.Len(cs, 2) + var foundAuthorized, foundConnected bool + for _, status := range cs { + if status.Status == StatusAuthorized { + foundAuthorized = true + } + if status.Status == StatusConnected { + foundConnected = true + } + } + require.True(foundAuthorized) + require.True(foundConnected) + } else if i%3 == 1 { + resp, err := connRepo.closeConnections(ctx, []CloseWith{ + { + ConnectionId: connId, + ClosedReason: ConnectionCanceled, + }, + }) + require.NoError(err) + require.Len(resp, 1) + cs := resp[0].ConnectionStates + require.Len(cs, 2) + var foundAuthorized, foundClosed bool + for _, status := range cs { + if status.Status == StatusAuthorized { + foundAuthorized = true + } + if status.Status == StatusClosed { + foundClosed = true + } + } + require.True(foundAuthorized) + require.True(foundClosed) + } + } + + // updateWorker 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. + updateWorker := func(t *testing.T, w *store.Worker) *store.Worker { + t.Helper() + _, rowsUpdated, err := serversRepo.UpsertWorker(ctx, w) + require.NoError(err) + require.Equal(1, rowsUpdated) + servers, err := serversRepo.ListWorkers(ctx) + require.NoError(err) + for _, server := range servers { + if server.PrivateId == w.PrivateId { + return server + } + } + + require.FailNowf("server %q not found after updating", w.PrivateId) + // Looks weird but needed to build, as we fail in testify instead + // of returning an error + return nil + } + + // requireConnectionStatus is a helper expecting all connections on a worker + // to be closed. + requireConnectionStatus := func(t *testing.T, connIds []string, expectAllClosed bool) { + t.Helper() + + var conns []*Connection + require.NoError(repo.list(ctx, &conns, "", nil)) + for i, connId := range connIds { + var expected ConnectionStatus + switch { + case expectAllClosed: + expected = StatusClosed + + case i%3 == 0: + expected = StatusConnected + + case i%3 == 1: + expected = StatusClosed + + case i%3 == 2: + expected = StatusAuthorized + } + + _, states, err := connRepo.LookupConnection(ctx, connId) + require.NoError(err) + require.Equal(expected, states[0].Status, "expected latest status for %q (index %d) to be %v", connId, i, expected) + } + } + + // We need this helper to fix the zone on protobuf timestamps + // versus what gets reported in the + // CloseConnectionsForDeadWorkersResult. + timestampPbAsUTC := func(t *testing.T, tm time.Time) time.Time { + t.Helper() + // utcLoc, err := time.LoadLocation("Etc/UTC") + // require.NoError(err) + return tm.In(time.Local) + } + + // Now try some scenarios. + { + // First, test the error/validation case. + result, err := connRepo.CloseConnectionsForDeadWorkers(ctx, -1) + require.Equal(err, errors.E(ctx, + errors.WithCode(errors.InvalidParameter), + errors.WithOp("session.(ConnectionRepository).CloseConnectionsForDeadWorkers"), + errors.WithMsg(fmt.Sprintf("gracePeriod must be at least %s", deadWorkerConnCloseMinGrace)), + )) + require.Nil(result) + } + + { + // Now, try the basis, or where all workers are reporting in. + worker1 = updateWorker(t, worker1) + worker2 = updateWorker(t, worker2) + worker3 = updateWorker(t, worker3) + updateWorker(t, worker4) // no re-assignment here because we never reference the server again + + result, err := connRepo.CloseConnectionsForDeadWorkers(ctx, deadWorkerConnCloseMinGrace) + require.NoError(err) + require.Empty(result) + // Expect appropriate split connection state on worker1 + requireConnectionStatus(t, worker1ConnIds, false) + // Expect appropriate split connection state on worker2 + requireConnectionStatus(t, worker2ConnIds, false) + // Expect appropriate split connection state on worker3 + requireConnectionStatus(t, worker3ConnIds, false) + } + + { + // Now try a zero case - similar to the basis, but only in that no results + // are expected to be returned for workers with no connections, even if + // they are dead. Here, the server with no connections is worker #4. + time.Sleep(deadWorkerConnCloseMinGrace) + worker1 = updateWorker(t, worker1) + worker2 = updateWorker(t, worker2) + worker3 = updateWorker(t, worker3) + + result, err := connRepo.CloseConnectionsForDeadWorkers(ctx, deadWorkerConnCloseMinGrace) + require.NoError(err) + require.Empty(result) + // Expect appropriate split connection state on worker1 + requireConnectionStatus(t, worker1ConnIds, false) + // Expect appropriate split connection state on worker2 + requireConnectionStatus(t, worker2ConnIds, false) + // Expect appropriate split connection state on worker3 + requireConnectionStatus(t, worker3ConnIds, false) + } + + { + // The first induction is letting the first worker "die" by not updating it + // too. All of its authorized and connected connections should be dead. + time.Sleep(deadWorkerConnCloseMinGrace) + worker2 = updateWorker(t, worker2) + worker3 = updateWorker(t, worker3) + + result, err := connRepo.CloseConnectionsForDeadWorkers(ctx, deadWorkerConnCloseMinGrace) + require.NoError(err) + // Assert that we have one result with the appropriate ID and + // number of connections closed. Due to how things are + require.Equal([]CloseConnectionsForDeadWorkersResult{ + { + WorkerId: worker1.PrivateId, + LastUpdateTime: timestampPbAsUTC(t, worker1.UpdateTime.AsTime()), + NumberConnectionsClosed: 12, // 18 per server, with 6 closed already + }, + }, result) + // Expect all connections closed on worker1 + requireConnectionStatus(t, worker1ConnIds, true) + // Expect appropriate split connection state on worker2 + requireConnectionStatus(t, worker2ConnIds, false) + // Expect appropriate split connection state on worker3 + requireConnectionStatus(t, worker3ConnIds, false) + } + + { + // The final case is having the other two workers die. After + // this, we should have all connections closed with the + // appropriate message from the next two servers acted on. + time.Sleep(deadWorkerConnCloseMinGrace) + + result, err := connRepo.CloseConnectionsForDeadWorkers(ctx, deadWorkerConnCloseMinGrace) + require.NoError(err) + // Assert that we have one result with the appropriate ID and number of connections closed. + require.Equal([]CloseConnectionsForDeadWorkersResult{ + { + WorkerId: worker2.PrivateId, + LastUpdateTime: timestampPbAsUTC(t, worker2.UpdateTime.AsTime()), + NumberConnectionsClosed: 12, // 18 per server, with 6 closed already + }, + { + WorkerId: worker3.PrivateId, + LastUpdateTime: timestampPbAsUTC(t, worker3.UpdateTime.AsTime()), + NumberConnectionsClosed: 12, // 18 per server, with 6 closed already + }, + }, result) + // Expect all connections closed on worker1 + requireConnectionStatus(t, worker1ConnIds, true) + // Expect all connections closed on worker2 + requireConnectionStatus(t, worker2ConnIds, true) + // Expect all connections closed on worker3 + requireConnectionStatus(t, worker3ConnIds, true) + } +} + func TestRepository_CloseConnections(t *testing.T) { t.Parallel() ctx := context.Background() @@ -455,7 +739,7 @@ func TestRepository_CloseConnections(t *testing.T) { s := TestDefaultSession(t, conn, wrapper, iamRepo) srv := TestWorker(t, conn, wrapper) tofu := TestTofu(t) - s, _, err := repo.ActivateSession(context.Background(), s.PublicId, s.Version, srv.PrivateId, srv.Type, tofu) + s, _, err := repo.ActivateSession(context.Background(), s.PublicId, s.Version, srv.PrivateId, tofu) require.NoError(t, err) cw := make([]CloseWith, 0, cnt) for i := 0; i < cnt; i++ { diff --git a/internal/session/repository_session.go b/internal/session/repository_session.go index d16edfa21..53448a806 100644 --- a/internal/session/repository_session.go +++ b/internal/session/repository_session.go @@ -18,7 +18,7 @@ import ( // CreateSession inserts into the repository and returns the new Session with // its State of "Pending". The following fields must be empty when creating a -// session: ServerId, ServerType, and PublicId. No options are +// session: WorkerId, and PublicId. No options are // currently supported. func (r *Repository) CreateSession(ctx context.Context, sessionWrapper wrapping.Wrapper, newSession *Session, workerAddresses []string, _ ...Option) (*Session, ed25519.PrivateKey, error) { const op = "session.(Repository).CreateSession" @@ -49,11 +49,8 @@ func (r *Repository) CreateSession(ctx context.Context, sessionWrapper wrapping. if newSession.ScopeId == "" { return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing scope id") } - if newSession.ServerId != "" { - return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "server id is not empty") - } - if newSession.ServerType != "" { - return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "server type is not empty") + if newSession.WorkerId != "" { + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "worker id is not empty") } if newSession.CtTofuToken != nil { return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "ct is not empty") @@ -251,7 +248,7 @@ func (r *Repository) FetchAuthzProtectedEntitiesByScope(ctx context.Context, sco return sessionsMap, nil } -// ListSessions will sessions. Supports the WithLimit, WithScopeId, WithSessionIds, and WithServerId options. +// ListSessions lists sessions. Supports the WithLimit, WithScopeId, WithSessionIds, and WithWorkerId options. func (r *Repository) ListSessions(ctx context.Context, opt ...Option) ([]*Session, error) { const op = "session.(Repository).ListSessions" opts := getOpts(opt...) @@ -292,9 +289,9 @@ func (r *Repository) ListSessions(ctx context.Context, opt ...Option) ([]*Sessio where = append(where, fmt.Sprintf("s.public_id in (%s)", strings.Join(idsInClause, ","))) } - if opts.withServerId != "" { + if opts.withWorkerId != "" { inClauseCnt += 1 - where, args = append(where, fmt.Sprintf("server_id = @%d", inClauseCnt)), append(args, sql.Named(fmt.Sprintf("%d", inClauseCnt), opts.withServerId)) + where, args = append(where, fmt.Sprintf("worker_id = @%d", inClauseCnt)), append(args, sql.Named(fmt.Sprintf("%d", inClauseCnt), opts.withWorkerId)) } var limit string @@ -495,7 +492,7 @@ func (r *Repository) sessionAuthzSummary(ctx context.Context, sessionId string) // activated. States are ordered by start time descending. Returns an // InvalidSessionState error code if a connection cannot be made because the session // was canceled or terminated. -func (r *Repository) ActivateSession(ctx context.Context, sessionId string, sessionVersion uint32, serverId, serverType string, tofuToken []byte) (*Session, []*State, error) { +func (r *Repository) ActivateSession(ctx context.Context, sessionId string, sessionVersion uint32, workerId string, tofuToken []byte) (*Session, []*State, error) { const op = "session.(Repository).ActivateSession" if sessionId == "" { return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing session id") @@ -503,11 +500,8 @@ func (r *Repository) ActivateSession(ctx context.Context, sessionId string, sess if sessionVersion == 0 { return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing version") } - if serverId == "" { - return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing server id") - } - if serverType == "" { - return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing server type") + if workerId == "" { + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing worker id") } if len(tofuToken) == 0 { return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing tofu token") @@ -545,8 +539,7 @@ func (r *Repository) ActivateSession(ctx context.Context, sessionId string, sess } updatedSession.TofuToken = tofuToken - updatedSession.ServerId = serverId - updatedSession.ServerType = serverType + updatedSession.WorkerId = workerId if err := updatedSession.encrypt(ctx, databaseWrapper); err != nil { return errors.Wrap(ctx, err, op) } diff --git a/internal/session/repository_session_test.go b/internal/session/repository_session_test.go index cc9a770c6..1c9172351 100644 --- a/internal/session/repository_session_test.go +++ b/internal/session/repository_session_test.go @@ -215,10 +215,10 @@ func TestRepository_ListSession(t *testing.T) { db.TestDeleteWhere(t, conn, func() interface{} { i := AllocSession(); return &i }(), "1=1") for i := 0; i < 6; i++ { if i%2 == 0 { - TestWorker(t, conn, wrapper, WithServerId(fmt.Sprintf("server-%d", i/2))) + TestWorker(t, conn, wrapper, WithWorkerId(fmt.Sprintf("server-%d", i/2))) } _ = TestSession(t, conn, wrapper, composedOf, - WithServerId(fmt.Sprintf("server-%d", i/2)), + WithWorkerId(fmt.Sprintf("server-%d", i/2)), WithDbOpts(db.WithSkipVetForWrite(true)), ) } @@ -228,11 +228,11 @@ func TestRepository_ListSession(t *testing.T) { for i := 0; i < 3; i++ { serverId := fmt.Sprintf("server-%d", i) - got, err = repo.ListSessions(context.Background(), WithServerId(serverId)) + got, err = repo.ListSessions(context.Background(), WithWorkerId(serverId)) require.NoError(err) assert.Equal(2, len(got)) for _, item := range got { - assert.Equal(serverId, item.ServerId) + assert.Equal(serverId, item.WorkerId) } } }) @@ -683,7 +683,7 @@ func TestRepository_TerminateCompletedSessions(t *testing.T) { srv := TestWorker(t, conn, wrapper) tofu := TestTofu(t) - s, _, err = repo.ActivateSession(context.Background(), s.PublicId, s.Version, srv.PrivateId, srv.Type, tofu) + s, _, err = repo.ActivateSession(context.Background(), s.PublicId, s.Version, srv.PrivateId, tofu) require.NoError(t, err) c := TestConnection(t, conn, s.PublicId, "127.0.0.1", 22, "127.0.0.1", 222, "127.0.0.1") if !leaveOpen { @@ -1247,7 +1247,7 @@ func TestRepository_ActivateSession(t *testing.T) { name: "already-active", session: func() *Session { s := TestDefaultSession(t, conn, wrapper, iamRepo) - activeSession, _, err := repo.ActivateSession(context.Background(), s.PublicId, s.Version, worker.PrivateId, worker.Type, tofu) + activeSession, _, err := repo.ActivateSession(context.Background(), s.PublicId, s.Version, worker.PrivateId, tofu) require.NoError(t, err) return activeSession }(), @@ -1310,7 +1310,7 @@ func TestRepository_ActivateSession(t *testing.T) { default: version = tt.session.Version } - s, ss, err := repo.ActivateSession(context.Background(), id, version, worker.PrivateId, worker.Type, tofu) + s, ss, err := repo.ActivateSession(context.Background(), id, version, worker.PrivateId, tofu) if tt.wantErr { require.Error(err) assert.Truef(errors.Match(errors.T(tt.wantIsError), err), "unexpected error %s", err.Error()) @@ -1326,17 +1326,17 @@ func TestRepository_ActivateSession(t *testing.T) { t.Run("already active", func(t *testing.T) { assert, require := assert.New(t), require.New(t) session := TestDefaultSession(t, conn, wrapper, iamRepo) - s, ss, err := repo.ActivateSession(context.Background(), session.PublicId, 1, worker.PrivateId, worker.Type, tofu) + s, ss, err := repo.ActivateSession(context.Background(), session.PublicId, 1, worker.PrivateId, tofu) require.NoError(err) require.NotNil(s) require.NotNil(ss) assert.Equal(2, len(ss)) assert.Equal(StatusActive, ss[0].Status) - _, _, err = repo.ActivateSession(context.Background(), session.PublicId, 1, worker.PrivateId, worker.Type, tofu) + _, _, err = repo.ActivateSession(context.Background(), session.PublicId, 1, worker.PrivateId, tofu) require.Error(err) - _, _, err = repo.ActivateSession(context.Background(), session.PublicId, 2, worker.PrivateId, worker.Type, tofu) + _, _, err = repo.ActivateSession(context.Background(), session.PublicId, 2, worker.PrivateId, tofu) require.Error(err) }) } diff --git a/internal/session/service_authorize_connection_test.go b/internal/session/service_authorize_connection_test.go index 0cd6e86a2..a6f3ad381 100644 --- a/internal/session/service_authorize_connection_test.go +++ b/internal/session/service_authorize_connection_test.go @@ -38,7 +38,7 @@ func TestService_AuthorizeConnection(t *testing.T) { srv := TestWorker(t, conn, wrapper) testServer = srv.PrivateId tofu := TestTofu(t) - _, _, err := repo.ActivateSession(context.Background(), s.PublicId, s.Version, srv.PrivateId, srv.Type, tofu) + _, _, err := repo.ActivateSession(context.Background(), s.PublicId, s.Version, srv.PrivateId, tofu) require.NoError(t, err) return s } diff --git a/internal/session/service_close_connections_test.go b/internal/session/service_close_connections_test.go index 8eb5d3fa6..de0c0f007 100644 --- a/internal/session/service_close_connections_test.go +++ b/internal/session/service_close_connections_test.go @@ -32,7 +32,7 @@ func TestServiceCloseConnections(t *testing.T) { s := TestDefaultSession(t, conn, wrapper, iamRepo) srv := TestWorker(t, conn, wrapper) tofu := TestTofu(t) - s, _, err = repo.ActivateSession(context.Background(), s.PublicId, s.Version, srv.PrivateId, srv.Type, tofu) + s, _, err = repo.ActivateSession(context.Background(), s.PublicId, s.Version, srv.PrivateId, tofu) require.NoError(t, err) require.NoError(t, err) diff --git a/internal/session/service_worker_status_report_test.go b/internal/session/service_worker_status_report_test.go index 0589bdea3..64a29c252 100644 --- a/internal/session/service_worker_status_report_test.go +++ b/internal/session/service_worker_status_report_test.go @@ -4,6 +4,8 @@ import ( "context" "testing" + "github.com/hashicorp/boundary/internal/servers/store" + "github.com/hashicorp/boundary/internal/authtoken" "github.com/hashicorp/boundary/internal/db" "github.com/hashicorp/boundary/internal/host/static" @@ -26,14 +28,12 @@ func TestWorkerStatusReport(t *testing.T) { org, prj := iam.TestScopes(t, iam.TestRepo(t, conn, wrapper)) serverRepo, _ := servers.NewRepository(rw, rw, kms) - serverRepo.UpsertServer(ctx, &servers.Server{ + serverRepo.UpsertController(ctx, &store.Controller{ PrivateId: "test_controller1", - Type: "controller", Address: "127.0.0.1", }) - serverRepo.UpsertServer(ctx, &servers.Server{ + serverRepo.UpsertWorker(ctx, &store.Worker{ PrivateId: "test_worker1", - Type: "worker", Address: "127.0.0.1", }) @@ -56,7 +56,7 @@ func TestWorkerStatusReport(t *testing.T) { ) type testCase struct { - worker *servers.Server + worker *store.Worker req []session.StateReport want []session.StateReport orphanedConnections []string @@ -91,7 +91,7 @@ func TestWorkerStatusReport(t *testing.T) { ConnectionLimit: 10, }) tofu := session.TestTofu(t) - sess, _, err = repo.ActivateSession(ctx, sess.PublicId, sess.Version, worker.PrivateId, worker.Type, tofu) + sess, _, err = repo.ActivateSession(ctx, sess.PublicId, sess.Version, worker.PrivateId, tofu) require.NoError(t, err) require.NoError(t, err) @@ -123,7 +123,7 @@ func TestWorkerStatusReport(t *testing.T) { ConnectionLimit: 10, }) tofu := session.TestTofu(t) - sess, _, err = repo.ActivateSession(ctx, sess.PublicId, sess.Version, worker.PrivateId, worker.Type, tofu) + sess, _, err = repo.ActivateSession(ctx, sess.PublicId, sess.Version, worker.PrivateId, tofu) require.NoError(t, err) require.NoError(t, err) @@ -157,7 +157,7 @@ func TestWorkerStatusReport(t *testing.T) { ConnectionLimit: 10, }) tofu := session.TestTofu(t) - sess, _, err = repo.ActivateSession(ctx, sess.PublicId, sess.Version, worker.PrivateId, worker.Type, tofu) + sess, _, err = repo.ActivateSession(ctx, sess.PublicId, sess.Version, worker.PrivateId, tofu) require.NoError(t, err) connection, _, err := connRepo.AuthorizeConnection(ctx, sess.PublicId, worker.PrivateId) require.NoError(t, err) @@ -197,7 +197,7 @@ func TestWorkerStatusReport(t *testing.T) { ConnectionLimit: 10, }) tofu := session.TestTofu(t) - sess, _, err = repo.ActivateSession(ctx, sess.PublicId, sess.Version, worker.PrivateId, worker.Type, tofu) + sess, _, err = repo.ActivateSession(ctx, sess.PublicId, sess.Version, worker.PrivateId, tofu) require.NoError(t, err) connection, _, err := connRepo.AuthorizeConnection(ctx, sess.PublicId, worker.PrivateId) require.NoError(t, err) @@ -215,7 +215,7 @@ func TestWorkerStatusReport(t *testing.T) { ConnectionLimit: 10, }) tofu2 := session.TestTofu(t) - sess2, _, err = repo.ActivateSession(ctx, sess2.PublicId, sess2.Version, worker.PrivateId, worker.Type, tofu2) + sess2, _, err = repo.ActivateSession(ctx, sess2.PublicId, sess2.Version, worker.PrivateId, tofu2) require.NoError(t, err) connection2, _, err := connRepo.AuthorizeConnection(ctx, sess2.PublicId, worker.PrivateId) require.NoError(t, err) @@ -264,7 +264,7 @@ func TestWorkerStatusReport(t *testing.T) { ConnectionLimit: 10, }) tofu := session.TestTofu(t) - sess, _, err = repo.ActivateSession(ctx, sess.PublicId, sess.Version, worker.PrivateId, worker.Type, tofu) + sess, _, err = repo.ActivateSession(ctx, sess.PublicId, sess.Version, worker.PrivateId, tofu) require.NoError(t, err) connection, _, err := connRepo.AuthorizeConnection(ctx, sess.PublicId, worker.PrivateId) require.NoError(t, err) @@ -280,7 +280,7 @@ func TestWorkerStatusReport(t *testing.T) { ConnectionLimit: 10, }) tofu2 := session.TestTofu(t) - sess2, _, err = repo.ActivateSession(ctx, sess2.PublicId, sess2.Version, worker.PrivateId, worker.Type, tofu2) + sess2, _, err = repo.ActivateSession(ctx, sess2.PublicId, sess2.Version, worker.PrivateId, tofu2) require.NoError(t, err) connection2, _, err := connRepo.AuthorizeConnection(ctx, sess2.PublicId, worker.PrivateId) require.NoError(t, err) @@ -315,7 +315,7 @@ func TestWorkerStatusReport(t *testing.T) { ConnectionLimit: 10, }) tofu := session.TestTofu(t) - sess, _, err = repo.ActivateSession(ctx, sess.PublicId, sess.Version, worker.PrivateId, worker.Type, tofu) + sess, _, err = repo.ActivateSession(ctx, sess.PublicId, sess.Version, worker.PrivateId, tofu) require.NoError(t, err) connection, _, err := connRepo.AuthorizeConnection(ctx, sess.PublicId, worker.PrivateId) require.NoError(t, err) @@ -333,7 +333,7 @@ func TestWorkerStatusReport(t *testing.T) { ConnectionLimit: 10, }) tofu2 := session.TestTofu(t) - sess2, _, err = repo.ActivateSession(ctx, sess2.PublicId, sess2.Version, worker.PrivateId, worker.Type, tofu2) + sess2, _, err = repo.ActivateSession(ctx, sess2.PublicId, sess2.Version, worker.PrivateId, tofu2) require.NoError(t, err) connection2, _, err := connRepo.AuthorizeConnection(ctx, sess2.PublicId, worker.PrivateId) require.NoError(t, err) diff --git a/internal/session/session.go b/internal/session/session.go index 7026656e7..7818aacc8 100644 --- a/internal/session/session.go +++ b/internal/session/session.go @@ -64,10 +64,8 @@ type Session struct { UserId string `json:"user_id,omitempty" gorm:"default:null"` // HostId of the session HostId string `json:"host_id,omitempty" gorm:"default:null"` - // ServerId that proxied the session - ServerId string `json:"server_id,omitempty" gorm:"default:null"` - // ServerType that proxied the session - ServerType string `json:"server_type,omitempty" gorm:"default:null"` + // WorkerId that proxied the session + WorkerId string `json:"worker_id,omitempty" gorm:"default:null"` // TargetId for the session TargetId string `json:"target_id,omitempty" gorm:"default:null"` // HostSetId for the session @@ -171,8 +169,7 @@ func (s *Session) Clone() interface{} { PublicId: s.PublicId, UserId: s.UserId, HostId: s.HostId, - ServerId: s.ServerId, - ServerType: s.ServerType, + WorkerId: s.WorkerId, TargetId: s.TargetId, HostSetId: s.HostSetId, AuthTokenId: s.AuthTokenId, @@ -337,11 +334,8 @@ func (s *Session) validateNewSession() error { if s.TerminationReason != "" { return errors.NewDeprecated(errors.InvalidParameter, op, "termination reason must be empty") } - if s.ServerId != "" { - return errors.NewDeprecated(errors.InvalidParameter, op, "server id must be empty") - } - if s.ServerType != "" { - return errors.NewDeprecated(errors.InvalidParameter, op, "server type must be empty") + if s.WorkerId != "" { + return errors.NewDeprecated(errors.InvalidParameter, op, "worker id must be empty") } if s.TofuToken != nil { return errors.NewDeprecated(errors.InvalidParameter, op, "tofu token must be empty") @@ -449,8 +443,7 @@ type sessionListView struct { PublicId string `json:"public_id,omitempty" gorm:"primary_key"` UserId string `json:"user_id,omitempty" gorm:"default:null"` HostId string `json:"host_id,omitempty" gorm:"default:null"` - ServerId string `json:"server_id,omitempty" gorm:"default:null"` - ServerType string `json:"server_type,omitempty" gorm:"default:null"` + WorkerId string `json:"worker_id,omitempty" gorm:"default:null"` TargetId string `json:"target_id,omitempty" gorm:"default:null"` HostSetId string `json:"host_set_id,omitempty" gorm:"default:null"` AuthTokenId string `json:"auth_token_id,omitempty" gorm:"default:null"` diff --git a/internal/session/session_cleanup_job.go b/internal/session/session_cleanup_job.go index c684ed1af..9ab10515a 100644 --- a/internal/session/session_cleanup_job.go +++ b/internal/session/session_cleanup_job.go @@ -107,7 +107,7 @@ func (j *sessionConnectionCleanupJob) Run(ctx context.Context) error { for _, result := range results { event.WriteError(ctx, op, stderrors.New("worker has not reported status within acceptable grace period, all connections closed"), event.WithInfo( - "private_id", result.ServerId, + "private_id", result.WorkerId, "update_time", result.LastUpdateTime, "grace_period_seconds", j.gracePeriod.Seconds(), "number_connections_closed", result.NumberConnectionsClosed, diff --git a/internal/session/testing.go b/internal/session/testing.go index 311a50b53..ced1f58c5 100644 --- a/internal/session/testing.go +++ b/internal/session/testing.go @@ -6,6 +6,8 @@ import ( "testing" "time" + "github.com/hashicorp/boundary/internal/servers/store" + "github.com/hashicorp/boundary/internal/auth/password" "github.com/hashicorp/boundary/internal/authtoken" "github.com/hashicorp/boundary/internal/db" @@ -88,7 +90,7 @@ func TestSession(t testing.TB, conn *db.DB, wrapper wrapping.Wrapper, c Composed _, certBytes, err := newCert(ctx, wrapper, c.UserId, id, []string{"127.0.0.1", "localhost"}, c.ExpirationTime.Timestamp.AsTime()) require.NoError(err) s.Certificate = certBytes - s.ServerId = opts.withServerId + s.WorkerId = opts.withWorkerId if len(s.TofuToken) != 0 { err = s.encrypt(ctx, wrapper) @@ -181,8 +183,7 @@ func TestTofu(t testing.TB) []byte { } // TestWorker inserts a worker into the db to satisfy foreign key constraints. -// Supports the WithServerId option. -func TestWorker(t testing.TB, conn *db.DB, wrapper wrapping.Wrapper, opt ...Option) *servers.Server { +func TestWorker(t *testing.T, conn *db.DB, wrapper wrapping.Wrapper, opt ...Option) *store.Worker { t.Helper() rw := db.New(conn) kms := kms.TestKms(t, conn, wrapper) @@ -190,19 +191,19 @@ func TestWorker(t testing.TB, conn *db.DB, wrapper wrapping.Wrapper, opt ...Opti require.NoError(t, err) opts := getOpts(opt...) - id := opts.withServerId + id := opts.withWorkerId if id == "" { id, err = uuid.GenerateUUID() require.NoError(t, err) id = "test-session-worker-" + id } - worker := &servers.Server{ - PrivateId: id, - Type: servers.ServerTypeWorker.String(), - Description: "Test Session Worker", - Address: "127.0.0.1", + name := "test-worker-" + id + worker := &store.Worker{ + PrivateId: id, + Name: name, + Address: "127.0.0.1", } - _, _, err = serversRepo.UpsertServer(context.Background(), worker) + _, _, 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 9178c559b..2455d2189 100644 --- a/testing/dbtest/session_list_benchmarks_dump_generation_test.go +++ b/testing/dbtest/session_list_benchmarks_dump_generation_test.go @@ -19,6 +19,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" @@ -131,12 +132,11 @@ func TestGenerateSessionBenchmarkTemplateDumps(t *testing.T) { require.NoError(err) serversRepo, err := servers.NewRepository(rw, rw, kms) require.NoError(err) - worker := &servers.Server{ + worker := &store.Worker{ PrivateId: "test1", - Type: "worker", Address: "127.0.0.1", } - _, _, err = serversRepo.UpsertServer(ctx, worker) + _, _, err = serversRepo.UpsertWorker(ctx, worker) require.NoError(err) usersStart := time.Now() @@ -220,8 +220,8 @@ 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 *servers.Server, numConns int) { - sess, _, err := sessRepo.ActivateSession(ctx, sess.PublicId, sess.Version, worker.PrivateId, worker.Type, []byte(`tofu`)) +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) { + sess, _, err := sessRepo.ActivateSession(ctx, sess.PublicId, sess.Version, worker.PrivateId, []byte(`tofu`)) require.NoError(t, err) var closeWiths []session.CloseWith for i := 0; i < numConns; i++ {