feat(worker): Support reloading of initial upstreams on SIGHUP (#2417)

On a SIGHUP, if the worker's `initial_upstreams` config option has
changed, it will use the new values as the address to send its status.
Any previously discovered upstream addresses from previous status
requests are dropped. Once a successful status request is made to the
new addresses, the worker will resume its normal behavior of using the
addresses from the response to the status request.

This allows a worker to be deployed in a more stable way in a dynamic
environment where controller IPs could change. In particular this helps
with a case where all controllers are replaced with new controllers
running on new IPs.

To illustrate with an example, suppose there is an instance group of
controllers instances. And there are worker instances that have their
config rendered from consul-template that sets `initial_upstreams` to
the IPs of the controller instances. If all controller instances are
restarted or replaced at the same time and assigned new IPs, a new
config will be rendered for the worker. Previously the only way for the
worker to use the new value would be for the worker process to be
restarted. Otherwise the worker would only know about the previous IPs
and would fail to connect to the new controllers. However, a restart
would mean any active sessions and session connections on the worker
instance would be terminated. With this update, consul-template can send
a SIGHUP to the worker, the worker will connect to the new controllers
and resume normal status reporting, allowing the sessions and session
connections to remain.

The same principles would apply for other examples of dynamic
environments such as nomad and kubernetes.

See:
    https://github.com/hashicorp/consul-template
    https://www.nomadproject.io/docs/job-specification/template

internal/cmd/commands/server/server.go

@@ -840,7 +840,7 @@ func (c *Command) Reload(newConf *config.Config) error {
 	}
 
 	if newConf != nil && c.worker != nil {
-		c.worker.ParseAndStoreTags(newConf.Worker.Tags)
+		c.worker.Reload(c.Context, newConf)
 	}
 
 	// Send a message that we reloaded. This prevents "guessing" sleep times

internal/cmd/commands/server/worker_initial_upstreams_reload_test.go

@@ -0,0 +1,141 @@
+//go:build !hsm
+
+package server
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/hashicorp/boundary/internal/server"
+	"github.com/hashicorp/boundary/testing/controller"
+	"github.com/mitchellh/cli"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/atomic"
+)
+
+const initialUpstreamConfig = `
+disable_mlock = true
+
+kms "aead" {
+	purpose = "worker-auth"
+	aead_type = "aes-gcm"
+	key = "%s"
+	key_id = "global_worker-auth"
+}
+
+listener "tcp" {
+	purpose = "proxy"
+	address = "127.0.0.1:9406"
+}
+
+worker {
+	name = "test"
+	description = "A default worker created in dev mode"
+	initial_upstreams = ["%s"]
+	tags {
+		type = ["dev", "local"]
+	}
+	auth_storage_path = "%s"
+}
+`
+
+func TestServer_ReloadInitialUpstreams(t *testing.T) {
+	t.Parallel()
+	require := require.New(t)
+
+	rootWrapper, _ := wrapperWithKey(t)
+	recoveryWrapper, _ := wrapperWithKey(t)
+	workerAuthWrapper, key := wrapperWithKey(t)
+
+	// Create two controllers, each with their own database. In practice it
+	// would be odd to have separate databases, but it makes it easy for the
+	// test to assert that the worker has connected to the second controller
+	// after a reload signal is sent.
+	testController := controller.NewTestController(t, controller.WithWorkerAuthKms(workerAuthWrapper), controller.WithRootKms(rootWrapper), controller.WithRecoveryKms(recoveryWrapper))
+	defer testController.Shutdown()
+	testController2 := controller.NewTestController(t, controller.WithWorkerAuthKms(workerAuthWrapper), controller.WithRootKms(rootWrapper), controller.WithRecoveryKms(recoveryWrapper))
+	defer testController2.Shutdown()
+	require.NotEqual(testController.Config().DatabaseUrl, testController2.Config().DatabaseUrl)
+
+	authStoragePath, err := os.MkdirTemp("", "")
+	require.NoError(err)
+	t.Cleanup(func() { os.RemoveAll(authStoragePath) })
+
+	wg := &sync.WaitGroup{}
+
+	cmd := testServerCommand(t, testServerCommandOpts{})
+	cmd.presetConfig = atomic.NewString(fmt.Sprintf(initialUpstreamConfig, key, testController.ClusterAddrs()[0], filepath.Join(authStoragePath, t.Name())))
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		if code := cmd.Run(nil); code != 0 {
+			output := cmd.UI.(*cli.MockUi).ErrorWriter.String() + cmd.UI.(*cli.MockUi).OutputWriter.String()
+			t.Errorf("got a non-zero exit status: %s", output)
+		}
+	}()
+
+	select {
+	case <-cmd.startedCh:
+	case <-time.After(15 * time.Second):
+		t.Fatalf("timeout waiting for worker start")
+	}
+
+	// Wait until the worker has connected to the first controller
+	timeout := time.NewTimer(15 * time.Second)
+	poll := time.NewTimer(0)
+	var w *server.Worker
+pollFirstController:
+	for {
+		select {
+		case <-timeout.C:
+			t.Fatalf("timeout wait for worker to connect to first controller")
+		case <-poll.C:
+			serversRepo, err := testController.Controller().ServersRepoFn()
+			require.NoError(err)
+			w, err = serversRepo.LookupWorkerByName(testController.Context(), "test")
+			require.NoError(err)
+			if w != nil {
+				timeout.Stop()
+				break pollFirstController
+			}
+			poll.Reset(1 * time.Millisecond)
+		}
+	}
+
+	// Reload the config after changing initial_upstreams to the second controller
+	cmd.presetConfig.Store(fmt.Sprintf(initialUpstreamConfig, key, testController2.ClusterAddrs()[0], filepath.Join(authStoragePath, t.Name())))
+	cmd.SighupCh <- struct{}{}
+	select {
+	case <-cmd.reloadedCh:
+	case <-time.After(15 * time.Second):
+		t.Fatalf("timeout waiting for worker reload")
+	}
+
+	// Wait until the worker connects to the second controller
+	timeout.Reset(15 * time.Second)
+	poll.Reset(10 * time.Millisecond)
+pollSecondController:
+	for {
+		select {
+		case <-timeout.C:
+			t.Fatalf("timeout wait for worker to connect to second controller")
+		case <-poll.C:
+			serversRepo, err := testController2.Controller().ServersRepoFn()
+			w, err = serversRepo.LookupWorkerByName(testController2.Context(), "test")
+			require.NoError(err)
+			if w != nil {
+				break pollSecondController
+			}
+			poll.Reset(1 * time.Millisecond)
+		}
+	}
+
+	close(cmd.ShutdownCh)
+
+	wg.Wait()
+}

