From 5d0cdf680f83bfc44f3c68237f087526911c944d Mon Sep 17 00:00:00 2001 From: Timothy Messier Date: Thu, 1 Sep 2022 14:13:42 -0400 Subject: [PATCH] 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 | 2 +- .../worker_initial_upstreams_reload_test.go | 141 ++++++++++++++++++ internal/daemon/worker/status.go | 2 + internal/daemon/worker/worker.go | 36 ++++- 4 files changed, 178 insertions(+), 3 deletions(-) create mode 100644 internal/cmd/commands/server/worker_initial_upstreams_reload_test.go diff --git a/internal/cmd/commands/server/server.go b/internal/cmd/commands/server/server.go index 3631c1ac0..b7a713548 100644 --- a/internal/cmd/commands/server/server.go +++ b/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 diff --git a/internal/cmd/commands/server/worker_initial_upstreams_reload_test.go b/internal/cmd/commands/server/worker_initial_upstreams_reload_test.go new file mode 100644 index 000000000..c83a50558 --- /dev/null +++ b/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() +} diff --git a/internal/daemon/worker/status.go b/internal/daemon/worker/status.go index 2fa90c97c..11b103b04 100644 --- a/internal/daemon/worker/status.go +++ b/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. diff --git a/internal/daemon/worker/worker.go b/internal/daemon/worker/worker.go index 8eb15ec01..e5eb86feb 100644 --- a/internal/daemon/worker/worker.go +++ b/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