diff --git a/internal/daemon/cluster/handlers/worker_service.go b/internal/daemon/cluster/handlers/worker_service.go index 83d7b8ecc..d8dc8acc5 100644 --- a/internal/daemon/cluster/handlers/worker_service.go +++ b/internal/daemon/cluster/handlers/worker_service.go @@ -270,8 +270,14 @@ func (ws *workerServiceServer) ListHcpbWorkers(ctx context.Context, req *pbs.Lis if err != nil { return nil, status.Errorf(codes.Internal, "Error getting servers repo: %v", err) } - - workers, err := serversRepo.ListWorkers(ctx, []string{scope.Global.String()}, server.WithWorkerType(server.KmsWorkerType)) + workers, err := serversRepo.ListWorkers(ctx, []string{scope.Global.String()}, + server.WithWorkerType(server.KmsWorkerType), + // We use the livenessTimeToStale here instead of WorkerStatusGracePeriod + // since WorkerStatusGracePeriod is more for deciding which workers + // should be used for session proxying, but here we care about providing + // the BYOW workers with a list of which upstreams to connect to as their + // upstreams. + server.WithLiveness(time.Duration(ws.livenessTimeToStale.Load()))) if err != nil { return nil, status.Errorf(codes.Internal, "Error looking up workers: %v", err) } diff --git a/internal/daemon/cluster/handlers/worker_service_test.go b/internal/daemon/cluster/handlers/worker_service_test.go index 401b5b497..07447ee61 100644 --- a/internal/daemon/cluster/handlers/worker_service_test.go +++ b/internal/daemon/cluster/handlers/worker_service_test.go @@ -7,6 +7,7 @@ import ( "sync" "sync/atomic" "testing" + "time" "github.com/google/go-cmp/cmp" "github.com/hashicorp/boundary/internal/authtoken" @@ -542,24 +543,34 @@ func TestHcpbWorkers(t *testing.T) { connectionRepoFn := func() (*session.ConnectionRepository, error) { return session.NewConnectionRepository(ctx, rw, rw, kmsCache) } + var liveDur atomic.Int64 + liveDur.Store(int64(1 * time.Second)) - for i := 0; i < 3; i++ { - var opt []server.Option - if i > 0 { - // Out of three KMS workers we expect to see two - opt = append(opt, server.WithWorkerTags(&server.Tag{Key: ManagedWorkerTagKey, Value: "true"})) - } - server.TestKmsWorker(t, conn, wrapper, append(opt, server.WithAddress(fmt.Sprintf("kms.%d", i)))...) - server.TestPkiWorker(t, conn, wrapper, opt...) - } + // Stale/unalive kms worker aren't expected... + server.TestKmsWorker(t, conn, wrapper, server.WithWorkerTags(&server.Tag{Key: ManagedWorkerTagKey, Value: "true"}), + server.WithAddress("old.kms.1")) + // Sleep + 500ms longer than the liveness duration. + time.Sleep(time.Duration(liveDur.Load()) + time.Second) - s := NewWorkerServiceServer(serversRepoFn, workerAuthRepoFn, sessionRepoFn, connectionRepoFn, nil, new(sync.Map), kmsCache, new(atomic.Int64)) + server.TestKmsWorker(t, conn, wrapper, server.WithWorkerTags(&server.Tag{Key: ManagedWorkerTagKey, Value: "true"}), + server.WithAddress("kms.1")) + server.TestKmsWorker(t, conn, wrapper, server.WithWorkerTags(&server.Tag{Key: ManagedWorkerTagKey, Value: "true"}), + server.WithAddress("kms.2")) + + // Shutdown workers will be removed from routes and sessions, but still returned + // to downstream workers + server.TestKmsWorker(t, conn, wrapper, server.WithWorkerTags(&server.Tag{Key: ManagedWorkerTagKey, Value: "true"}), + server.WithAddress("shutdown.kms.3"), server.WithOperationalState(server.ShutdownOperationalState.String())) + // PKI workers aren't expected + server.TestPkiWorker(t, conn, wrapper, server.WithWorkerTags(&server.Tag{Key: ManagedWorkerTagKey, Value: "true"})) + + s := NewWorkerServiceServer(serversRepoFn, workerAuthRepoFn, sessionRepoFn, connectionRepoFn, nil, new(sync.Map), kmsCache, &liveDur) require.NotNil(t, s) res, err := s.ListHcpbWorkers(ctx, &pbs.ListHcpbWorkersRequest{}) require.NoError(err) require.NotNil(res) - expValues := []string{"kms.1", "kms.2"} + expValues := []string{"kms.1", "kms.2", "shutdown.kms.3"} var gotValues []string for _, worker := range res.Workers { gotValues = append(gotValues, worker.Address) diff --git a/internal/server/testing.go b/internal/server/testing.go index 1681bd263..f6215a0e0 100644 --- a/internal/server/testing.go +++ b/internal/server/testing.go @@ -101,23 +101,22 @@ func TestKmsWorker(t *testing.T, conn *db.DB, wrapper wrapping.Wrapper, opt ...O ctx := context.Background() opts := GetOpts(opt...) - namePart, err := newWorkerId(ctx) - require.NoError(t, err) - name := "test-worker-" + strings.ToLower(namePart) - if opts.withName != "" { - name = opts.withName + if opts.withName == "" { + namePart, err := newWorkerId(ctx) + require.NoError(t, err) + name := "test-worker-" + strings.ToLower(namePart) + opt = append(opt, WithName(name)) } - address := "127.0.0.1" - if opts.withAddress != "" { - address = opts.withAddress + if opts.withAddress == "" { + address := "127.0.0.1" + opt = append(opt, WithAddress(address)) } versionInfo := version.Get() relVer := versionInfo.FullVersionNumber(false) - wrk := NewWorker(scope.Global.String(), - WithName(name), - WithAddress(address), - WithDescription(opts.withDescription), - WithReleaseVersion(relVer)) + + opt = append(opt, WithReleaseVersion(relVer)) + + wrk := NewWorker(scope.Global.String(), opt...) wrk, err = serversRepo.UpsertWorkerStatus(ctx, wrk) require.NoError(t, err) require.NotNil(t, wrk) diff --git a/internal/server/testing_test.go b/internal/server/testing_test.go index 184c0209b..8f4d9347e 100644 --- a/internal/server/testing_test.go +++ b/internal/server/testing_test.go @@ -21,7 +21,7 @@ func TestTestKmsWorker(t *testing.T) { description = "test description" address = "test address" ) - tWorker := TestKmsWorker(t, conn, wrapper, WithName(name), WithDescription(description), WithAddress(address)) + tWorker := TestKmsWorker(t, conn, wrapper, WithName(name), WithDescription(description), WithAddress(address), WithOperationalState(ShutdownOperationalState.String())) assert.NotNil(t, tWorker) assert.True(t, strings.HasPrefix(tWorker.GetPublicId(), WorkerPrefix)) @@ -35,6 +35,7 @@ func TestTestKmsWorker(t *testing.T) { assert.Equal(t, name, lkpWorker.GetName()) assert.Equal(t, description, lkpWorker.GetDescription()) assert.Equal(t, address, lkpWorker.GetAddress()) + assert.Equal(t, ShutdownOperationalState.String(), lkpWorker.OperationalState) } func TestTestPkiWorker(t *testing.T) {