feat(workers): worker operational state (#2427)

* feat(workers): worker operational state

internal/db/schema/migrations/oss/postgres/52/01_worker_operational_state.up.sql

@@ -0,0 +1,76 @@
+begin;
+
+create table server_worker_operational_state_enm (
+  state text primary key
+  constraint only_predefined_operational_states_allowed
+    check (
+      state in (
+        'active',
+        'shutdown',
+        'unknown'
+      )
+    )
+);
+comment on table server_worker_operational_state_enm is
+  'server_worker_operational_state_enm is an enumeration table for worker operational states.';
+
+insert into server_worker_operational_state_enm (state) values
+  ('active'),
+  ('shutdown'),
+  ('unknown');
+
+alter table server_worker
+  add column operational_state text not null default 'active'
+    constraint server_worker_operational_state_enm_fkey
+      references server_worker_operational_state_enm (state)
+      on delete restrict
+      on update cascade;
+
+drop view server_worker_aggregate;
+-- Updates view created in 51/01_server_worker_release_version.up.sql to add the worker operational 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,
+  cc.count as active_connection_count,
+  -- keys and tags can be any lowercase printable character so use uppercase characters as delimitors.
+  wt.tags as api_tags,
+  ct.tags as worker_config_tags
+from server_worker w
+ left join 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;

internal/db/schema/migrations/oss/postgres_52_01_test.go

@@ -0,0 +1,101 @@
+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 (
+	insertWorkerQuery = "insert into server_worker (public_id, scope_id, type) values ($1, 'global', 'pki')"
+	selectWorkerQuery = "select public_id, operational_state from server_worker where public_id = $1"
+)
+
+type testWorker struct {
+	OperationalState string
+	PublicId         string
+}
+
+func Test_WorkerOperationalStateChanges(t *testing.T) {
+	t.Parallel()
+	require := require.New(t)
+	const priorMigration = 51001
+	const serverEnumMigration = 52001
+	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, insertWorkerQuery, "test-worker")
+	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": serverEnumMigration}),
+	))
+	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:   serverEnumMigration,
+				DatabaseSchemaVersion: serverEnumMigration,
+				DatabaseSchemaState:   schema.Equal,
+			},
+		},
+	}
+	require.Equal(want, state)
+
+	// Assert worker has been migrated and received default state of 'active'
+	actualWorker := new(testWorker)
+	row := d.QueryRowContext(ctx, selectWorkerQuery, "test-worker")
+	require.NoError(row.Scan(
+		&actualWorker.PublicId,
+		&actualWorker.OperationalState,
+	))
+	require.Equal("active", actualWorker.OperationalState)
+}