internal/daemon/worker/status.go

@@ -107,6 +107,8 @@ func (w *Worker) WaitForNextSuccessfulStatusUpdate() error {
 
 func (w *Worker) sendWorkerStatus(cancelCtx context.Context, sessionManager session.Manager, addressReceivers *[]addressReceiver) {
 	const op = "worker.(Worker).sendWorkerStatus"
+	w.statusLock.Lock()
+	defer w.statusLock.Unlock()
 
 	// First send info as-is. We'll perform cleanup duties after we
 	// get cancel/job change info back.

internal/daemon/worker/worker.go

@@ -30,6 +30,7 @@ import (
 	"github.com/hashicorp/go-hclog"
 	"github.com/hashicorp/go-secure-stdlib/base62"
 	"github.com/hashicorp/go-secure-stdlib/mlock"
+	"github.com/hashicorp/go-secure-stdlib/strutil"
 	nodeefile "github.com/hashicorp/nodeenrollment/storage/file"
 	"github.com/hashicorp/nodeenrollment/types"
 	ua "go.uber.org/atomic"
@@ -119,6 +120,8 @@ type Worker struct {
 	TestOverrideX509VerifyDnsName  string
 	TestOverrideX509VerifyCertPool *x509.CertPool
 	TestOverrideAuthRotationPeriod time.Duration
+
+	statusLock sync.Mutex
 }
 
 func New(conf *Config) (*Worker, error) {
@@ -154,7 +157,7 @@ func New(conf *Config) (*Worker, error) {
 		conf.RawConfig.Worker = new(config.Worker)
 	}
 
-	w.ParseAndStoreTags(conf.RawConfig.Worker.Tags)
+	w.parseAndStoreTags(conf.RawConfig.Worker.Tags)
 
 	if conf.SecureRandomReader == nil {
 		conf.SecureRandomReader = rand.Reader
@@ -200,6 +203,35 @@ func New(conf *Config) (*Worker, error) {
 	return w, nil
 }
 
+// Reload will update a worker with a new Config. The worker will only use
+// relevant parts of the new config, specifically:
+// - Worker Tags
+// - Initial Upstream addresses
+func (w *Worker) Reload(ctx context.Context, newConf *config.Config) {
+	const op = "worker.(Worker).Reload"
+
+	w.parseAndStoreTags(newConf.Worker.Tags)
+
+	if !strutil.EquivalentSlices(newConf.Worker.InitialUpstreams, w.conf.RawConfig.Worker.InitialUpstreams) {
+		w.statusLock.Lock()
+		defer w.statusLock.Unlock()
+
+		upstreamsMessage := fmt.Sprintf(
+			"Initial Upstreams has changed; old upstreams were: %s, new upstreams are: %s",
+			w.conf.RawConfig.Worker.InitialUpstreams,
+			newConf.Worker.InitialUpstreams,
+		)
+		event.WriteSysEvent(ctx, op, upstreamsMessage)
+		w.conf.RawConfig.Worker.InitialUpstreams = newConf.Worker.InitialUpstreams
+
+		for _, ar := range w.addressReceivers {
+			ar.SetAddresses(w.conf.RawConfig.Worker.InitialUpstreams)
+			// set InitialAddresses in case the worker has not successfully dialed yet
+			ar.InitialAddresses(w.conf.RawConfig.Worker.InitialUpstreams)
+		}
+	}
+}
+
 func (w *Worker) Start() error {
 	const op = "worker.(Worker).Start"
 
@@ -431,7 +463,7 @@ func (w *Worker) Shutdown() error {
 	return nil
 }
 
-func (w *Worker) ParseAndStoreTags(incoming map[string][]string) {
+func (w *Worker) parseAndStoreTags(incoming map[string][]string) {
 	if len(incoming) == 0 {
 		w.tags.Store([]*pb.TagPair{})
 		return