CyberCoveLLC/threshold
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(worker): add 'initial_upstreams' field to worker config to prepare to replace 'controllers' field. Update worker config description on website.

Browse Source
This commit is contained in:
Haotian authored and Haotian Zeng committed 2022-06-13 15:45:03 -07:00
1 parent 4b526538af
commit 40ba275759
18 files changed
+208 -150
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

internal/cmd/commands/dev/dev.go
+1 -1
View File
@@ -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"
internal/cmd/commands/server/server.go
+15 -12
View File
@@ -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
}
}
internal/cmd/commands/server/worker_shutdown_reload_test.go
+1 -1
View File
@@ -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" {
internal/cmd/commands/server/worker_tags_reload_test.go
+2 -2
View File
@@ -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"]
}
internal/cmd/config/config.go
+37 -16
View File
@@ -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)
internal/cmd/config/config_test.go
+94 -60
View File
@@ -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)
})
}
}
internal/credential/vault/setup_manual_test.go
+4 -4
View File
@@ -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]
internal/daemon/worker/auth_rotation_test.go
+1 -1
View File
@@ -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,
})
internal/daemon/worker/controller_connection.go
+6 -6
View File
@@ -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))
internal/daemon/worker/testing.go
+6 -6
View File
@@ -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,
}
internal/tests/cluster/ipv6_listener_test.go
+4 -4
View File
@@ -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()
internal/tests/cluster/multi_controller_worker_test.go
+6 -6
View File
@@ -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()
internal/tests/cluster/session_cleanup_test.go
+2 -2
View File
@@ -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),
})
internal/tests/cluster/unix_listener_test.go
+4 -4
View File
@@ -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()
internal/tests/cluster/worker_replay_test.go
+6 -6
View File
@@ -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
},
internal/tests/cluster/worker_tagging_test.go
+12 -12
View File
@@ -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]
internal/tests/cluster/x509_verification_test.go
+5 -5
View File
@@ -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
website/content/docs/configuration/worker.mdx
+2 -2
View File
@@ -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