test(server): Fix flakiness in TestServer_ReloadWorkerTags

internal/cmd/commands/server/worker_initial_upstreams_reload_test.go

@@ -97,7 +97,7 @@ pollFirstController:
 	for {
 		select {
 		case <-timeout.C:
-			t.Fatalf("timeout wait for worker to connect to first controller")
+			t.Fatalf("timeout waiting for worker to connect to first controller")
 		case <-poll.C:
 			w, err = serversRepo.LookupWorkerByName(testController.Context(), "test")
 			require.NoError(err)
@@ -168,7 +168,7 @@ pollSecondController:
 	for {
 		select {
 		case <-timeout.C:
-			t.Fatalf("timeout wait for worker to connect to second controller")
+			t.Fatalf("timeout waiting for worker to connect to second controller")
 		case <-poll.C:
 			w, err = serversRepo.LookupWorkerByName(testController2.Context(), "test")
 			require.NoError(err)

internal/cmd/commands/server/worker_tags_reload_test.go

@@ -21,6 +21,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/hashicorp/boundary/internal/server"
 	"github.com/hashicorp/boundary/testing/controller"
 	wrapping "github.com/hashicorp/go-kms-wrapping/v2"
 	"github.com/hashicorp/go-kms-wrapping/v2/aead"
@@ -74,8 +75,14 @@ func TestServer_ReloadWorkerTags(t *testing.T) {
 	rootWrapper, _ := wrapperWithKey(t)
 	recoveryWrapper, _ := wrapperWithKey(t)
 	workerAuthWrapper, key := wrapperWithKey(t)
-	testController := controller.NewTestController(t, controller.WithWorkerAuthKms(workerAuthWrapper), controller.WithRootKms(rootWrapper), controller.WithRecoveryKms(recoveryWrapper))
-	defer testController.Shutdown()
+	testController := controller.NewTestController(
+		t,
+		controller.WithWorkerAuthKms(workerAuthWrapper),
+		controller.WithRootKms(rootWrapper),
+		controller.WithRecoveryKms(recoveryWrapper),
+		controller.WithEnableEventing(),
+	)
+	t.Cleanup(testController.Shutdown)
 
 	wg := &sync.WaitGroup{}
 
@@ -94,7 +101,7 @@ func TestServer_ReloadWorkerTags(t *testing.T) {
 	select {
 	case <-cmd.startedCh:
 	case <-time.After(15 * time.Second):
-		t.Fatalf("timeout")
+		t.Fatalf("timeout waiting for worker start")
 	}
 
 	fetchWorkerTags := func(name string, key string, values []string) {
@@ -104,29 +111,95 @@ func TestServer_ReloadWorkerTags(t *testing.T) {
 		w, err := serversRepo.LookupWorkerByName(testController.Context(), name)
 		require.NoError(err)
 		require.NotNil(w)
-		v, ok := w.CanonicalTags()[key]
-		require.True(ok)
+		tags := w.CanonicalTags()
+		v, ok := tags[key]
+		require.True(ok, fmt.Sprintf("EXPECTED: map[%s:%s], ACTUAL: %+v", key, values, tags))
 		require.ElementsMatch(values, v)
 	}
 
 	// Give time to populate up to the controller
-	time.Sleep(10 * time.Second)
+	// Wait until the worker has connected to the controller as seen via
+	// two status updates
+	t.Log("Waiting for worker to connect to controller...")
+	timeout := time.NewTimer(15 * time.Second)
+	poll := time.NewTimer(0)
+	var w *server.Worker
+	var lastStatusTime time.Time
+	serversRepo, err := testController.Controller().ServersRepoFn()
+	require.NoError(err)
+pollController:
+	for {
+		select {
+		case <-timeout.C:
+			t.Fatalf("timeout waiting for worker to connect to controller")
+		case <-poll.C:
+			w, err = serversRepo.LookupWorkerByName(testController.Context(), "test")
+			require.NoError(err)
+			if w != nil {
+				switch {
+				case lastStatusTime.IsZero():
+					lastStatusTime = w.GetLastStatusTime().AsTime().Round(time.Second)
+				default:
+					if !lastStatusTime.Equal(w.GetLastStatusTime().AsTime().Round(time.Second)) {
+						timeout.Stop()
+						break pollController
+					}
+				}
+			}
+			poll.Reset(time.Second)
+		}
+	}
+
 	fetchWorkerTags("test", "type", []string{"dev", "local"})
 
+	// Set new tags on worker
+	t.Log("Restart worker with new tags...")
 	cmd.presetConfig.Store(fmt.Sprintf(workerBaseConfig+tag2Config, key, testController.ClusterAddrs()[0]))
-
+	t.Logf("%s", workerBaseConfig+tag2Config)
 	cmd.SighupCh <- struct{}{}
 	select {
 	case <-cmd.reloadedCh:
 	case <-time.After(15 * time.Second):
-		t.Fatalf("timeout")
+		t.Fatalf("timeout waiting for worker start")
+	}
+
+	// Give time to populate up to the controller
+	t.Log("Waiting for worker to connect to controller again...")
+	timeout.Reset(15 * time.Second)
+	poll.Reset(time.Second)
+	lastStatusTime = time.Time{}
+	i := 0
+pollControllerAgain:
+	for {
+		select {
+		case <-timeout.C:
+			t.Fatalf("timeout waiting for worker to connect to controller")
+		case <-poll.C:
+			w, err = serversRepo.LookupWorkerByName(testController.Context(), "test")
+			require.NoError(err)
+			if w != nil {
+				t.Logf("%s, %+v", w.GetLastStatusTime().AsTime().Round(time.Second), w.GetConfigTags())
+				switch {
+				case lastStatusTime.IsZero():
+					lastStatusTime = w.GetLastStatusTime().AsTime().Round(time.Second)
+				default:
+					if !lastStatusTime.Round(time.Second).Equal(w.GetLastStatusTime().AsTime().Round(time.Second)) {
+						if i == 3 {
+							timeout.Stop()
+							break pollControllerAgain
+						} else {
+							i += 1
+						}
+					}
+				}
+			}
+			poll.Reset(time.Second)
+		}
 	}
 
-	time.Sleep(10 * time.Second)
 	fetchWorkerTags("test", "foo", []string{"bar", "baz"})
 
 	cmd.ShutdownCh <- struct{}{}
-
 	wg.Wait()
 }
 

testing/controller/controller.go

@@ -124,8 +124,8 @@ func DisableDatabaseCreation() Option {
 	}
 }
 
-// DisableDatabaseCreation skips creating a database in docker and allows one to
-// be provided through a tcOptions.
+// DisableDatabaseDestruction skips destoying the database in docker and allows
+// it to be examined after-the-fact
 func DisableDatabaseDestruction() Option {
 	return func(c *option) error {
 		c.setDisableDatabaseDestruction = true