feat(worker): Worker Local Storage State DB Column

- Add new `local_storage_state` column to `server_worker` table which keeps track of the state of the local disk space.

- The valid values for `local_storage_state` are:
-- `available`: The worker local storage state is at an acceptable state
-- `lowStorage`: The worker is below the minimum threshold for local storage
-- `criticallyLowStorage`: The worker local storage state is below the critical minimum threshold for local storage
-- `outOfStorage`: The worker is out of local disk space
-- `notConfigured`: The worker does not have a local storage path configured
-- `unknown`: The default local storage state of a worker. Used when the local storage state of a worker is not yet known
This commit is contained in:
Elim Tsiagbey committed 2024-04-05 16:44:54 -04:00
1 parent 6cb7bebaa7
commit 79917f74b8
13 files changed
+652 -169

No files matched your search

@@ -31,6 +31,7 @@ alter table server_worker
drop view server_worker_aggregate;
-- Updates view created in 51/01_server_worker_release_version.up.sql to add the worker operational state
-- Replaced in 86/01_server_worker_local_storage.up.sql
create view server_worker_aggregate as
with worker_config_tags(worker_id, source, tags) as (
select
@@ -0,0 +1,85 @@
-- Copyright (c) HashiCorp, Inc.
-- SPDX-License-Identifier: BUSL-1.1
begin;
create table server_worker_local_storage_state_state_enm (
state text primary key
constraint only_predefined_local_storage_states_allowed
check (
state in (
'available',
'low storage',
'critically low storage',
'out of storage',
'not configured',
'unknown'
)
)
);
comment on table server_worker_local_storage_state_state_enm is
'server_worker_local_storage_state_state_enm is an enumeration table for worker local storage states.';
insert into server_worker_local_storage_state_state_enm (state) values
('available'),
('low storage'),
('critically low storage'),
('out of storage'),
('not configured'),
('unknown');
alter table server_worker
add column local_storage_state text not null default 'unknown'
constraint server_worker_local_storage_state_state_enm_fkey
references server_worker_local_storage_state_state_enm (state)
on delete restrict
on update cascade;
drop view server_worker_aggregate;
-- Replaces view created in 52/01_worker_operational_state.up.sql to add the worker local storage state
create view server_worker_aggregate as
with worker_config_tags(worker_id, source, tags) as (
select
ct.worker_id,
ct.source,
-- keys and tags can be any lowercase printable character so use uppercase characters as delimitors.
string_agg(distinct concat_ws('Y', ct.key, ct.value), 'Z') as tags
from server_worker_tag ct
group by ct.worker_id, ct.source
),
connection_count (worker_id, count) as (
select
worker_id,
count(1) as count
from session_connection
where closed_reason is null
group by worker_id
)
select
w.public_id,
w.scope_id,
w.description,
w.name,
w.address,
w.create_time,
w.update_time,
w.version,
w.last_status_time,
w.type,
w.release_version,
w.operational_state,
w.local_storage_state,
cc.count as active_connection_count,
wt.tags as api_tags,
ct.tags as worker_config_tags
from server_worker w
left join worker_config_tags wt on
w.public_id = wt.worker_id and wt.source = 'api'
left join worker_config_tags ct on
w.public_id = ct.worker_id and ct.source = 'configuration'
left join connection_count as cc on
w.public_id = cc.worker_id;
comment on view server_worker_aggregate is
'server_worker_aggregate contains the worker resource with its worker provided config values and its configuration and api provided tags.';
commit;
@@ -0,0 +1,104 @@
// Copyright (c) HashiCorp, Inc.
// SPDX-License-Identifier: BUSL-1.1
package oss_test
import (
"context"
"testing"
"github.com/hashicorp/boundary/internal/db/common"
"github.com/hashicorp/boundary/internal/db/schema"
"github.com/hashicorp/boundary/testing/dbtest"
"github.com/stretchr/testify/require"
)
const (
insertWorkerQuery79 = "insert into server_worker (public_id, scope_id, type) values ($1, 'global', 'pki')"
selectWorkerQuery79 = "select public_id, local_storage_state from server_worker where public_id = $1"
priorMigration = 85001
currentMigration = 86001
)
type testWorker79 struct {
LocalStorageState string
PublicId string
}
func Test_WorkerLocalStorageStateChanges(t *testing.T) {
t.Parallel()
require := require.New(t)
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)
_, err = m.ApplyMigrations(ctx)
require.NoError(err)
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 data
execResult, err := d.ExecContext(ctx, insertWorkerQuery79, "test-worker-79")
require.NoError(err)
rowsAffected, err := execResult.RowsAffected()
require.NoError(err)
require.Equal(int64(1), rowsAffected)
// 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)
_, err = m.ApplyMigrations(ctx)
require.NoError(err)
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)
// Assert worker has been migrated and received default state of 'unknown'
actualWorker := new(testWorker79)
row := d.QueryRowContext(ctx, selectWorkerQuery79, "test-worker-79")
require.NoError(row.Scan(
&actualWorker.PublicId,
&actualWorker.LocalStorageState,
))
require.Equal("unknown", actualWorker.LocalStorageState)
}
+2 -1
View File
@@ -36,7 +36,8 @@ TESTS ?= tests/setup/*.sql \
tests/auth/*/*.sql \
tests/purge/*.sql \
tests/pagination/*.sql \
tests/policy/*.sql
tests/policy/*.sql \
tests/worker/*.sql
POSTGRES_DOCKER_IMAGE_BASE ?= postgres
@@ -0,0 +1,52 @@
-- Copyright (c) HashiCorp, Inc.
-- SPDX-License-Identifier: BUSL-1.1
begin;
select plan(1);
-- Insert valid local_storage_state values
insert into server_worker
(public_id, scope_id, type, local_storage_state)
values
('w_1234567891', 'global', 'pki', 'available');
insert into server_worker
(public_id, scope_id, type, local_storage_state)
values
('w_1234567892', 'global', 'pki', 'low storage');
insert into server_worker
(public_id, scope_id, type, local_storage_state)
values
('w_1234567893', 'global', 'pki', 'critically low storage');
insert into server_worker
(public_id, scope_id, type, local_storage_state)
values
('w_1234567894', 'global', 'pki', 'out of storage');
insert into server_worker
(public_id, scope_id, type, local_storage_state)
values
('w_1234567895', 'global', 'pki', 'not configured');
insert into server_worker
(public_id, scope_id, type, local_storage_state)
values
('w_1234567896', 'global', 'pki', 'unknown');
-- Try to insert invalid local_storage_state value to test constraint
prepare invalid_local_storage_state_value as
insert into server_worker
(public_id, scope_id, type, local_storage_state)
values
('w_1234567897', 'global', 'pki', 'invalid state');
select throws_ok(
'invalid_local_storage_state_value',
'23503',
'insert or update on table "server_worker" violates foreign key constraint "server_worker_local_storage_state_state_enm_fkey"',
'inserting a row with invalid local_storage_state value'
);
select * from finish();
rollback;
@@ -65,6 +65,17 @@ message Worker {
// The state of the worker, to indicate if the worker is active or in shutdown.
// @inject_tag: `gorm:"not_null"`
string operational_state = 150;
// The local_storage_state indicates the state of the local disk space of the worker.
// Possible values are:
// - available: The worker local storage state is at an acceptable state
// - low storage: The worker is below the minimum threshold for local storage
// - critically low storage: The worker local storage state is below the critical minimum threshold for local storage
// - out of storage: The worker is out of local disk space
// - not configured: The worker does not have a local storage path configured
// - unknown: The default local storage state of a worker. Used when the local storage state of a worker is not yet known
// @inject_tag: `gorm:"not_null"`
string local_storage_state = 160;
}
// WorkerTag is a tag for a worker. The primary key is comprised of the
+11 -2
View File
@@ -49,6 +49,7 @@ type options struct {
WithCreateControllerLedActivationToken bool
withReleaseVersion string
withOperationalState string
withLocalStorageState string
withActiveWorkers bool
withFeature version.Feature
withDirectlyConnected bool
@@ -57,8 +58,9 @@ type options struct {
func getDefaultOptions() options {
return options{
withNewIdFunc: newWorkerId,
withOperationalState: ActiveOperationalState.String(),
withNewIdFunc: newWorkerId,
withOperationalState: ActiveOperationalState.String(),
withLocalStorageState: UnknownWorkerType.String(),
}
}
@@ -267,3 +269,10 @@ func WithWorkerPool(workerIds []string) Option {
o.withWorkerPool = workerIds
}
}
// WithLocalStorageState provides an optional local storage state.
func WithLocalStorageState(state string) Option {
return func(o *options) {
o.withLocalStorageState = state
}
}
+8
View File
@@ -243,4 +243,12 @@ func Test_GetOpts(t *testing.T) {
testOpts.withNewIdFunc = nil
assert.Equal(t, opts, testOpts)
})
t.Run("WithLocalStorageState", func(t *testing.T) {
opts := GetOpts(WithLocalStorageState(AvailableLocalStorageState.String()))
testOpts := getDefaultOptions()
testOpts.withLocalStorageState = AvailableLocalStorageState.String()
opts.withNewIdFunc = nil
testOpts.withNewIdFunc = nil
assert.Equal(t, opts, testOpts)
})
}
+13 -2
View File
@@ -570,8 +570,10 @@ func (r *Repository) UpdateWorker(ctx context.Context, worker *Worker, version u
// ReportedStatus and Tags are intentionally ignored when creating a worker (not
// included). Currently, a worker can only be created in the global scope
//
// Options supported: WithNewIdFunc (this option is likely only useful for
// tests), WithFetchNodeCredentialsRequest,
// Options supported:
// WithNewIdFunc (this option is likely only useful for tests),
// withLocalStorageState (this option is likely only useful for tests),
// WithFetchNodeCredentialsRequest,
// WithCreateControllerLedActivationToken. The latter two are mutually
// exclusive.
func (r *Repository) CreateWorker(ctx context.Context, worker *Worker, opt ...Option) (*Worker, error) {
@@ -597,6 +599,15 @@ func (r *Repository) CreateWorker(ctx context.Context, worker *Worker, opt ...Op
}
worker.OperationalState = UnknownOperationalState.String()
worker.LocalStorageState = UnknownLocalStorageState.String()
if opts.withLocalStorageState != "" {
validLocalStorageState := ValidLocalStorageState(opts.withLocalStorageState)
if !validLocalStorageState {
return nil, errors.New(ctx, errors.InvalidParameter, op, "invalid local storage state")
}
worker.LocalStorageState = opts.withLocalStorageState
}
var err error
if worker.PublicId, err = opts.withNewIdFunc(ctx); err != nil {
+23
View File
@@ -1163,6 +1163,29 @@ func TestRepository_CreateWorker(t *testing.T) {
opt: []server.Option{server.WithCreateControllerLedActivationToken(true)},
repo: testRepo,
},
{
name: "success-with-valid-local-storage-state",
setup: func() *server.Worker {
w := server.NewWorker(scope.Global.String())
return w
},
reader: rw,
opt: []server.Option{server.WithLocalStorageState("available")},
repo: testRepo,
},
{
name: "failure-with-invalid-local-storage-state",
setup: func() *server.Worker {
w := server.NewWorker(scope.Global.String())
return w
},
reader: rw,
opt: []server.Option{server.WithLocalStorageState("invalid")},
repo: testRepo,
wantErr: true,
wantErrIs: errors.InvalidParameter,
wantErrContains: "invalid local storage state",
},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
+32 -12
View File
@@ -67,6 +67,16 @@ type Worker struct {
// The state of the worker, to indicate if the worker is active or in shutdown.
// @inject_tag: `gorm:"not_null"`
OperationalState string `protobuf:"bytes,150,opt,name=operational_state,json=operationalState,proto3" json:"operational_state,omitempty" gorm:"not_null"`
// The local_storage_state indicates the state of the local disk space of the worker.
// Possible values are:
// - available: The worker local storage state is at an acceptable state
// - low storage: The worker is below the minimum threshold for local storage
// - critically low storage: The worker local storage state is below the critical minimum threshold for local storage
// - out of storage: The worker is out of local disk space
// - not configured: The worker does not have a local storage path configured
// - unknown: The default local storage state of a worker. Used when the local storage state of a worker is not yet known
// @inject_tag: `gorm:"not_null"`
LocalStorageState string `protobuf:"bytes,160,opt,name=local_storage_state,json=localStorageState,proto3" json:"local_storage_state,omitempty" gorm:"not_null"`
}
func (x *Worker) Reset() {
@@ -185,6 +195,13 @@ func (x *Worker) GetOperationalState() string {
return ""
}
func (x *Worker) GetLocalStorageState() string {
if x != nil {
return x.LocalStorageState
}
return ""
}
// WorkerTag is a tag for a worker. The primary key is comprised of the
// worker_id, key, value, and source.
type WorkerTag struct {
@@ -280,7 +297,7 @@ var file_controller_storage_servers_store_v1_worker_proto_rawDesc = []byte{
0x6f, 0x74, 0x6f, 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, 0xb9, 0x04, 0x0a, 0x06, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x12,
0x72, 0x6f, 0x74, 0x6f, 0x22, 0xea, 0x04, 0x0a, 0x06, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x12,
0x1b, 0x0a, 0x09, 0x70, 0x75, 0x62, 0x6c, 0x69, 0x63, 0x5f, 0x69, 0x64, 0x18, 0x0a, 0x20, 0x01,
0x28, 0x09, 0x52, 0x08, 0x70, 0x75, 0x62, 0x6c, 0x69, 0x63, 0x49, 0x64, 0x12, 0x24, 0x0a, 0x04,
0x6e, 0x61, 0x6d, 0x65, 0x18, 0x14, 0x20, 0x01, 0x28, 0x09, 0x42, 0x10, 0xc2, 0xdd, 0x29, 0x0c,
@@ -316,17 +333,20 @@ var file_controller_storage_servers_store_v1_worker_proto_rawDesc = []byte{
0x6f, 0x6e, 0x12, 0x2c, 0x0a, 0x11, 0x6f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x61,
0x6c, 0x5f, 0x73, 0x74, 0x61, 0x74, 0x65, 0x18, 0x96, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x10,
0x6f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x61, 0x6c, 0x53, 0x74, 0x61, 0x74, 0x65,
0x22, 0x68, 0x0a, 0x09, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x54, 0x61, 0x67, 0x12, 0x1b, 0x0a,
0x09, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x5f, 0x69, 0x64, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x09,
0x52, 0x08, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x49, 0x64, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65,
0x79, 0x18, 0x14, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x14, 0x0a, 0x05,
0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x1e, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x76, 0x61, 0x6c,
0x75, 0x65, 0x12, 0x16, 0x0a, 0x06, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x18, 0x28, 0x20, 0x01,
0x28, 0x09, 0x52, 0x06, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x42, 0x3b, 0x5a, 0x39, 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, 0x2f, 0x73, 0x74, 0x6f, 0x72,
0x65, 0x3b, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
0x12, 0x2f, 0x0a, 0x13, 0x6c, 0x6f, 0x63, 0x61, 0x6c, 0x5f, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67,
0x65, 0x5f, 0x73, 0x74, 0x61, 0x74, 0x65, 0x18, 0xa0, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x11,
0x6c, 0x6f, 0x63, 0x61, 0x6c, 0x53, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x53, 0x74, 0x61, 0x74,
0x65, 0x22, 0x68, 0x0a, 0x09, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x54, 0x61, 0x67, 0x12, 0x1b,
0x0a, 0x09, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x5f, 0x69, 0x64, 0x18, 0x0a, 0x20, 0x01, 0x28,
0x09, 0x52, 0x08, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x49, 0x64, 0x12, 0x10, 0x0a, 0x03, 0x6b,
0x65, 0x79, 0x18, 0x14, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x14, 0x0a,
0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x1e, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x76, 0x61,
0x6c, 0x75, 0x65, 0x12, 0x16, 0x0a, 0x06, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x18, 0x28, 0x20,
0x01, 0x28, 0x09, 0x52, 0x06, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x42, 0x3b, 0x5a, 0x39, 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, 0x2f, 0x73, 0x74, 0x6f,
0x72, 0x65, 0x3b, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
}
var (
+55 -26
View File
@@ -16,17 +16,24 @@ import (
)
type (
WorkerType string
OperationalState string
WorkerType string
OperationalState string
LocalStorageState string
)
const (
UnknownWorkerType WorkerType = "unknown"
KmsWorkerType WorkerType = "kms"
PkiWorkerType WorkerType = "pki"
ActiveOperationalState OperationalState = "active"
ShutdownOperationalState OperationalState = "shutdown"
UnknownOperationalState OperationalState = "unknown"
UnknownWorkerType WorkerType = "unknown"
KmsWorkerType WorkerType = "kms"
PkiWorkerType WorkerType = "pki"
ActiveOperationalState OperationalState = "active"
ShutdownOperationalState OperationalState = "shutdown"
UnknownOperationalState OperationalState = "unknown"
AvailableLocalStorageState LocalStorageState = "available"
LowStorageLocalStorageState LocalStorageState = "low storage"
CriticallyLowStorageLocalStorageState LocalStorageState = "critically low storage"
OutOfStorageLocalStorageState LocalStorageState = "out of storage"
NotConfiguredLocalStorageState LocalStorageState = "not configured"
UnknownLocalStorageState LocalStorageState = "unknown"
)
func (t WorkerType) Valid() bool {
@@ -65,6 +72,25 @@ func (t OperationalState) String() string {
return string(UnknownOperationalState)
}
func ValidLocalStorageState(s string) bool {
switch s {
case AvailableLocalStorageState.String(), LowStorageLocalStorageState.String(),
CriticallyLowStorageLocalStorageState.String(), OutOfStorageLocalStorageState.String(),
NotConfiguredLocalStorageState.String(), UnknownLocalStorageState.String():
return true
}
return false
}
func (t LocalStorageState) String() string {
switch t {
case AvailableLocalStorageState, LowStorageLocalStorageState,
OutOfStorageLocalStorageState, NotConfiguredLocalStorageState:
return string(t)
}
return string(UnknownLocalStorageState)
}
// AttachWorkerIdToState accepts a workerId and creates a struct for use with the Nodeenrollment lib
// This is intended for use in worker authorization; AuthorizeNode in the lib accepts the option WithState
// so that the workerId is passed through to storage and associated with a WorkerAuth record
@@ -108,12 +134,13 @@ func NewWorker(scopeId string, opt ...Option) *Worker {
opts := GetOpts(opt...)
worker := &Worker{
Worker: &store.Worker{
ScopeId: scopeId,
Name: opts.withName,
Description: opts.withDescription,
Address: opts.withAddress,
ReleaseVersion: opts.withReleaseVersion,
OperationalState: opts.withOperationalState,
ScopeId: scopeId,
Name: opts.withName,
Description: opts.withDescription,
Address: opts.withAddress,
ReleaseVersion: opts.withReleaseVersion,
OperationalState: opts.withOperationalState,
LocalStorageState: opts.withLocalStorageState,
},
inputTags: opts.withWorkerTags,
}
@@ -231,6 +258,7 @@ type workerAggregate struct {
ApiTags string
ActiveConnectionCount uint32
OperationalState string
LocalStorageState string
// Config Fields
LastStatusTime *timestamp.Timestamp
WorkerConfigTags string
@@ -240,18 +268,19 @@ func (a *workerAggregate) toWorker(ctx context.Context) (*Worker, error) {
const op = "server.(workerAggregate).toWorker"
worker := &Worker{
Worker: &store.Worker{
PublicId: a.PublicId,
Name: a.Name,
Description: a.Description,
Address: a.Address,
CreateTime: a.CreateTime,
UpdateTime: a.UpdateTime,
ScopeId: a.ScopeId,
Version: a.Version,
LastStatusTime: a.LastStatusTime,
Type: a.Type,
ReleaseVersion: a.ReleaseVersion,
OperationalState: a.OperationalState,
PublicId: a.PublicId,
Name: a.Name,
Description: a.Description,
Address: a.Address,
CreateTime: a.CreateTime,
UpdateTime: a.UpdateTime,
ScopeId: a.ScopeId,
Version: a.Version,
LastStatusTime: a.LastStatusTime,
Type: a.Type,
ReleaseVersion: a.ReleaseVersion,
OperationalState: a.OperationalState,
LocalStorageState: a.LocalStorageState,
},
activeConnectionCount: a.ActiveConnectionCount,
}
+255 -126
View File
@@ -227,6 +227,37 @@ func TestWorkerAggregate(t *testing.T) {
{Key: "key", Value: "val"},
})
})
t.Run("worker with default unknown local storage state", func(t *testing.T) {
id, err := newWorkerId(ctx)
require.NoError(t, err)
id = strings.ToLower(id)
w := NewWorker(scope.Global.String(),
WithName(id),
WithAddress("address"))
w.Type = KmsWorkerType.String()
w.PublicId = id
require.NoError(t, rw.Create(ctx, w))
got := getAggWorker(id)
assert.Equal(t, UnknownLocalStorageState.String(), got.LocalStorageState)
})
t.Run("worker with available local storage state", func(t *testing.T) {
id, err := newWorkerId(ctx)
require.NoError(t, err)
id = strings.ToLower(id)
w := NewWorker(scope.Global.String(),
WithName(id),
WithAddress("address"),
WithLocalStorageState("available"))
w.Type = KmsWorkerType.String()
w.PublicId = id
require.NoError(t, rw.Create(ctx, w))
got := getAggWorker(id)
assert.Equal(t, AvailableLocalStorageState.String(), got.LocalStorageState)
})
}
func TestWorker_Update(t *testing.T) {
@@ -254,12 +285,13 @@ func TestWorker_Update(t *testing.T) {
name: "kms update address",
initial: Worker{
Worker: &store.Worker{
Type: KmsWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Name: "kms update address",
Address: "kms update address",
OperationalState: ActiveOperationalState.String(),
Type: KmsWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Name: "kms update address",
Address: "kms update address",
OperationalState: ActiveOperationalState.String(),
LocalStorageState: AvailableLocalStorageState.String(),
},
},
update: Worker{
@@ -280,12 +312,13 @@ func TestWorker_Update(t *testing.T) {
name: "kms update name",
initial: Worker{
Worker: &store.Worker{
Type: KmsWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Name: "kms update name",
Address: "kms update name",
OperationalState: ActiveOperationalState.String(),
Type: KmsWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Name: "kms update name",
Address: "kms update name",
OperationalState: ActiveOperationalState.String(),
LocalStorageState: AvailableLocalStorageState.String(),
},
},
update: Worker{
@@ -300,12 +333,13 @@ func TestWorker_Update(t *testing.T) {
name: "kms clear name",
initial: Worker{
Worker: &store.Worker{
Type: KmsWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Name: "kms clear name",
Address: "kms clear name",
OperationalState: ActiveOperationalState.String(),
Type: KmsWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Name: "kms clear name",
Address: "kms clear name",
OperationalState: ActiveOperationalState.String(),
LocalStorageState: AvailableLocalStorageState.String(),
},
},
update: Worker{
@@ -318,11 +352,12 @@ func TestWorker_Update(t *testing.T) {
name: "pki update address",
initial: Worker{
Worker: &store.Worker{
Type: PkiWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Name: "pki update address",
OperationalState: ActiveOperationalState.String(),
Type: PkiWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Name: "pki update address",
OperationalState: ActiveOperationalState.String(),
LocalStorageState: AvailableLocalStorageState.String(),
},
},
update: Worker{
@@ -343,11 +378,12 @@ func TestWorker_Update(t *testing.T) {
name: "pki update name",
initial: Worker{
Worker: &store.Worker{
Type: PkiWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Name: "pki update name",
OperationalState: ActiveOperationalState.String(),
Type: PkiWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Name: "pki update name",
OperationalState: ActiveOperationalState.String(),
LocalStorageState: AvailableLocalStorageState.String(),
},
},
update: Worker{
@@ -367,11 +403,12 @@ func TestWorker_Update(t *testing.T) {
name: "pki update description",
initial: Worker{
Worker: &store.Worker{
Type: PkiWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Name: "pki update description",
OperationalState: ActiveOperationalState.String(),
Type: PkiWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Name: "pki update description",
OperationalState: ActiveOperationalState.String(),
LocalStorageState: AvailableLocalStorageState.String(),
},
},
update: Worker{
@@ -392,13 +429,14 @@ func TestWorker_Update(t *testing.T) {
name: "pki clear address",
initial: Worker{
Worker: &store.Worker{
Type: PkiWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Name: "pki clear address",
Address: "pki clear address",
LastStatusTime: &timestamp.Timestamp{Timestamp: timestamppb.New(time.Now().Add(time.Hour))},
OperationalState: ActiveOperationalState.String(),
Type: PkiWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Name: "pki clear address",
Address: "pki clear address",
LastStatusTime: &timestamp.Timestamp{Timestamp: timestamppb.New(time.Now().Add(time.Hour))},
OperationalState: ActiveOperationalState.String(),
LocalStorageState: UnknownLocalStorageState.String(),
},
},
update: Worker{
@@ -411,13 +449,14 @@ func TestWorker_Update(t *testing.T) {
name: "pki clear name",
initial: Worker{
Worker: &store.Worker{
Type: PkiWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Name: "pki clear name",
Address: "pki clear name",
LastStatusTime: &timestamp.Timestamp{Timestamp: timestamppb.New(time.Now().Add(time.Hour))},
OperationalState: ActiveOperationalState.String(),
Type: PkiWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Name: "pki clear name",
Address: "pki clear name",
LastStatusTime: &timestamp.Timestamp{Timestamp: timestamppb.New(time.Now().Add(time.Hour))},
OperationalState: ActiveOperationalState.String(),
LocalStorageState: AvailableLocalStorageState.String(),
},
},
update: Worker{
@@ -436,11 +475,12 @@ func TestWorker_Update(t *testing.T) {
name: "pki update name to capital",
initial: Worker{
Worker: &store.Worker{
Type: PkiWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Name: "pki update name to capital",
OperationalState: ActiveOperationalState.String(),
Type: PkiWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Name: "pki update name to capital",
OperationalState: ActiveOperationalState.String(),
LocalStorageState: AvailableLocalStorageState.String(),
},
},
update: Worker{
@@ -455,11 +495,12 @@ func TestWorker_Update(t *testing.T) {
name: "pki update name to unprintable",
initial: Worker{
Worker: &store.Worker{
Type: PkiWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Name: "worker reported update name to unprintable",
OperationalState: ActiveOperationalState.String(),
Type: PkiWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Name: "worker reported update name to unprintable",
OperationalState: ActiveOperationalState.String(),
LocalStorageState: AvailableLocalStorageState.String(),
},
},
update: Worker{
@@ -470,11 +511,54 @@ func TestWorker_Update(t *testing.T) {
mask: []string{"Name"},
wantUpdateError: true,
},
{
name: "local storage state update to available",
initial: Worker{
Worker: &store.Worker{
Type: PkiWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Name: "valid local storage state",
OperationalState: ActiveOperationalState.String(),
LocalStorageState: UnknownLocalStorageState.String(),
},
},
update: Worker{
Worker: &store.Worker{
LocalStorageState: AvailableLocalStorageState.String(),
},
},
mask: []string{"LocalStorageState"},
assert: func(t *testing.T, init, up *Worker) {
t.Helper()
assert.Equal(t, AvailableLocalStorageState.String(), up.LocalStorageState)
},
},
{
name: "local storage state update to invalid value",
initial: Worker{
Worker: &store.Worker{
Type: PkiWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Name: "invalid local storage state",
OperationalState: ActiveOperationalState.String(),
LocalStorageState: UnknownLocalStorageState.String(),
},
},
update: Worker{
Worker: &store.Worker{
LocalStorageState: "Invalid Local Storage State",
},
},
mask: []string{"LocalStorageState"},
wantUpdateError: true,
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
init := tc.initial.clone()
require.NoError(t, rw.Create(ctx, init))
require.NoError(t, rw.Create(ctx, init), tc.name)
up := tc.update.clone()
up.PublicId = init.PublicId
_, err := rw.Update(ctx, up, tc.mask, tc.nullMask)
@@ -516,13 +600,14 @@ func TestWorker_Create(t *testing.T) {
name: "kms base",
in: Worker{
Worker: &store.Worker{
Type: KmsWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Address: "address",
Name: "kms base",
Description: "kms base",
OperationalState: ActiveOperationalState.String(),
Type: KmsWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Address: "address",
Name: "kms base",
Description: "kms base",
OperationalState: ActiveOperationalState.String(),
LocalStorageState: AvailableLocalStorageState.String(),
},
},
want: wanted{lastStatusUpdated: true},
@@ -531,11 +616,12 @@ func TestWorker_Create(t *testing.T) {
name: "kms with no name field",
in: Worker{
Worker: &store.Worker{
Type: KmsWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Address: "address",
OperationalState: ActiveOperationalState.String(),
Type: KmsWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Address: "address",
OperationalState: ActiveOperationalState.String(),
LocalStorageState: AvailableLocalStorageState.String(),
},
},
want: wanted{createError: true},
@@ -557,12 +643,13 @@ func TestWorker_Create(t *testing.T) {
name: "kms with upper case name",
in: Worker{
Worker: &store.Worker{
Type: KmsWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Name: "With Upper Case Worker Reported Name",
Address: "address",
OperationalState: ActiveOperationalState.String(),
Type: KmsWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Name: "With Upper Case Worker Reported Name",
Address: "address",
OperationalState: ActiveOperationalState.String(),
LocalStorageState: AvailableLocalStorageState.String(),
},
},
want: wanted{createError: true},
@@ -571,12 +658,13 @@ func TestWorker_Create(t *testing.T) {
name: "kms with unprintable worker reported name",
in: Worker{
Worker: &store.Worker{
Type: KmsWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Name: "unprintable \u0008 worker reported name",
Address: "address",
OperationalState: ActiveOperationalState.String(),
Type: KmsWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Name: "unprintable \u0008 worker reported name",
Address: "address",
OperationalState: ActiveOperationalState.String(),
LocalStorageState: AvailableLocalStorageState.String(),
},
},
want: wanted{createError: true},
@@ -587,11 +675,12 @@ func TestWorker_Create(t *testing.T) {
name: "pki base",
in: Worker{
Worker: &store.Worker{
Type: PkiWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Name: "pki base",
OperationalState: ActiveOperationalState.String(),
Type: PkiWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Name: "pki base",
OperationalState: ActiveOperationalState.String(),
LocalStorageState: AvailableLocalStorageState.String(),
},
},
want: wanted{},
@@ -600,10 +689,11 @@ func TestWorker_Create(t *testing.T) {
name: "pki with no name field",
in: Worker{
Worker: &store.Worker{
Type: PkiWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
OperationalState: ActiveOperationalState.String(),
Type: PkiWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
OperationalState: ActiveOperationalState.String(),
LocalStorageState: AvailableLocalStorageState.String(),
},
},
want: wanted{},
@@ -612,11 +702,12 @@ func TestWorker_Create(t *testing.T) {
name: "pki with address field",
in: Worker{
Worker: &store.Worker{
Type: PkiWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Address: "address",
OperationalState: ActiveOperationalState.String(),
Type: PkiWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Address: "address",
OperationalState: ActiveOperationalState.String(),
LocalStorageState: AvailableLocalStorageState.String(),
},
},
want: wanted{createError: true},
@@ -625,12 +716,13 @@ func TestWorker_Create(t *testing.T) {
name: "pki with upper case name",
in: Worker{
Worker: &store.Worker{
Type: PkiWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Name: "PKI With Upper Case Worker Reported Name",
Address: "address",
OperationalState: ActiveOperationalState.String(),
Type: PkiWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Name: "PKI With Upper Case Worker Reported Name",
Address: "address",
OperationalState: ActiveOperationalState.String(),
LocalStorageState: AvailableLocalStorageState.String(),
},
},
want: wanted{createError: true},
@@ -639,12 +731,41 @@ func TestWorker_Create(t *testing.T) {
name: "pki with unprintable worker reported name",
in: Worker{
Worker: &store.Worker{
Type: PkiWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Name: "unprintable \u0008 worker reported name",
Address: "address",
OperationalState: ActiveOperationalState.String(),
Type: PkiWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Name: "unprintable \u0008 worker reported name",
Address: "address",
OperationalState: ActiveOperationalState.String(),
LocalStorageState: AvailableLocalStorageState.String(),
},
},
want: wanted{createError: true},
},
{
name: "pki with unknown local storage state",
in: Worker{
Worker: &store.Worker{
Type: PkiWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Name: "pki unknown local storage state",
OperationalState: ActiveOperationalState.String(),
LocalStorageState: UnknownLocalStorageState.String(),
},
},
want: wanted{},
},
{
name: "pki with invalid local storage state",
in: Worker{
Worker: &store.Worker{
Type: PkiWorkerType.String(),
ScopeId: scope.Global.String(),
PublicId: newId(),
Name: "pki invalid local storage state",
OperationalState: ActiveOperationalState.String(),
LocalStorageState: "invalid state",
},
},
want: wanted{createError: true},
@@ -695,7 +816,8 @@ func TestWorker_New(t *testing.T) {
args: args{},
want: &Worker{
Worker: &store.Worker{
OperationalState: ActiveOperationalState.String(),
OperationalState: ActiveOperationalState.String(),
LocalStorageState: UnknownLocalStorageState.String(),
},
},
},
@@ -706,8 +828,9 @@ func TestWorker_New(t *testing.T) {
},
want: &Worker{
Worker: &store.Worker{
ScopeId: scope.Global.String(),
OperationalState: ActiveOperationalState.String(),
ScopeId: scope.Global.String(),
OperationalState: ActiveOperationalState.String(),
LocalStorageState: UnknownLocalStorageState.String(),
},
},
},
@@ -718,8 +841,9 @@ func TestWorker_New(t *testing.T) {
},
want: &Worker{
Worker: &store.Worker{
ScopeId: org.GetPublicId(),
OperationalState: ActiveOperationalState.String(),
ScopeId: org.GetPublicId(),
OperationalState: ActiveOperationalState.String(),
LocalStorageState: UnknownLocalStorageState.String(),
},
},
},
@@ -730,8 +854,9 @@ func TestWorker_New(t *testing.T) {
},
want: &Worker{
Worker: &store.Worker{
ScopeId: prj.GetPublicId(),
OperationalState: ActiveOperationalState.String(),
ScopeId: prj.GetPublicId(),
OperationalState: ActiveOperationalState.String(),
LocalStorageState: UnknownLocalStorageState.String(),
},
},
},
@@ -745,9 +870,10 @@ func TestWorker_New(t *testing.T) {
},
want: &Worker{
Worker: &store.Worker{
ScopeId: scope.Global.String(),
Name: "foo",
OperationalState: ActiveOperationalState.String(),
ScopeId: scope.Global.String(),
Name: "foo",
OperationalState: ActiveOperationalState.String(),
LocalStorageState: UnknownLocalStorageState.String(),
},
},
},
@@ -761,9 +887,10 @@ func TestWorker_New(t *testing.T) {
},
want: &Worker{
Worker: &store.Worker{
ScopeId: scope.Global.String(),
Description: "foo",
OperationalState: ActiveOperationalState.String(),
ScopeId: scope.Global.String(),
Description: "foo",
OperationalState: ActiveOperationalState.String(),
LocalStorageState: UnknownLocalStorageState.String(),
},
},
},
@@ -777,9 +904,10 @@ func TestWorker_New(t *testing.T) {
},
want: &Worker{
Worker: &store.Worker{
ScopeId: scope.Global.String(),
Address: "foo",
OperationalState: ActiveOperationalState.String(),
ScopeId: scope.Global.String(),
Address: "foo",
OperationalState: ActiveOperationalState.String(),
LocalStorageState: UnknownLocalStorageState.String(),
},
},
},
@@ -796,8 +924,9 @@ func TestWorker_New(t *testing.T) {
},
want: &Worker{
Worker: &store.Worker{
ScopeId: scope.Global.String(),
OperationalState: ActiveOperationalState.String(),
ScopeId: scope.Global.String(),
OperationalState: ActiveOperationalState.String(),
LocalStorageState: UnknownLocalStorageState.String(),
},
inputTags: []*Tag{
{