From 40ba275759fd8aa9a6daa7d49f17c1b47a58c1a2 Mon Sep 17 00:00:00 2001 From: Haotian Date: Mon, 13 Jun 2022 15:06:03 -0700 Subject: [PATCH] feat(worker): add 'initial_upstreams' field to worker config to prepare to replace 'controllers' field. Update worker config description on website. --- internal/cmd/commands/dev/dev.go | 2 +- internal/cmd/commands/server/server.go | 27 +-- .../server/worker_shutdown_reload_test.go | 2 +- .../server/worker_tags_reload_test.go | 4 +- internal/cmd/config/config.go | 53 ++++-- internal/cmd/config/config_test.go | 154 +++++++++++------- .../credential/vault/setup_manual_test.go | 8 +- internal/daemon/worker/auth_rotation_test.go | 2 +- .../daemon/worker/controller_connection.go | 12 +- internal/daemon/worker/testing.go | 12 +- internal/tests/cluster/ipv6_listener_test.go | 8 +- .../cluster/multi_controller_worker_test.go | 12 +- .../tests/cluster/session_cleanup_test.go | 4 +- internal/tests/cluster/unix_listener_test.go | 8 +- internal/tests/cluster/worker_replay_test.go | 12 +- internal/tests/cluster/worker_tagging_test.go | 24 +-- .../tests/cluster/x509_verification_test.go | 10 +- website/content/docs/configuration/worker.mdx | 4 +- 18 files changed, 208 insertions(+), 150 deletions(-) diff --git a/internal/cmd/commands/dev/dev.go b/internal/cmd/commands/dev/dev.go index 6d3c65904..8081efa49 100644 --- a/internal/cmd/commands/dev/dev.go +++ b/internal/cmd/commands/dev/dev.go @@ -483,7 +483,7 @@ func (c *Command) Run(args []string) int { if c.flagControllerClusterListenAddr != "" { l.Address = c.flagControllerClusterListenAddr if !c.flagControllerOnly { - c.Config.Worker.Controllers = []string{l.Address} + c.Config.Worker.InitialUpstreams = []string{l.Address} } } else { l.Address = "127.0.0.1:9201" diff --git a/internal/cmd/commands/server/server.go b/internal/cmd/commands/server/server.go index 0f5e21451..9a7a07897 100644 --- a/internal/cmd/commands/server/server.go +++ b/internal/cmd/commands/server/server.go @@ -280,6 +280,9 @@ func (c *Command) Run(args []string) int { c.UI.Error(`Config activates worker but no listener with "proxy" purpose found`) return base.CommandUserError } + if c.Config.Worker.ControllersRaw != nil { + c.UI.Warn("The \"controllers\" field for worker config is deprecated. Please use \"initial_upstreams\" instead.") + } if err := c.SetupWorkerPublicAddress(c.Config, ""); err != nil { c.UI.Error(err.Error()) @@ -289,25 +292,25 @@ func (c *Command) Run(args []string) int { c.Info["worker public proxy addr"] = c.Config.Worker.PublicAddr if c.Config.Controller != nil { - switch len(c.Config.Worker.Controllers) { + switch len(c.Config.Worker.InitialUpstreams) { case 0: if c.Config.Controller.PublicClusterAddr != "" { clusterAddr = c.Config.Controller.PublicClusterAddr } - c.Config.Worker.Controllers = []string{clusterAddr} + c.Config.Worker.InitialUpstreams = []string{clusterAddr} case 1: - if c.Config.Worker.Controllers[0] == clusterAddr { + if c.Config.Worker.InitialUpstreams[0] == clusterAddr { break } if c.Config.Controller.PublicClusterAddr != "" && - c.Config.Worker.Controllers[0] == c.Config.Controller.PublicClusterAddr { + c.Config.Worker.InitialUpstreams[0] == c.Config.Controller.PublicClusterAddr { break } // Best effort see if it's a domain name and if not assume it must match - host, _, err := net.SplitHostPort(c.Config.Worker.Controllers[0]) + host, _, err := net.SplitHostPort(c.Config.Worker.InitialUpstreams[0]) if err != nil && strings.Contains(err.Error(), "missing port in address") { err = nil - host = c.Config.Worker.Controllers[0] + host = c.Config.Worker.InitialUpstreams[0] } if err == nil { ip := net.ParseIP(host) @@ -318,17 +321,17 @@ func (c *Command) Run(args []string) int { } fallthrough default: - c.UI.Error(`When running a combined controller and worker, it's invalid to specify a "controllers" key in the worker block with any value other than the controller cluster address/port when using IPs rather than DNS names`) + c.UI.Error(`When running a combined controller and worker, it's invalid to specify a "initial_upstreams" or "controllers" key in the worker block with any values other than the controller cluster or upstream worker address/port when using IPs rather than DNS names`) return base.CommandUserError } } - for _, controller := range c.Config.Worker.Controllers { - host, _, err := net.SplitHostPort(controller) + for _, upstream := range c.Config.Worker.InitialUpstreams { + host, _, err := net.SplitHostPort(upstream) if err != nil { if strings.Contains(err.Error(), "missing port in address") { - host = controller + host = upstream } else { - c.UI.Error(fmt.Errorf("Invalid controller address %q: %w", controller, err).Error()) + c.UI.Error(fmt.Errorf("Invalid worker upstream address %q: %w", upstream, err).Error()) return base.CommandUserError } } @@ -342,7 +345,7 @@ func (c *Command) Run(args []string) int { errMsg = "a multicast" } if errMsg != "" { - c.UI.Error(fmt.Sprintf("Controller address %q is invalid: cannot be %s address", controller, errMsg)) + c.UI.Error(fmt.Sprintf("Worker upstream address %q is invalid: cannot be %s address", upstream, errMsg)) return base.CommandUserError } } diff --git a/internal/cmd/commands/server/worker_shutdown_reload_test.go b/internal/cmd/commands/server/worker_shutdown_reload_test.go index 7876e870e..546840464 100644 --- a/internal/cmd/commands/server/worker_shutdown_reload_test.go +++ b/internal/cmd/commands/server/worker_shutdown_reload_test.go @@ -31,7 +31,7 @@ disable_mlock = true worker { name = "w_1234567890" description = "A default worker created in dev mode" - controllers = ["%s"] + initial_upstreams = ["%s"] } kms "aead" { diff --git a/internal/cmd/commands/server/worker_tags_reload_test.go b/internal/cmd/commands/server/worker_tags_reload_test.go index 01114b811..74bc29f01 100644 --- a/internal/cmd/commands/server/worker_tags_reload_test.go +++ b/internal/cmd/commands/server/worker_tags_reload_test.go @@ -48,7 +48,7 @@ const tag1Config = ` worker { name = "test" description = "A default worker created in dev mode" - controllers = ["%s"] + initial_upstreams = ["%s"] tags { type = ["dev", "local"] } @@ -60,7 +60,7 @@ const tag2Config = ` worker { name = "test" description = "A default worker created in dev mode" - controllers = ["%s"] + initial_upstreams = ["%s"] tags { foo = ["bar", "baz"] } diff --git a/internal/cmd/config/config.go b/internal/cmd/config/config.go index 7e054b082..95c048d49 100644 --- a/internal/cmd/config/config.go +++ b/internal/cmd/config/config.go @@ -96,7 +96,7 @@ listener "tcp" { worker { name = "w_1234567890" description = "A default worker created in dev mode" - controllers = ["127.0.0.1"] + initial_upstreams = ["127.0.0.1"] tags { type = ["dev", "local"] } @@ -181,8 +181,12 @@ type Worker struct { // We use a raw interface here so that we can take in a string // value pointing to an env var or file. We then resolve that - // and get the actual controller addresses. - Controllers []string `hcl:"-"` + // and get the actual upstream controller or worker addresses. + InitialUpstreams []string `hcl:"-"` + InitialUpstreamsRaw any `hcl:"initial_upstreams"` + + // The ControllersRaw field is deprecated and users should use InitialUpstreamsRaw instead. + // TODO: remove this field when support is discontinued. ControllersRaw interface{} `hcl:"controllers"` // We use a raw interface for parsing so that people can use JSON-like @@ -525,9 +529,9 @@ func Parse(d string) (*Config, error) { } } - result.Worker.Controllers, err = parseWorkerControllers(result) + result.Worker.InitialUpstreams, err = parseWorkerUpstreams(result) if err != nil { - return nil, fmt.Errorf("Failed to parse worker controllers: %w", err) + return nil, fmt.Errorf("Failed to parse worker upstreams: %w", err) } } @@ -587,35 +591,52 @@ func Parse(d string) (*Config, error) { return result, nil } -func parseWorkerControllers(c *Config) ([]string, error) { +// supportControllersRawConfig returns either initialUpstreamsRaw or controllersRaw depending on which is populated. Errors when both fields are populated. +// +func supportControllersRawConfig(initialUpstreamsRaw, controllersRaw any) (any, error) { + switch { + case initialUpstreamsRaw == nil && controllersRaw != nil: + return controllersRaw, nil + case initialUpstreamsRaw != nil && controllersRaw != nil: + return nil, fmt.Errorf("both initial_upstreams and controllers fields are populated") + } + return initialUpstreamsRaw, nil +} + +func parseWorkerUpstreams(c *Config) ([]string, error) { if c == nil || c.Worker == nil { return nil, fmt.Errorf("config or worker field is nil") } - if c.Worker.ControllersRaw == nil { + if c.Worker.InitialUpstreamsRaw == nil && c.Worker.ControllersRaw == nil { + // return nil here so that other address sources can be provided outside of config return nil, nil } + rawUpstreams, err := supportControllersRawConfig(c.Worker.InitialUpstreamsRaw, c.Worker.ControllersRaw) + if err != nil { + return nil, err + } - switch t := c.Worker.ControllersRaw.(type) { + switch t := rawUpstreams.(type) { case []interface{}: // An array was configured directly in Boundary's HCL Config file. - var controllers []string - err := mapstructure.WeakDecode(c.Worker.ControllersRaw, &controllers) + var upstreams []string + err := mapstructure.WeakDecode(rawUpstreams, &upstreams) if err != nil { - return nil, fmt.Errorf("failed to decode worker controllers block into config field: %w", err) + return nil, fmt.Errorf("failed to decode worker initial_upstreams block into config field: %w", err) } - return controllers, nil + return upstreams, nil case string: - controllersStr, err := parseutil.ParsePath(t) + upstreamsStr, err := parseutil.ParsePath(t) if err != nil { return nil, fmt.Errorf("bad env var or file pointer: %w", err) } - var addrs []string - err = json.Unmarshal([]byte(controllersStr), &addrs) + var upstreams []string + err = json.Unmarshal([]byte(upstreamsStr), &upstreams) if err != nil { return nil, fmt.Errorf("failed to unmarshal env/file contents: %w", err) } - return addrs, nil + return upstreams, nil default: typ := reflect.TypeOf(t) diff --git a/internal/cmd/config/config_test.go b/internal/cmd/config/config_test.go index 44bbc5e34..6afefae26 100644 --- a/internal/cmd/config/config_test.go +++ b/internal/cmd/config/config_test.go @@ -190,10 +190,10 @@ func TestDevWorker(t *testing.T) { }, }, Worker: &Worker{ - Name: "w_1234567890", - Description: "A default worker created in dev mode", - Controllers: []string{"127.0.0.1"}, - ControllersRaw: []interface{}{"127.0.0.1"}, + Name: "w_1234567890", + Description: "A default worker created in dev mode", + InitialUpstreams: []string{"127.0.0.1"}, + InitialUpstreamsRaw: []interface{}{"127.0.0.1"}, Tags: map[string][]string{ "type": {"dev", "local"}, }, @@ -213,7 +213,7 @@ func TestDevWorker(t *testing.T) { worker { name = "w_1234567890" description = "A default worker created in dev mode" - controllers = ["127.0.0.1"] + initial_upstreams = ["127.0.0.1"] tags = ["type=dev", "type=local"] } ` @@ -233,7 +233,7 @@ func TestDevWorker(t *testing.T) { worker { name = "w_1234567890" description = "A default worker created in dev mode" - controllers = ["127.0.0.1"] + initial_upstreams = ["127.0.0.1"] tags { type = "local" } @@ -258,7 +258,7 @@ func TestDevWorker(t *testing.T) { worker { name = "w_1234567890" description = "A default worker created in dev mode" - controllers = ["127.0.0.1"] + initial_upstreams = ["127.0.0.1"] tags = ["tyPe=dev", "type=local"] } ` @@ -275,7 +275,7 @@ func TestDevWorker(t *testing.T) { worker { name = "w_1234567890" description = "A default worker created in dev mode" - controllers = ["127.0.0.1"] + initial_upstreams = ["127.0.0.1"] tags = ["type=dev", "type=loCal"] } ` @@ -292,7 +292,7 @@ func TestDevWorker(t *testing.T) { worker { name = "dev-work\u0000er" description = "A default worker created in dev mode" - controllers = ["127.0.0.1"] + initial_upstreams = ["127.0.0.1"] tags = ["type=dev", "type=local"] } ` @@ -376,10 +376,10 @@ func TestDevCombined(t *testing.T) { }, DevController: true, Worker: &Worker{ - Name: "w_1234567890", - Description: "A default worker created in dev mode", - Controllers: []string{"127.0.0.1"}, - ControllersRaw: []interface{}{"127.0.0.1"}, + Name: "w_1234567890", + Description: "A default worker created in dev mode", + InitialUpstreams: []string{"127.0.0.1"}, + InitialUpstreamsRaw: []interface{}{"127.0.0.1"}, Tags: map[string][]string{ "type": {"dev", "local"}, }, @@ -419,7 +419,7 @@ func TestDevWorkerCredentialStorageDir(t *testing.T) { worker { name = "w_1234567890" description = "A default worker created in dev mode" - controllers = ["127.0.0.1"] + initial_upstreams = ["127.0.0.1"] tags { type = ["dev", "local"] } @@ -438,7 +438,7 @@ func TestDevWorkerCredentialStorageDir(t *testing.T) { worker { name = "w_1234567890" description = "A default worker created in dev mode" - controllers = ["127.0.0.1"] + initial_upstreams = ["127.0.0.1"] tags { type = ["dev", "local"] } @@ -1039,118 +1039,152 @@ func TestController_EventingConfig(t *testing.T) { } } -func TestWorkerControllers(t *testing.T) { +func TestWorkerUpstreams(t *testing.T) { tests := []struct { - name string - in string - stateFn func(t *testing.T) - expWorkerControllers []string - expErr bool - expErrIs error - expErrStr string + name string + in string + stateFn func(t *testing.T) + expWorkerUpstreams []string + expErr bool + expErrIs error + expErrStr string }{ { - name: "No Controllers", + name: "No Upstreams", in: ` worker { name = "test" } `, - expWorkerControllers: nil, - expErr: false, + expWorkerUpstreams: nil, + expErr: false, }, { - name: "One Controller", + name: "One Upstream", in: ` worker { name = "test" - controllers = ["127.0.0.1"] + initial_upstreams = ["127.0.0.1"] } `, - expWorkerControllers: []string{"127.0.0.1"}, - expErr: false, + expWorkerUpstreams: []string{"127.0.0.1"}, + expErr: false, }, { - name: "Multiple controllers", + name: "Multiple Upstreams", in: ` worker { name = "test" - controllers = ["127.0.0.1", "127.0.0.2", "127.0.0.3"] + initial_upstreams = ["127.0.0.1", "127.0.0.2", "127.0.0.3"] } `, - expWorkerControllers: []string{"127.0.0.1", "127.0.0.2", "127.0.0.3"}, - expErr: false, + expWorkerUpstreams: []string{"127.0.0.1", "127.0.0.2", "127.0.0.3"}, + expErr: false, }, { name: "Using env var", in: ` worker { name = "test" - controllers = "env://BOUNDARY_WORKER_CONTROLLERS" + initial_upstreams = "env://BOUNDARY_WORKER_UPSTREAMS" } `, - stateFn: func(t *testing.T) { t.Setenv("BOUNDARY_WORKER_CONTROLLERS", `["127.0.0.1", "127.0.0.2", "127.0.0.3"]`) }, - expWorkerControllers: []string{"127.0.0.1", "127.0.0.2", "127.0.0.3"}, - expErr: false, + stateFn: func(t *testing.T) { t.Setenv("BOUNDARY_WORKER_UPSTREAMS", `["127.0.0.1", "127.0.0.2", "127.0.0.3"]`) }, + expWorkerUpstreams: []string{"127.0.0.1", "127.0.0.2", "127.0.0.3"}, + expErr: false, }, { name: "Using env var - invalid input 1", in: ` worker { name = "test" - controllers = "env://BOUNDARY_WORKER_CONTROLLERS" + initial_upstreams = "env://BOUNDARY_WORKER_UPSTREAMS" } `, stateFn: func(t *testing.T) { - controllers := ` + upstreams := ` worker { - controllers = ["127.0.0.1"] + initial_upstreams = ["127.0.0.1"] } ` - t.Setenv("BOUNDARY_WORKER_CONTROLLERS", controllers) + t.Setenv("BOUNDARY_WORKER_UPSTREAMS", upstreams) }, - expWorkerControllers: nil, - expErr: true, - expErrStr: "Failed to parse worker controllers: failed to unmarshal env/file contents: invalid character 'w' looking for beginning of value", + expWorkerUpstreams: nil, + expErr: true, + expErrStr: "Failed to parse worker upstreams: failed to unmarshal env/file contents: invalid character 'w' looking for beginning of value", }, { name: "Using env var - invalid input 2", in: ` worker { name = "test" - controllers = "env://BOUNDARY_WORKER_CONTROLLERS" + initial_upstreams = "env://BOUNDARY_WORKER_UPSTREAMS" } `, - stateFn: func(t *testing.T) { t.Setenv("BOUNDARY_WORKER_CONTROLLERS", `controllers = ["127.0.0.1"]`) }, - expWorkerControllers: nil, - expErr: true, - expErrStr: "Failed to parse worker controllers: failed to unmarshal env/file contents: invalid character 'c' looking for beginning of value", + stateFn: func(t *testing.T) { t.Setenv("BOUNDARY_WORKER_UPSTREAMS", `initial_upstreams = ["127.0.0.1"]`) }, + expWorkerUpstreams: nil, + expErr: true, + expErrStr: "Failed to parse worker upstreams: failed to unmarshal env/file contents: invalid character 'i' looking for beginning of value", }, { name: "Unsupported object", in: ` worker { name = "test" - controllers = { + initial_upstreams = { ip = "127.0.0.1" ip = "127.0.0.2" } } `, - expWorkerControllers: nil, - expErr: true, - expErrStr: "Failed to parse worker controllers: unexpected type \"[]map[string]interface {}\"", + expWorkerUpstreams: nil, + expErr: true, + expErrStr: "Failed to parse worker upstreams: unexpected type \"[]map[string]interface {}\"", }, { - name: "worker controllers set to invalid url", + name: "Worker initial_upstreams set to invalid url", in: ` worker { name = "test" - controllers = "env://\x00" + initial_upstreams = "env://\x00" }`, - expWorkerControllers: nil, - expErr: true, - expErrIs: parseutil.ErrNotAUrl, + expWorkerUpstreams: nil, + expErr: true, + expErrIs: parseutil.ErrNotAUrl, + }, + { + name: "Worker using deprecated controllers field", + in: ` + worker { + name = "test" + controllers = ["127.0.0.1", "127.0.0.2", "127.0.0.3"] + }`, + expWorkerUpstreams: []string{"127.0.0.1", "127.0.0.2", "127.0.0.3"}, + expErr: false, + }, + { + name: "Different values in controllers and initial_upstreams field", + in: ` + worker { + name = "test" + controllers = ["127.0.0.1", "127.0.0.2", "127.0.0.3"] + initial_upstreams = ["127.0.0.1"] + }`, + expWorkerUpstreams: nil, + expErr: true, + expErrStr: "Failed to parse worker upstreams: both initial_upstreams and controllers fields are populated", + }, + { + name: "Identical values in controllers and initial_upstreams field", + in: ` + worker { + name = "test" + controllers = ["127.0.0.1"] + initial_upstreams = ["127.0.0.1"] + }`, + expWorkerUpstreams: nil, + expErr: true, + expErrStr: "Failed to parse worker upstreams: both initial_upstreams and controllers fields are populated", }, } @@ -1174,7 +1208,7 @@ func TestWorkerControllers(t *testing.T) { require.NoError(t, err) require.NotNil(t, c) require.NotNil(t, c.Worker) - require.EqualValues(t, tt.expWorkerControllers, c.Worker.Controllers) + require.EqualValues(t, tt.expWorkerUpstreams, c.Worker.InitialUpstreams) }) } } diff --git a/internal/credential/vault/setup_manual_test.go b/internal/credential/vault/setup_manual_test.go index cbd01a81f..23ed9c902 100644 --- a/internal/credential/vault/setup_manual_test.go +++ b/internal/credential/vault/setup_manual_test.go @@ -91,10 +91,10 @@ func TestSetupSleepyDevEnvironment(t *testing.T) { "foo": {"bar"}, } w1 := worker.NewTestWorker(t, &worker.TestWorkerOpts{ - Config: conf, - WorkerAuthKms: c1.Config().WorkerAuthKms, - InitialControllers: c1.ClusterAddrs(), - Logger: logger.Named("w1"), + Config: conf, + WorkerAuthKms: c1.Config().WorkerAuthKms, + InitialUpstreams: c1.ClusterAddrs(), + Logger: logger.Named("w1"), }) defer w1.Shutdown() addr := c1.ApiAddrs()[0] diff --git a/internal/daemon/worker/auth_rotation_test.go b/internal/daemon/worker/auth_rotation_test.go index c35e6213e..4e306d66c 100644 --- a/internal/daemon/worker/auth_rotation_test.go +++ b/internal/daemon/worker/auth_rotation_test.go @@ -42,7 +42,7 @@ func TestRotationTicking(t *testing.T) { const rotationPeriod = 20 * time.Second w := worker.NewTestWorker(t, &worker.TestWorkerOpts{ - InitialControllers: c.ClusterAddrs(), + InitialUpstreams: c.ClusterAddrs(), Logger: logger.Named("worker"), AuthRotationPeriod: rotationPeriod, }) diff --git a/internal/daemon/worker/controller_connection.go b/internal/daemon/worker/controller_connection.go index aa4d1934d..a7d77bbf9 100644 --- a/internal/daemon/worker/controller_connection.go +++ b/internal/daemon/worker/controller_connection.go @@ -41,8 +41,8 @@ const hcpbUrlSuffix = ".proxy.boundary.hashicorp.cloud" // connection client creation func (w *Worker) StartControllerConnections() error { const op = "worker.(Worker).StartControllerConnections" - initialAddrs := make([]resolver.Address, 0, len(w.conf.RawConfig.Worker.Controllers)) - for _, addr := range w.conf.RawConfig.Worker.Controllers { + initialAddrs := make([]resolver.Address, 0, len(w.conf.RawConfig.Worker.InitialUpstreams)) + for _, addr := range w.conf.RawConfig.Worker.InitialUpstreams { switch { case strings.HasPrefix(addr, "/"): initialAddrs = append(initialAddrs, resolver.Address{Addr: addr}) @@ -52,7 +52,7 @@ func (w *Worker) StartControllerConnections() error { host, port, err = net.SplitHostPort(net.JoinHostPort(addr, "9201")) } if err != nil { - return fmt.Errorf("error parsing controller address: %w", err) + return fmt.Errorf("error parsing upstream address: %w", err) } initialAddrs = append(initialAddrs, resolver.Address{Addr: net.JoinHostPort(host, port)}) } @@ -64,7 +64,7 @@ func (w *Worker) StartControllerConnections() error { initialAddrs = append(initialAddrs, resolver.Address{Addr: clusterAddress}) event.WriteSysEvent(w.baseContext, op, fmt.Sprintf("Setting HCPB Cluster address %s as upstream address", clusterAddress)) } else { - return errors.New("no initial controller addresses found") + return errors.New("no initial upstream addresses found") } } @@ -72,7 +72,7 @@ func (w *Worker) StartControllerConnections() error { Addresses: initialAddrs, }) if err := w.createClientConn(initialAddrs[0].Addr); err != nil { - return fmt.Errorf("error making client connection to controller: %w", err) + return fmt.Errorf("error making client connection to upstream address: %w", err) } return nil @@ -135,7 +135,7 @@ func (w *Worker) v1KmsAuthDialFn(ctx context.Context, addr string) (net.Conn, er nonTlsConn, err = dialer.DialContext(ctx, "tcp", addr) } if err != nil { - return nil, fmt.Errorf("unable to dial to controller: %w", err) + return nil, fmt.Errorf("unable to dial to upstream: %w", err) } tlsConn := tls.Client(nonTlsConn, tlsConf) written, err := tlsConn.Write([]byte(authInfo.ConnectionNonce)) diff --git a/internal/daemon/worker/testing.go b/internal/daemon/worker/testing.go index d2711b7d9..865352b92 100644 --- a/internal/daemon/worker/testing.go +++ b/internal/daemon/worker/testing.go @@ -53,7 +53,7 @@ func (tw *TestWorker) Name() string { return tw.name } -func (tw *TestWorker) ControllerAddrs() []string { +func (tw *TestWorker) UpstreamAddrs() []string { var addrs []string lastStatus := tw.w.LastStatusSuccess() for _, v := range lastStatus.GetCalculatedUpstreams() { @@ -165,8 +165,8 @@ type TestWorkerOpts struct { // Config; if not provided a dev one will be created Config *config.Config - // Sets initial controller addresses - InitialControllers []string + // Sets initial upstream addresses + InitialUpstreams []string // If true, the worker will not be started DisableAutoStart bool @@ -233,8 +233,8 @@ func NewTestWorker(t testing.TB, opts *TestWorkerOpts) *TestWorker { opts.Config.Worker.Name = opts.Name } - if len(opts.InitialControllers) > 0 { - opts.Config.Worker.Controllers = opts.InitialControllers + if len(opts.InitialUpstreams) > 0 { + opts.Config.Worker.InitialUpstreams = opts.InitialUpstreams } // Start a logger @@ -325,7 +325,7 @@ func (tw *TestWorker) AddClusterWorkerMember(t testing.TB, opts *TestWorkerOpts) WorkerAuthKms: tw.w.conf.WorkerAuthKms, WorkerAuthStorageKms: tw.w.conf.WorkerAuthStorageKms, Name: opts.Name, - InitialControllers: tw.ControllerAddrs(), + InitialUpstreams: tw.UpstreamAddrs(), Logger: tw.w.conf.Logger, StatusGracePeriodDuration: opts.StatusGracePeriodDuration, } diff --git a/internal/tests/cluster/ipv6_listener_test.go b/internal/tests/cluster/ipv6_listener_test.go index 9de82f5d6..3cca5b714 100644 --- a/internal/tests/cluster/ipv6_listener_test.go +++ b/internal/tests/cluster/ipv6_listener_test.go @@ -45,10 +45,10 @@ func TestIPv6Listener(t *testing.T) { require.NoError(err) w1 := worker.NewTestWorker(t, &worker.TestWorkerOpts{ - Config: wconf, - WorkerAuthKms: c1.Config().WorkerAuthKms, - InitialControllers: c1.ClusterAddrs(), - Logger: logger.Named("w1"), + Config: wconf, + WorkerAuthKms: c1.Config().WorkerAuthKms, + InitialUpstreams: c1.ClusterAddrs(), + Logger: logger.Named("w1"), }) defer w1.Shutdown() diff --git a/internal/tests/cluster/multi_controller_worker_test.go b/internal/tests/cluster/multi_controller_worker_test.go index 98b040d63..df6c023b9 100644 --- a/internal/tests/cluster/multi_controller_worker_test.go +++ b/internal/tests/cluster/multi_controller_worker_test.go @@ -35,9 +35,9 @@ func TestMultiControllerMultiWorkerConnections(t *testing.T) { expectWorkers(t, c2) w1 := worker.NewTestWorker(t, &worker.TestWorkerOpts{ - WorkerAuthKms: c1.Config().WorkerAuthKms, - InitialControllers: c1.ClusterAddrs(), - Logger: logger.Named("w1"), + WorkerAuthKms: c1.Config().WorkerAuthKms, + InitialUpstreams: c1.ClusterAddrs(), + Logger: logger.Named("w1"), }) defer w1.Shutdown() @@ -60,9 +60,9 @@ func TestMultiControllerMultiWorkerConnections(t *testing.T) { expectWorkers(t, c2, w2) w1 = worker.NewTestWorker(t, &worker.TestWorkerOpts{ - WorkerAuthKms: c1.Config().WorkerAuthKms, - InitialControllers: c1.ClusterAddrs(), - Logger: logger.Named("w1"), + WorkerAuthKms: c1.Config().WorkerAuthKms, + InitialUpstreams: c1.ClusterAddrs(), + Logger: logger.Named("w1"), }) defer w1.Shutdown() diff --git a/internal/tests/cluster/session_cleanup_test.go b/internal/tests/cluster/session_cleanup_test.go index 150f474ce..ad6739430 100644 --- a/internal/tests/cluster/session_cleanup_test.go +++ b/internal/tests/cluster/session_cleanup_test.go @@ -109,7 +109,7 @@ func testWorkerSessionCleanupSingle(burdenCase timeoutBurdenType) func(t *testin w1 := worker.NewTestWorker(t, &worker.TestWorkerOpts{ WorkerAuthKms: c1.Config().WorkerAuthKms, - InitialControllers: []string{proxy.ListenerAddr()}, + InitialUpstreams: []string{proxy.ListenerAddr()}, Logger: logger.Named("w1"), StatusGracePeriodDuration: workerGracePeriod(burdenCase), }) @@ -269,7 +269,7 @@ func testWorkerSessionCleanupMulti(burdenCase timeoutBurdenType) func(t *testing // ************ w1 := worker.NewTestWorker(t, &worker.TestWorkerOpts{ WorkerAuthKms: c1.Config().WorkerAuthKms, - InitialControllers: []string{p1.ListenerAddr(), p2.ListenerAddr()}, + InitialUpstreams: []string{p1.ListenerAddr(), p2.ListenerAddr()}, Logger: logger.Named("w1"), StatusGracePeriodDuration: workerGracePeriod(burdenCase), }) diff --git a/internal/tests/cluster/unix_listener_test.go b/internal/tests/cluster/unix_listener_test.go index 9fa104bf4..97a0b9f80 100644 --- a/internal/tests/cluster/unix_listener_test.go +++ b/internal/tests/cluster/unix_listener_test.go @@ -58,10 +58,10 @@ func TestUnixListener(t *testing.T) { require.NoError(err) w1 := worker.NewTestWorker(t, &worker.TestWorkerOpts{ - Config: wconf, - WorkerAuthKms: c1.Config().WorkerAuthKms, - InitialControllers: c1.ClusterAddrs(), - Logger: logger.Named("w1"), + Config: wconf, + WorkerAuthKms: c1.Config().WorkerAuthKms, + InitialUpstreams: c1.ClusterAddrs(), + Logger: logger.Named("w1"), }) defer w1.Shutdown() diff --git a/internal/tests/cluster/worker_replay_test.go b/internal/tests/cluster/worker_replay_test.go index ff2458e80..a5a1fd2c6 100644 --- a/internal/tests/cluster/worker_replay_test.go +++ b/internal/tests/cluster/worker_replay_test.go @@ -37,9 +37,9 @@ func TestWorkerReplay(t *testing.T) { conf.Eventing = &ec.EventerConfig require.NoError(t, err) w1 := worker.NewTestWorker(t, &worker.TestWorkerOpts{ - Config: conf, - WorkerAuthKms: c1.Config().WorkerAuthKms, - InitialControllers: c1.ClusterAddrs(), + Config: conf, + WorkerAuthKms: c1.Config().WorkerAuthKms, + InitialUpstreams: c1.ClusterAddrs(), NonceFn: func(length int) (string, error) { return "test_noncetest_nonce", nil }, @@ -63,9 +63,9 @@ func TestWorkerReplay(t *testing.T) { // Now, start up again with the same nonce w1 = worker.NewTestWorker(t, &worker.TestWorkerOpts{ - Config: conf, - WorkerAuthKms: c1.Config().WorkerAuthKms, - InitialControllers: c1.ClusterAddrs(), + Config: conf, + WorkerAuthKms: c1.Config().WorkerAuthKms, + InitialUpstreams: c1.ClusterAddrs(), NonceFn: func(length int) (string, error) { return "test_noncetest_nonce", nil }, diff --git a/internal/tests/cluster/worker_tagging_test.go b/internal/tests/cluster/worker_tagging_test.go index 7c6132810..540f1d2f0 100644 --- a/internal/tests/cluster/worker_tagging_test.go +++ b/internal/tests/cluster/worker_tagging_test.go @@ -53,10 +53,10 @@ func TestWorkerTagging(t *testing.T) { "foo": {"bar"}, } w1 := worker.NewTestWorker(t, &worker.TestWorkerOpts{ - Config: conf, - WorkerAuthKms: c1.Config().WorkerAuthKms, - InitialControllers: c1.ClusterAddrs(), - Logger: logger.Named("w1"), + Config: conf, + WorkerAuthKms: c1.Config().WorkerAuthKms, + InitialUpstreams: c1.ClusterAddrs(), + Logger: logger.Named("w1"), }) defer w1.Shutdown() w1Addr := w1.ProxyAddrs()[0] @@ -70,10 +70,10 @@ func TestWorkerTagging(t *testing.T) { "az": {"one", "two", "three"}, } w2 := worker.NewTestWorker(t, &worker.TestWorkerOpts{ - Config: conf, - WorkerAuthKms: c1.Config().WorkerAuthKms, - InitialControllers: c1.ClusterAddrs(), - Logger: logger.Named("w2"), + Config: conf, + WorkerAuthKms: c1.Config().WorkerAuthKms, + InitialUpstreams: c1.ClusterAddrs(), + Logger: logger.Named("w2"), }) defer w2.Shutdown() w2Addr := w2.ProxyAddrs()[0] @@ -87,10 +87,10 @@ func TestWorkerTagging(t *testing.T) { "az": {"one", "three"}, } w3 := worker.NewTestWorker(t, &worker.TestWorkerOpts{ - Config: conf, - WorkerAuthKms: c1.Config().WorkerAuthKms, - InitialControllers: c1.ClusterAddrs(), - Logger: logger.Named("w3"), + Config: conf, + WorkerAuthKms: c1.Config().WorkerAuthKms, + InitialUpstreams: c1.ClusterAddrs(), + Logger: logger.Named("w3"), }) defer w3.Shutdown() w3Addr := w3.ProxyAddrs()[0] diff --git a/internal/tests/cluster/x509_verification_test.go b/internal/tests/cluster/x509_verification_test.go index bf6aeca78..2aa142e52 100644 --- a/internal/tests/cluster/x509_verification_test.go +++ b/internal/tests/cluster/x509_verification_test.go @@ -49,9 +49,9 @@ func TestCustomX509Verification_Client(t *testing.T) { conf.Eventing = &ec.EventerConfig req.NoError(err) w1 := worker.NewTestWorker(t, &worker.TestWorkerOpts{ - WorkerAuthKms: c1.Config().WorkerAuthKms, - InitialControllers: c1.ClusterAddrs(), - Logger: logger.Named("w1"), + WorkerAuthKms: c1.Config().WorkerAuthKms, + InitialUpstreams: c1.ClusterAddrs(), + Logger: logger.Named("w1"), }) t.Cleanup(w1.Shutdown) @@ -188,8 +188,8 @@ func testCustomX509Verification_Server(ec event.TestConfig, certPool *x509.CertP conf.Eventing = &ec.EventerConfig req.NoError(err) w1 := worker.NewTestWorker(t, &worker.TestWorkerOpts{ - WorkerAuthKms: c1.Config().WorkerAuthKms, - InitialControllers: c1.ClusterAddrs(), + WorkerAuthKms: c1.Config().WorkerAuthKms, + InitialUpstreams: c1.ClusterAddrs(), }) w1.Worker().TestOverrideX509VerifyCertPool = certPool w1.Worker().TestOverrideX509VerifyDnsName = dnsName diff --git a/website/content/docs/configuration/worker.mdx b/website/content/docs/configuration/worker.mdx index 72e1be5a7..7a4a62c5c 100644 --- a/website/content/docs/configuration/worker.mdx +++ b/website/content/docs/configuration/worker.mdx @@ -36,8 +36,8 @@ worker { an env var (env://) from which the address will be read; or a [go-sockaddr template](https://godoc.org/github.com/hashicorp/go-sockaddr/template). -- `controllers` - A list of hosts/IP addresses and optionally ports for reaching - controllers. The port will default to :9201 if not specified. This value can be +- `initial_upstreams` - A list of hosts/IP addresses and optionally ports for reaching + controllers or workers. The port will default to :9201 if not specified. This value can be a direct access string array with the addresses, or it can refer to a file on disk (file://) from which the addresses will be read, or an env var (env://) from which the addresses will be read. When using env or file, their contents