diff --git a/coderd/apidoc/docs.go b/coderd/apidoc/docs.go index e5839bd177..a76cce2536 100644 --- a/coderd/apidoc/docs.go +++ b/coderd/apidoc/docs.go @@ -17838,6 +17838,14 @@ const docTemplate = `{ "ChatWatchEventKindContextDirty" ] }, + "codersdk.ClusterConfig": { + "type": "object", + "properties": { + "host": { + "type": "string" + } + } + }, "codersdk.ConnectionLatency": { "type": "object", "properties": { @@ -19174,6 +19182,9 @@ const docTemplate = `{ "cli_upgrade_message": { "type": "string" }, + "cluster": { + "$ref": "#/definitions/codersdk.ClusterConfig" + }, "config": { "type": "string" }, diff --git a/coderd/apidoc/swagger.json b/coderd/apidoc/swagger.json index 9b41a6ebc0..135e16c67b 100644 --- a/coderd/apidoc/swagger.json +++ b/coderd/apidoc/swagger.json @@ -16091,6 +16091,14 @@ "ChatWatchEventKindContextDirty" ] }, + "codersdk.ClusterConfig": { + "type": "object", + "properties": { + "host": { + "type": "string" + } + } + }, "codersdk.ConnectionLatency": { "type": "object", "properties": { @@ -17376,6 +17384,9 @@ "cli_upgrade_message": { "type": "string" }, + "cluster": { + "$ref": "#/definitions/codersdk.ClusterConfig" + }, "config": { "type": "string" }, diff --git a/coderd/x/nats/cluster.go b/coderd/x/nats/cluster.go index aa12c748fe..d70de4dc3a 100644 --- a/coderd/x/nats/cluster.go +++ b/coderd/x/nats/cluster.go @@ -17,12 +17,15 @@ const defaultClusterTokenUsername = "coder" // PeerFetcher fetches NATS peer route addresses. type PeerFetcher interface { - PrimaryPeerAddresses() []string + FetchNATSPeers() []string + SetSelfNATSPort(port int32) } type NopPeerFetcher struct{} -func (NopPeerFetcher) PrimaryPeerAddresses() []string { +func (NopPeerFetcher) SetSelfNATSPort(int32) {} + +func (NopPeerFetcher) FetchNATSPeers() []string { return nil } @@ -35,6 +38,14 @@ func (p *Pubsub) SetPeerFetcher(fetcher PeerFetcher) { } p.peerFetcher = fetcher p.mu.Unlock() + if ca := p.Server.ClusterAddr(); ca != nil { + if ca.Port >= 1 && ca.Port <= 65535 { + //nolint:gosec // range checked above so conversion is safe. + fetcher.SetSelfNATSPort(int32(ca.Port)) + } else { + p.logger.Warn(p.ctx, "unexpected NATS cluster port", slog.F("port", ca.Port)) + } + } p.RefreshPeers() } @@ -53,7 +64,7 @@ func (p *Pubsub) runPeerRefresh() { fetcher := p.peerFetcher p.mu.Unlock() - addrs := fetcher.PrimaryPeerAddresses() + addrs := fetcher.FetchNATSPeers() if err := p.setPeerAddresses(addrs); err != nil { if errors.Is(err, errClosed) && p.ctx.Err() != nil { return @@ -81,7 +92,7 @@ func (p *Pubsub) setPeerAddresses(addresses []string) error { return xerrors.New("nats pubsub was not started with clustering enabled") } - routes, err := p.parsePeerAddresses(addresses) + routes, err := parsePeerAddresses(addresses) if err != nil { return err } @@ -109,7 +120,7 @@ func (p *Pubsub) setPeerAddresses(addresses []string) error { return nil } -func (p *Pubsub) parsePeerAddresses(addresses []string) ([]*url.URL, error) { +func parsePeerAddresses(addresses []string) ([]*url.URL, error) { routesByAddress := make(map[string]*url.URL, len(addresses)) for i, address := range addresses { trimmed := strings.TrimSpace(address) @@ -122,16 +133,6 @@ func (p *Pubsub) parsePeerAddresses(addresses []string) ([]*url.URL, error) { return nil, err } - // This is a hack to enable testing with an arbitrary port. The logic here - // is to presume if the default port is being used then we are running in prod - // and all peers are using the same port. If the port is not the default then - // we are running a test in which case we should pass through the custom port. - // This hack will be removed when https://github.com/coder/scaletest/issues/149 - // is resolved. - if p.opts.ClusterPort == defaultClusterPort { - port = defaultClusterPort - } - hostPort := net.JoinHostPort(host, strconv.Itoa(port)) routesByAddress[hostPort] = &url.URL{ Scheme: "nats", @@ -168,6 +169,9 @@ func normalizeHostPort(address string) (string, int, error) { if route.Path != "" || route.RawQuery != "" || route.Fragment != "" { return "", 0, xerrors.Errorf("peer address %q must not include path, query, or fragment", address) } + if route.Scheme != "nats" { + return "", 0, xerrors.Errorf("peer address %q must use nats scheme", address) + } host, port, err := net.SplitHostPort(route.Host) if err != nil { diff --git a/coderd/x/nats/cluster_internal_test.go b/coderd/x/nats/cluster_internal_test.go index e3dc10c29f..174ebfd29d 100644 --- a/coderd/x/nats/cluster_internal_test.go +++ b/coderd/x/nats/cluster_internal_test.go @@ -10,15 +10,19 @@ import ( "github.com/coder/coder/v2/testutil" ) +const ( + minTCPPort int32 = 1 + maxTCPPort int32 = 65535 +) + func Test_parsePeerAddresses(t *testing.T) { t.Parallel() t.Run("Valid", func(t *testing.T) { t.Parallel() - ps := &Pubsub{} - routes, err := ps.parsePeerAddresses([]string{ - "whatever://127.0.0.1:4222 ", - "http://[::1]:7222", + routes, err := parsePeerAddresses([]string{ + "nats://127.0.0.1:4222 ", + "nats://[::1]:7222", "nats://example.com:6222", }) require.NoError(t, err) @@ -29,51 +33,16 @@ func Test_parsePeerAddresses(t *testing.T) { }, routeStrings(routes)) }) - // Test that when a pubsub is running with the default port, it assumes all peers are also using - // the default port. - t.Run("PrefersDefaultPort", func(t *testing.T) { - t.Parallel() - ps := &Pubsub{} - ps.opts.ClusterPort = defaultClusterPort - routes, err := ps.parsePeerAddresses([]string{ - "whatever://127.0.0.1:4222 ", - "http://[::1]:7222", - "nats://example.com:1234", - }) - require.NoError(t, err) - require.ElementsMatch(t, []string{ - "nats://127.0.0.1:6222", - "nats://[::1]:6222", - "nats://example.com:6222", - }, routeStrings(routes)) - }) - - // Regression: in production the relay URL host carries the coderd HTTP - // port (e.g. 8080), and routes must be rewritten to the NATS cluster - // port. This only works because New defaults ClusterPort to - // defaultClusterPort; if it were left at the zero value the rewrite - // would be skipped and routes would dial the HTTP port. - t.Run("RewritesRelayHTTPPort", func(t *testing.T) { - t.Parallel() - ps := &Pubsub{} - ps.opts.ClusterPort = defaultClusterPort - routes, err := ps.parsePeerAddresses([]string{"http://10.0.0.7:8080"}) - require.NoError(t, err) - require.Equal(t, []string{"nats://10.0.0.7:6222"}, routeStrings(routes)) - }) - t.Run("Empty", func(t *testing.T) { t.Parallel() - ps := &Pubsub{} - routes, err := ps.parsePeerAddresses(nil) + routes, err := parsePeerAddresses(nil) require.NoError(t, err) require.Empty(t, routes) }) t.Run("Dedupes", func(t *testing.T) { t.Parallel() - ps := &Pubsub{} - routes, err := ps.parsePeerAddresses([]string{ + routes, err := parsePeerAddresses([]string{ "nats://b.example:6222", "nats://a.example:6222", "nats://b.example:6222", @@ -103,11 +72,12 @@ func Test_parsePeerAddresses(t *testing.T) { "nats://127.0.0.1:4222/path", "nats://127.0.0.1:4222?x=1", "nats://127.0.0.1:4222#frag", + "whatever://127.0.0.1:4222 ", + "http://[::1]:7222", } { t.Run(address, func(t *testing.T) { t.Parallel() - ps := &Pubsub{} - _, err := ps.parsePeerAddresses([]string{address}) + _, err := parsePeerAddresses([]string{address}) require.Error(t, err) }) } @@ -117,10 +87,9 @@ func Test_parsePeerAddresses(t *testing.T) { func Test_filterSelfRoutes(t *testing.T) { t.Parallel() - ps := &Pubsub{} - routes, err := ps.parsePeerAddresses([]string{ + routes, err := parsePeerAddresses([]string{ "nats://b.example:6222", - "http://self.example:6222", + "nats://self.example:6222", }) require.NoError(t, err) @@ -141,6 +110,8 @@ func TestPubsub_RefreshPeers(t *testing.T) { opts := clusterTestOptions(t) opts.PeerFetcher = fetcher a := newTestPubsub(t, opts) + require.GreaterOrEqual(t, fetcher.port, minTCPPort) + require.LessOrEqual(t, fetcher.port, maxTCPPort) require.Eventually(t, func() bool { routes := currentRouteURLs(a) @@ -159,11 +130,13 @@ func TestPubsub_RefreshPeers(t *testing.T) { "nats://127.0.0.1:1234", "nats://127.0.0.1:1235", } - fetcher := &testPeerFetcher{routes} + fetcher := &testPeerFetcher{addresses: routes} expectedRoutes := routesWithAuth(mustParsePeerAddresses(t, fetcher.addresses...), opts.ClusterAuthToken) a.SetPeerFetcher(fetcher) + require.GreaterOrEqual(t, fetcher.port, minTCPPort) + require.LessOrEqual(t, fetcher.port, maxTCPPort) require.Eventually(t, func() bool { return sortedURLsEqual(currentRouteURLs(a), sortRouteURLs(expectedRoutes)) }, testutil.WaitShort, testutil.IntervalFast) @@ -194,26 +167,17 @@ func currentRouteURLs(ps *Pubsub) []*url.URL { type testPeerFetcher struct { addresses []string + port int32 } -func (f *testPeerFetcher) PrimaryPeerAddresses() []string { +func (f *testPeerFetcher) SetSelfNATSPort(port int32) { + f.port = port +} + +func (f *testPeerFetcher) FetchNATSPeers() []string { return f.addresses } -// TestPubsub_New_DefaultsClusterPort guards the production wiring: New -// must persist the default cluster port onto opts so the peer route -// rewrite in parsePeerAddresses recognizes prod and forces routes to the -// NATS port. The cli constructs Options without a ClusterPort, so leaving -// it at the zero value made every replica dial peers at the relay URL's -// HTTP port instead of the NATS route port. -func TestPubsub_New_DefaultsClusterPort(t *testing.T) { - t.Parallel() - // defaultTestOptions disables clustering (no fixed-port listener to - // collide with parallel tests) and leaves ClusterPort unset. - ps := newTestPubsub(t, defaultTestOptions()) - require.Equal(t, defaultClusterPort, ps.opts.ClusterPort) -} - func TestPubsub_setPeerAddresses(t *testing.T) { t.Parallel() t.Run("OK", func(t *testing.T) { diff --git a/coderd/x/nats/natsbench/topology.go b/coderd/x/nats/natsbench/topology.go index f44744743c..0fb7308887 100644 --- a/coderd/x/nats/natsbench/topology.go +++ b/coderd/x/nats/natsbench/topology.go @@ -34,9 +34,11 @@ type staticPeerFetcher struct { addrs []string } +func (*staticPeerFetcher) SetSelfNATSPort(int32) {} + var _ nats.PeerFetcher = (*staticPeerFetcher)(nil) -func (f *staticPeerFetcher) PrimaryPeerAddresses() []string { +func (f *staticPeerFetcher) FetchNATSPeers() []string { f.mu.Lock() defer f.mu.Unlock() return slices.Clone(f.addrs) diff --git a/coderd/x/nats/pubsub.go b/coderd/x/nats/pubsub.go index 0f00f60056..57ced2aeb3 100644 --- a/coderd/x/nats/pubsub.go +++ b/coderd/x/nats/pubsub.go @@ -113,7 +113,8 @@ type Options struct { ClusterHost string // ClusterPort is the embedded NATS route listener port. Zero means - // 6222 when cluster mode is enabled. + // 6222 when cluster mode is enabled. NATS `server.RANDOM_PORT` can be + // used to select a random port. ClusterPort int // ClusterAuthToken is the shared route authentication token for @@ -297,18 +298,7 @@ func (p *Pubsub) buildConnHandlers() connHandlers { // New creates an embedded NATS Pubsub. The returned *Pubsub owns the // embedded server and the publisher and subscriber connection pools. // Close shuts down all owned resources. -func New(ctx context.Context, logger slog.Logger, opts Options) (*Pubsub, error) { - // Persist the default cluster port onto opts so it is the same value the - // listener (buildServerOptions) binds and the value parsePeerAddresses - // compares against. parsePeerAddresses overwrites each peer's parsed port - // with defaultClusterPort, but only when opts.ClusterPort already equals - // defaultClusterPort. Callers like the cli leave ClusterPort at 0, so - // without this that branch is skipped and peers are dialed on the relay - // URL's port (e.g. 8080) instead of the NATS route port (6222). - if opts.ClusterPort == 0 { - opts.ClusterPort = defaultClusterPort - } - +func New(ctx context.Context, logger slog.Logger, opts Options) (pubSub *Pubsub, retErr error) { sopts, err := buildServerOptions(opts) if err != nil { return nil, err @@ -318,6 +308,12 @@ func New(ctx context.Context, logger slog.Logger, opts Options) (*Pubsub, error) if err != nil { return nil, err } + defer func() { + if retErr != nil { + ns.Shutdown() + ns.WaitForShutdown() + } + }() logger.Info(context.Background(), "embedded nats server started", slog.F("client_url", ns.ClientURL()), @@ -328,6 +324,11 @@ func New(ctx context.Context, logger slog.Logger, opts Options) (*Pubsub, error) } p := newPubsub(ctx, logger, opts) + defer func() { + if retErr != nil { + p.cancel() + } + }() p.Server = ns p.clustered = !opts.disableCluster p.serverOpts = sopts.Clone() @@ -336,29 +337,43 @@ func New(ctx context.Context, logger slog.Logger, opts Options) (*Pubsub, error) publishPool, err := newConnPool(ns, opts, handlers, opts.PublishConns, "coder-pubsub-pub") if err != nil { - p.cancel() - ns.Shutdown() - ns.WaitForShutdown() return nil, err } + defer func() { + if retErr != nil { + for _, c := range publishPool { + c.Close() + } + } + }() + p.publishPool = publishPool subscribePool, err := newConnPool(ns, opts, handlers, opts.SubscribeConns, "coder-pubsub-sub") if err != nil { - p.cancel() - for _, c := range publishPool { - c.Close() - } - ns.Shutdown() - ns.WaitForShutdown() return nil, err } - - p.publishPool = publishPool + defer func() { + if retErr != nil { + for _, c := range subscribePool { + c.Close() + } + } + }() p.subscribePool = subscribePool // All owned connections dialed successfully above. p.metrics.markConnected(len(publishPool) + len(subscribePool)) if p.clustered { + ca := ns.ClusterAddr() + if ca == nil { + return nil, xerrors.New("no cluster address") + } + // sec checks, just to be sure + if ca.Port < 0 || ca.Port > 65535 { + return nil, xerrors.Errorf("invalid cluster port: %d", ca.Port) + } + //nolint:gosec // range checked above so conversion is safe. + opts.PeerFetcher.SetSelfNATSPort(int32(ca.Port)) go p.runPeerRefresh() } go func() { diff --git a/codersdk/deployment.go b/codersdk/deployment.go index a0447cff9c..dd4e22a1d4 100644 --- a/codersdk/deployment.go +++ b/codersdk/deployment.go @@ -623,6 +623,7 @@ type DeploymentValues struct { HTTPAddress serpent.String `json:"http_address,omitempty" typescript:",notnull"` AutobuildPollInterval serpent.Duration `json:"autobuild_poll_interval,omitempty"` JobReaperDetectorInterval serpent.Duration `json:"job_hang_detector_interval,omitempty"` + Cluster ClusterConfig `json:"cluster,omitempty" typescript:",notnull"` DERP DERP `json:"derp,omitempty" typescript:",notnull"` Prometheus PrometheusConfig `json:"prometheus,omitempty" typescript:",notnull"` Pprof PprofConfig `json:"pprof,omitempty" typescript:",notnull"` @@ -882,6 +883,10 @@ type DERPConfig struct { Path serpent.String `json:"path" typescript:",notnull"` } +type ClusterConfig struct { + Host serpent.String `json:"host" typescript:",notnull"` +} + type UsageStatsConfig struct { Enable serpent.Bool `json:"enable" typescript:",notnull"` } diff --git a/docs/reference/api/general.md b/docs/reference/api/general.md index 4ae0f8d73b..7f964f4894 100644 --- a/docs/reference/api/general.md +++ b/docs/reference/api/general.md @@ -233,6 +233,9 @@ curl -X GET http://coder-server:8080/api/v2/deployment/config \ "browser_only": true, "cache_directory": "string", "cli_upgrade_message": "string", + "cluster": { + "host": "string" + }, "config": "string", "config_ssh": { "deploymentName": "string", diff --git a/docs/reference/api/schemas.md b/docs/reference/api/schemas.md index af90fe3da3..879f5afadf 100644 --- a/docs/reference/api/schemas.md +++ b/docs/reference/api/schemas.md @@ -3987,6 +3987,20 @@ AuthorizationObject can represent a "set" of objects, such as: all workspaces in |-----------------------------------------------------------------------------------------------------------------------------------| | `action_required`, `context_dirty`, `created`, `deleted`, `diff_status_change`, `status_change`, `summary_change`, `title_change` | +## codersdk.ClusterConfig + +```json +{ + "host": "string" +} +``` + +### Properties + +| Name | Type | Required | Restrictions | Description | +|--------|--------|----------|--------------|-------------| +| `host` | string | false | | | + ## codersdk.ConnectionLatency ```json @@ -5578,6 +5592,9 @@ CreateWorkspaceRequest provides options for creating a new workspace. Only one o "browser_only": true, "cache_directory": "string", "cli_upgrade_message": "string", + "cluster": { + "host": "string" + }, "config": "string", "config_ssh": { "deploymentName": "string", @@ -6180,6 +6197,9 @@ CreateWorkspaceRequest provides options for creating a new workspace. Only one o "browser_only": true, "cache_directory": "string", "cli_upgrade_message": "string", + "cluster": { + "host": "string" + }, "config": "string", "config_ssh": { "deploymentName": "string", @@ -6607,6 +6627,7 @@ CreateWorkspaceRequest provides options for creating a new workspace. Only one o | `browser_only` | boolean | false | | | | `cache_directory` | string | false | | | | `cli_upgrade_message` | string | false | | | +| `cluster` | [codersdk.ClusterConfig](#codersdkclusterconfig) | false | | | | `config` | string | false | | | | `config_ssh` | [codersdk.SSHConfig](#codersdksshconfig) | false | | | | `dangerous` | [codersdk.DangerousConfig](#codersdkdangerousconfig) | false | | | diff --git a/enterprise/cli/server.go b/enterprise/cli/server.go index 37febd028b..22b3ddbf18 100644 --- a/enterprise/cli/server.go +++ b/enterprise/cli/server.go @@ -32,18 +32,28 @@ import ( func (r *RootCmd) Server(_ func()) *serpent.Command { cmd := r.RootCmd.Server(func(ctx context.Context, options *agplcoderd.Options) (*agplcoderd.API, io.Closer, error) { + var ( + derpURL *url.URL + err error + ) if options.DeploymentValues.DERP.Server.RelayURL.String() != "" { - _, err := url.Parse(options.DeploymentValues.DERP.Server.RelayURL.String()) + derpURL, err = url.Parse(options.DeploymentValues.DERP.Server.RelayURL.String()) if err != nil { return nil, nil, xerrors.Errorf("derp-server-relay-address must be a valid HTTP URL: %w", err) } } + clusterHost := options.DeploymentValues.Cluster.Host.String() + if clusterHost == "" && derpURL != nil { + // Use the DERP host if the operator didn't specify an explicit cluster host, since this is an older setting + // and more likely to be configured by longtime HA customers. + clusterHost = derpURL.Hostname() + } // Always generate a mesh key, even if the built-in DERP server is // disabled. This mesh key is still used by workspace proxies running // HA. var meshKey string - err := options.Database.InTx(func(tx database.Store) error { + err = options.Database.InTx(func(tx database.Store) error { // This will block until the lock is acquired, and will be // automatically released when the transaction ends. err := tx.AcquireLock(ctx, database.LockIDEnterpriseDeploymentSetup) @@ -97,6 +107,7 @@ func (r *RootCmd) Server(_ func()) *serpent.Command { SCIMAPIKey: []byte(options.DeploymentValues.SCIMAPIKey.Value()), UseLegacySCIM: options.DeploymentValues.UseLegacySCIM.Value(), RBAC: true, + ClusterHost: clusterHost, DERPServerRelayAddress: options.DeploymentValues.DERP.Server.RelayURL.String(), DERPServerRegionID: int(options.DeploymentValues.DERP.Server.RegionID.Value()), ProxyHealthInterval: options.DeploymentValues.ProxyHealthStatusInterval.Value(), diff --git a/enterprise/coderd/coderd.go b/enterprise/coderd/coderd.go index 1c68c556b2..ac5b357c29 100644 --- a/enterprise/coderd/coderd.go +++ b/enterprise/coderd/coderd.go @@ -678,7 +678,8 @@ func New(ctx context.Context, options *Options) (_ *API, err error) { } // We always want to run the replica manager even if we don't have DERP - // enabled, since it's used to detect other coder servers for licensing. + // enabled, since it's used to detect other coder servers for licensing, + // and NATS clustering for HA pubsub. api.replicaManager, err = replicasync.New(ctx, options.Logger, options.Database, options.ReplicaSyncPubsub, &replicasync.Options{ ID: api.AGPL.ID, RelayAddress: options.DERPServerRelayAddress, @@ -686,6 +687,7 @@ func New(ctx context.Context, options *Options) (_ *API, err error) { RegionID: int32(options.DERPServerRegionID), TLSConfig: meshTLSConfig, UpdateInterval: options.ReplicaSyncUpdateInterval, + ClusterHost: options.ClusterHost, }) if err != nil { return nil, xerrors.Errorf("initialize replica: %w", err) @@ -788,6 +790,7 @@ type Options struct { // Used for high availability. ReplicaSyncUpdateInterval time.Duration ReplicaErrorGracePeriod time.Duration + ClusterHost string // IP or hostname to reach this specific replica DERPServerRelayAddress string DERPServerRegionID int diff --git a/enterprise/coderd/coderd_test.go b/enterprise/coderd/coderd_test.go index d0956057fc..32292b211a 100644 --- a/enterprise/coderd/coderd_test.go +++ b/enterprise/coderd/coderd_test.go @@ -631,7 +631,7 @@ func TestMultiReplica_NATSPubsubPeers(t *testing.T) { t.Parallel() ctx := testutil.Context(t, testutil.WaitLong) - logger := slogtest.Make(t, &slogtest.Options{IgnoreErrors: true}) + logger := slogtest.Make(t, &slogtest.Options{IgnoreErrors: true}).Leveled(slog.LevelDebug) db, pgPubsub := dbtestutil.NewDB(t) clusterToken := "shared-token" @@ -671,13 +671,19 @@ func TestMultiReplica_NATSPubsubPeers(t *testing.T) { t.Cleanup(func() { _ = natsB.Close() }) mgr, err := replicasync.New(ctx, logger.Named("replica-b"), db, pgPubsub, &replicasync.Options{ - ID: uuid.New(), - RelayAddress: fmt.Sprintf("nats://127.0.0.1:%d", natsB.Server.ClusterAddr().Port), + ID: uuid.New(), + // port doesn't matter because we don't have an API up, but replicasync will refuse peers that don't set + // RelayAddress at all. + RelayAddress: "https://127.0.0.1", + ClusterHost: "127.0.0.1", RegionID: 12345, UpdateInterval: testutil.IntervalFast, }) require.NoError(t, err) t.Cleanup(func() { _ = mgr.Close() }) + require.NotNil(t, natsB.Server.ClusterAddr()) + // nolint: gosec // nats listens on TCP ports + mgr.SetSelfNATSPort(int32(natsB.Server.ClusterAddr().Port)) subject := "nats.replica" messages := make(chan []byte, 1) diff --git a/enterprise/coderd/workspaceproxy.go b/enterprise/coderd/workspaceproxy.go index 530aa89e09..c70d02e9cf 100644 --- a/enterprise/coderd/workspaceproxy.go +++ b/enterprise/coderd/workspaceproxy.go @@ -667,8 +667,8 @@ func (api *API) workspaceProxyRegister(rw http.ResponseWriter, r *http.Request) Error: req.ReplicaError, DatabaseLatency: 0, Primary: false, - ClusterHost: "", // TODO - NATSPort: 0, // TODO + ClusterHost: "", + NATSPort: 0, // proxies do not run NATS }) if err != nil { return xerrors.Errorf("update replica: %w", err) @@ -686,8 +686,8 @@ func (api *API) workspaceProxyRegister(rw http.ResponseWriter, r *http.Request) Version: req.Version, DatabaseLatency: 0, Primary: false, - ClusterHost: "", // TODO - NATSPort: 0, // TODO + ClusterHost: "", + NATSPort: 0, // proxies do not run NATS }) if err != nil { return xerrors.Errorf("insert replica: %w", err) @@ -830,8 +830,8 @@ func (api *API) workspaceProxyDeregister(rw http.ResponseWriter, r *http.Request Error: replica.Error, DatabaseLatency: replica.DatabaseLatency, Primary: replica.Primary, - ClusterHost: "", // TODO - NATSPort: 0, // TODO + ClusterHost: "", + NATSPort: 0, // proxies do not run NATS }) if err != nil { return xerrors.Errorf("update replica: %w", err) diff --git a/enterprise/replicasync/replicasync.go b/enterprise/replicasync/replicasync.go index 7ea6dd5818..f1acececfe 100644 --- a/enterprise/replicasync/replicasync.go +++ b/enterprise/replicasync/replicasync.go @@ -36,6 +36,7 @@ type Options struct { RelayAddress string RegionID int32 TLSConfig *tls.Config + ClusterHost string } // New registers the replica with the database and periodically updates to @@ -77,8 +78,8 @@ func New(ctx context.Context, logger slog.Logger, db database.Store, ps pubsub.P // #nosec G115 - Safe conversion for microseconds latency which is expected to be within int32 range DatabaseLatency: int32(databaseLatency.Microseconds()), Primary: true, - ClusterHost: "", // TODO - NATSPort: 0, // TODO + ClusterHost: options.ClusterHost, + NATSPort: 0, // set later via SetSelfNATSPort }) if err != nil { return nil, xerrors.Errorf("insert replica: %w", err) @@ -329,8 +330,8 @@ func (m *Manager) syncReplicas(ctx context.Context) error { // #nosec G115 - Safe conversion for microseconds latency which is expected to be within int32 range DatabaseLatency: int32(databaseLatency.Microseconds()), Primary: m.self.Primary, - ClusterHost: "", // TODO - NATSPort: 0, // TODO + ClusterHost: m.self.ClusterHost, + NATSPort: m.self.NATSPort, }) if err != nil { if !errors.Is(err, sql.ErrNoRows) { @@ -350,8 +351,8 @@ func (m *Manager) syncReplicas(ctx context.Context) error { // #nosec G115 - Safe conversion for microseconds latency which is expected to be within int32 range DatabaseLatency: int32(databaseLatency.Microseconds()), Primary: m.self.Primary, - ClusterHost: "", // TODO - NATSPort: 0, // TODO + ClusterHost: m.self.ClusterHost, + NATSPort: m.self.NATSPort, }) if err != nil { return xerrors.Errorf("update replica: %w", err) @@ -420,14 +421,27 @@ func (m *Manager) AllPrimary() []database.Replica { return replicas } -func (m *Manager) PrimaryPeerAddresses() []string { +func (m *Manager) FetchNATSPeers() []string { addresses := make([]string, 0, len(m.AllPrimary())) for _, replica := range m.AllPrimary() { - addresses = append(addresses, replica.RelayAddress) + if replica.ClusterHost == "" || replica.NATSPort == 0 { + continue + } + natsAddr := fmt.Sprintf("nats://%s:%d", replica.ClusterHost, replica.NATSPort) + addresses = append(addresses, natsAddr) } return addresses } +func (m *Manager) SetSelfNATSPort(port int32) { + m.mutex.Lock() + defer m.mutex.Unlock() + m.self.NATSPort = port + m.logger.Debug(context.Background(), "nats port updated", slog.F("port", port)) + // We're not really in a rush here, since it will take some time for our peers to dial and establish connections + // to us. So, we're not going to trigger a synchronous update. We'll just wait for the periodic update ticker. +} + // InRegion returns every replica in the given DERP region excluding itself. func (m *Manager) InRegion(regionID int32) []database.Replica { m.mutex.Lock() @@ -503,8 +517,8 @@ func (m *Manager) Close() error { Error: m.self.Error, DatabaseLatency: 0, // A stopped replica has no latency. Primary: false, // A stopped replica cannot be primary. - ClusterHost: "", // TODO - NATSPort: 0, // TODO + ClusterHost: m.self.ClusterHost, + NATSPort: 0, // A stopped replica cannot cluster with NATS }) if err != nil { return xerrors.Errorf("update replica: %w", err) diff --git a/enterprise/replicasync/replicasync_test.go b/enterprise/replicasync/replicasync_test.go index dfbd2fa2b1..6b2fc517d2 100644 --- a/enterprise/replicasync/replicasync_test.go +++ b/enterprise/replicasync/replicasync_test.go @@ -279,17 +279,19 @@ func TestReplica(t *testing.T) { require.NoError(t, server.UpdateNow(ctx)) requireNoCallback(t, called) }) - t.Run("PrimaryPeerAddresses", func(t *testing.T) { + t.Run("FetchNATSPeers", func(t *testing.T) { t.Parallel() db, pubsub := dbtestutil.NewDB(t) ctx := testutil.Context(t, testutil.WaitShort) - primary, err := db.InsertReplica(ctx, database.InsertReplicaParams{ + _, err := db.InsertReplica(ctx, database.InsertReplicaParams{ ID: uuid.New(), CreatedAt: dbtime.Now(), StartedAt: dbtime.Now(), UpdatedAt: dbtime.Now(), - RelayAddress: "nats://primary.example:6222", + RelayAddress: "https://primary-relay.example", Primary: true, + ClusterHost: "primary.example", + NATSPort: 6222, }) require.NoError(t, err) _, err = db.InsertReplica(ctx, database.InsertReplicaParams{ @@ -297,7 +299,7 @@ func TestReplica(t *testing.T) { CreatedAt: dbtime.Now(), StartedAt: dbtime.Now(), UpdatedAt: dbtime.Now(), - RelayAddress: "nats://proxy.example:6222", + RelayAddress: "https://proxy-relay.example", Primary: false, }) require.NoError(t, err) @@ -310,15 +312,24 @@ func TestReplica(t *testing.T) { }) require.NoError(t, err) server, err := replicasync.New(ctx, testutil.Logger(t), db, pubsub, &replicasync.Options{ - RelayAddress: "nats://self.example:6222", + RelayAddress: "https://self-relay.example", + ClusterHost: "self.example", + UpdateInterval: time.Hour, // we'll explicitly trigger this }) require.NoError(t, err) defer server.Close() - require.Contains(t, server.PrimaryPeerAddresses(), primary.RelayAddress) require.ElementsMatch(t, []string{ "nats://primary.example:6222", - "nats://self.example:6222", - }, server.PrimaryPeerAddresses()) + }, server.FetchNATSPeers()) + + server.SetSelfNATSPort(6223) + err = server.UpdateNow(ctx) + require.NoError(t, err) + + require.ElementsMatch(t, []string{ + "nats://primary.example:6222", + "nats://self.example:6223", + }, server.FetchNATSPeers()) }) t.Run("TwentyConcurrent", func(t *testing.T) { // Ensures that twenty concurrent replicas can spawn and all diff --git a/site/src/api/typesGenerated.ts b/site/src/api/typesGenerated.ts index 6ecb349f45..0f529fe491 100644 --- a/site/src/api/typesGenerated.ts +++ b/site/src/api/typesGenerated.ts @@ -3267,6 +3267,11 @@ export interface ChatWorkspaceTTLResponse { readonly workspace_ttl_ms: number; } +// From codersdk/deployment.go +export interface ClusterConfig { + readonly host: string; +} + // From codersdk/client.go /** * CoderDesktopTelemetryHeader contains a JSON-encoded representation of Desktop telemetry @@ -4282,6 +4287,7 @@ export interface DeploymentValues { readonly http_address?: string; readonly autobuild_poll_interval?: number; readonly job_hang_detector_interval?: number; + readonly cluster?: ClusterConfig; readonly derp?: DERP; readonly prometheus?: PrometheusConfig; readonly pprof?: PprofConfig;