CyberCoveLLC/threshold
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refact(servers): Split server table into worker and controller

Browse Source
This commit is contained in:
irenarindos authored and Jeff Mitchell committed 2022-05-09 13:30:01 -04:00
1 parent f748af7b02
commit 9e27605213
39 files changed
+1427 -460
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

Makefile
+3 -1
View File
@@ -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
internal/cmd/commands/server/worker_tags_reload_test.go
+1 -1
View File
@@ -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)
internal/db/schema/migrations/oss/postgres/99/01_worker_controller_tables.up.sql
+130
View File
@@ -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;
internal/db/schema/migrations/oss/postgres_24_01_test.go
-166
View File
@@ -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)
}
internal/gen/controller/servers/services/server_coordination_service.pb.go
+2 -3
View File
@@ -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
internal/proto/controller/servers/services/v1/server_coordination_service.proto
+4 -3
View File
@@ -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
internal/proto/controller/storage/servers/store/v1/controller.proto
+25
View File
@@ -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;
}
internal/proto/controller/storage/servers/store/v1/worker.proto
+39
View File
@@ -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<string, TagValues> tags = 70;
}
// TagValues is used because map fields cannot be repeated but can be a
// message
message TagValues {
repeated string values = 10;
}
internal/scheduler/job/testing.go
+7 -7
View File
@@ -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
}
internal/scheduler/testing.go
+6 -6
View File
@@ -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) {
internal/servers/controller/controller.go
+3 -3
View File
@@ -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)
internal/servers/controller/handlers/targets/target_service.go
+4 -5
View File
@@ -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
internal/servers/controller/handlers/workers/worker_service.go
+33 -6
View File
@@ -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)
internal/servers/controller/handlers/workers/worker_service_status_test.go
+37 -25
View File
@@ -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",
},
},
internal/servers/controller/tickers.go
+9 -12
View File
@@ -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"))
}
internal/servers/gorm.go
+3 -2
View File
@@ -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"
}
internal/servers/query.go
+1 -1
View File
@@ -2,5 +2,5 @@ package servers
const (
deleteWhereCreateTimeSql = `create_time < ?`
deleteTagsSql = `server_id = ?`
deleteTagsSql = `worker_id = ?`
)
internal/servers/repository.go
+111 -59
View File
@@ -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
internal/servers/repository_test.go
+37 -39
View File
@@ -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)
internal/servers/store/controller.pb.go
+209
View File
@@ -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
}
internal/servers/store/gorm.go
+11
View File
@@ -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"
}
internal/servers/store/worker.pb.go
+308
View File
@@ -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
}
internal/servers/worker/status.go
+6 -7
View File
@@ -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(),
})
internal/servers/worker/worker.go
+2 -1
View File
@@ -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"
internal/session/options.go
+15 -15
View File
@@ -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
}
}
internal/session/options_test.go
+3 -3
View File
@@ -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)
})
}
internal/session/query.go
+14 -14
View File
@@ -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
internal/session/repository.go
+1 -2
View File
@@ -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,
internal/session/repository_connection.go
+54 -1
View File
@@ -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
internal/session/repository_connection_test.go
+290 -6
View File
@@ -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++ {
internal/session/repository_session.go
+10 -17
View File
@@ -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)
}
internal/session/repository_session_test.go
+10 -10
View File
@@ -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)
})
}
internal/session/service_authorize_connection_test.go
+1 -1
View File
@@ -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
}
internal/session/service_close_connections_test.go
+1 -1
View File
@@ -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)
internal/session/service_worker_status_report_test.go
+14 -14
View File
@@ -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)
internal/session/session.go
+6 -13
View File
@@ -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"`
internal/session/session_cleanup_job.go
+1 -1
View File
@@ -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,
internal/session/testing.go
+11 -10
View File
@@ -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
}
testing/dbtest/session_list_benchmarks_dump_generation_test.go
+5 -5
View File
@@ -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++ {