Only return live workers in ListHcpbWorkersResponse (#2876)

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)
 	}

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)

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)

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) {