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