From 98e1ce133c5a2cda5eea985dc4bcce583503339d Mon Sep 17 00:00:00 2001 From: Spike Curtis Date: Fri, 26 Jun 2026 08:36:32 -0400 Subject: [PATCH] chore: modify replicasync to handle NATS explicitly (#26666) relates to GRU-69 Modifies replicasync to handle discovering NATS enabled primary replicas explicitly, and passing that info to the NATS Pubsub. This PR adds a new deployment value to explicitly represent the host or IP that the replica can be reached on. It isn't wired up to the CLI, but piggybacks on the DERP config for now. We learn the NATS port directly from NATS at runtime, and propagate it thru replicasync to learn all peers for clustering. --- coderd/apidoc/docs.go | 11 +++ coderd/apidoc/swagger.json | 11 +++ coderd/x/nats/cluster.go | 34 +++++---- coderd/x/nats/cluster_internal_test.go | 88 +++++++--------------- coderd/x/nats/natsbench/topology.go | 4 +- coderd/x/nats/pubsub.go | 63 ++++++++++------ codersdk/deployment.go | 5 ++ docs/reference/api/general.md | 3 + docs/reference/api/schemas.md | 21 ++++++ enterprise/cli/server.go | 15 +++- enterprise/coderd/coderd.go | 5 +- enterprise/coderd/coderd_test.go | 12 ++- enterprise/coderd/workspaceproxy.go | 12 +-- enterprise/replicasync/replicasync.go | 34 ++++++--- enterprise/replicasync/replicasync_test.go | 27 +++++-- site/src/api/typesGenerated.ts | 6 ++ 16 files changed, 219 insertions(+), 132 deletions(-) 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;