From e8bd5004a20276dd361e079f58440e2c25a9db2d Mon Sep 17 00:00:00 2001 From: Spike Curtis Date: Wed, 24 Jun 2026 15:04:04 -0400 Subject: [PATCH] chore: add replica_host and nats_port to replicas table (#26665) relates to GRU-69 Adds cluster_host and nats_port to replicas table, to explicitly track NATS routes in the cluster. I decided to make the NATS support explicit and transport the port number over the replicasync so that different Coder Servers can run on different ports. This is not something customers will typically care about, but is very useful for testing, so that they can all run on localhost within one machine. I've also gone with a design where the NATS pubsub directly tells replicasync the port number _after_ it opens the socket. This is also very useful for testing because it allows us to have the OS assign the port number at runtime, avoiding races where we fail to bind to a free port. --- coderd/database/check_constraint.go | 1 + coderd/database/dump.sql | 11 ++++++- .../000530_relay_host_nats_port.down.sql | 2 ++ .../000530_relay_host_nats_port.up.sql | 5 +++ coderd/database/models.go | 29 ++++++++++------- coderd/database/queries.sql.go | 32 +++++++++++++++---- coderd/database/queries/replicas.sql | 10 ++++-- coderd/database/sqlc.yaml | 1 + enterprise/coderd/workspaceproxy.go | 6 ++++ enterprise/replicasync/replicasync.go | 8 +++++ 10 files changed, 83 insertions(+), 22 deletions(-) create mode 100644 coderd/database/migrations/000530_relay_host_nats_port.down.sql create mode 100644 coderd/database/migrations/000530_relay_host_nats_port.up.sql diff --git a/coderd/database/check_constraint.go b/coderd/database/check_constraint.go index 935c38c0d1..76b350816d 100644 --- a/coderd/database/check_constraint.go +++ b/coderd/database/check_constraint.go @@ -44,6 +44,7 @@ const ( CheckMcpServerConfigsAvailabilityCheck CheckConstraint = "mcp_server_configs_availability_check" // mcp_server_configs CheckMcpServerConfigsTransportCheck CheckConstraint = "mcp_server_configs_transport_check" // mcp_server_configs CheckMaxProvisionerLogsLength CheckConstraint = "max_provisioner_logs_length" // provisioner_jobs + CheckNatsPortValidTcp CheckConstraint = "nats_port_valid_tcp" // replicas CheckMaxLogsLength CheckConstraint = "max_logs_length" // workspace_agents CheckSubsystemsNotNone CheckConstraint = "subsystems_not_none" // workspace_agents CheckWorkspaceBuildsDeadlineBelowMaxDeadline CheckConstraint = "workspace_builds_deadline_below_max_deadline" // workspace_builds diff --git a/coderd/database/dump.sql b/coderd/database/dump.sql index 5c2ec2ef0f..7bb6c2c972 100644 --- a/coderd/database/dump.sql +++ b/coderd/database/dump.sql @@ -2791,9 +2791,18 @@ CREATE TABLE replicas ( database_latency integer NOT NULL, version text NOT NULL, error text DEFAULT ''::text NOT NULL, - "primary" boolean DEFAULT true NOT NULL + "primary" boolean DEFAULT true NOT NULL, + cluster_host text DEFAULT ''::text NOT NULL, + nats_port integer DEFAULT 0 NOT NULL, + CONSTRAINT nats_port_valid_tcp CHECK (((nats_port >= 0) AND (nats_port <= 65535))) ); +COMMENT ON COLUMN replicas.relay_address IS 'URL for DERP relays.'; + +COMMENT ON COLUMN replicas.cluster_host IS 'Hostname or IP address the replica is reachable at for clustering purposes.'; + +COMMENT ON COLUMN replicas.nats_port IS 'Port number for NATS clustering. 0 means NATS is disabled.'; + CREATE TABLE site_configs ( key character varying(256) NOT NULL, value text NOT NULL diff --git a/coderd/database/migrations/000530_relay_host_nats_port.down.sql b/coderd/database/migrations/000530_relay_host_nats_port.down.sql new file mode 100644 index 0000000000..fc4e18fe10 --- /dev/null +++ b/coderd/database/migrations/000530_relay_host_nats_port.down.sql @@ -0,0 +1,2 @@ +ALTER TABLE replicas DROP COLUMN nats_port; +ALTER TABLE replicas DROP COLUMN cluster_host; diff --git a/coderd/database/migrations/000530_relay_host_nats_port.up.sql b/coderd/database/migrations/000530_relay_host_nats_port.up.sql new file mode 100644 index 0000000000..e086de8514 --- /dev/null +++ b/coderd/database/migrations/000530_relay_host_nats_port.up.sql @@ -0,0 +1,5 @@ +COMMENT ON COLUMN replicas.relay_address IS 'URL for DERP relays.'; +ALTER TABLE replicas ADD COLUMN cluster_host text DEFAULT ''::text NOT NULL; +COMMENT ON COLUMN replicas.cluster_host IS 'Hostname or IP address the replica is reachable at for clustering purposes.'; +ALTER TABLE replicas ADD COLUMN nats_port integer DEFAULT 0 NOT NULL CONSTRAINT nats_port_valid_tcp CHECK ( nats_port >= 0 AND nats_port <= 65535); +COMMENT ON COLUMN replicas.nats_port IS 'Port number for NATS clustering. 0 means NATS is disabled.'; diff --git a/coderd/database/models.go b/coderd/database/models.go index dbba7981c3..5a5e2cbe67 100644 --- a/coderd/database/models.go +++ b/coderd/database/models.go @@ -5558,18 +5558,23 @@ type ProvisionerKey struct { } type Replica struct { - ID uuid.UUID `db:"id" json:"id"` - CreatedAt time.Time `db:"created_at" json:"created_at"` - StartedAt time.Time `db:"started_at" json:"started_at"` - StoppedAt sql.NullTime `db:"stopped_at" json:"stopped_at"` - UpdatedAt time.Time `db:"updated_at" json:"updated_at"` - Hostname string `db:"hostname" json:"hostname"` - RegionID int32 `db:"region_id" json:"region_id"` - RelayAddress string `db:"relay_address" json:"relay_address"` - DatabaseLatency int32 `db:"database_latency" json:"database_latency"` - Version string `db:"version" json:"version"` - Error string `db:"error" json:"error"` - Primary bool `db:"primary" json:"primary"` + ID uuid.UUID `db:"id" json:"id"` + CreatedAt time.Time `db:"created_at" json:"created_at"` + StartedAt time.Time `db:"started_at" json:"started_at"` + StoppedAt sql.NullTime `db:"stopped_at" json:"stopped_at"` + UpdatedAt time.Time `db:"updated_at" json:"updated_at"` + Hostname string `db:"hostname" json:"hostname"` + RegionID int32 `db:"region_id" json:"region_id"` + // URL for DERP relays. + RelayAddress string `db:"relay_address" json:"relay_address"` + DatabaseLatency int32 `db:"database_latency" json:"database_latency"` + Version string `db:"version" json:"version"` + Error string `db:"error" json:"error"` + Primary bool `db:"primary" json:"primary"` + // Hostname or IP address the replica is reachable at for clustering purposes. + ClusterHost string `db:"cluster_host" json:"cluster_host"` + // Port number for NATS clustering. 0 means NATS is disabled. + NATSPort int32 `db:"nats_port" json:"nats_port"` } type SiteConfig struct { diff --git a/coderd/database/queries.sql.go b/coderd/database/queries.sql.go index 3825913982..6b44efc737 100644 --- a/coderd/database/queries.sql.go +++ b/coderd/database/queries.sql.go @@ -23739,7 +23739,7 @@ func (q *sqlQuerier) DeleteReplicasUpdatedBefore(ctx context.Context, updatedAt } const getReplicaByID = `-- name: GetReplicaByID :one -SELECT id, created_at, started_at, stopped_at, updated_at, hostname, region_id, relay_address, database_latency, version, error, "primary" FROM replicas WHERE id = $1 +SELECT id, created_at, started_at, stopped_at, updated_at, hostname, region_id, relay_address, database_latency, version, error, "primary", cluster_host, nats_port FROM replicas WHERE id = $1 ` func (q *sqlQuerier) GetReplicaByID(ctx context.Context, id uuid.UUID) (Replica, error) { @@ -23758,12 +23758,14 @@ func (q *sqlQuerier) GetReplicaByID(ctx context.Context, id uuid.UUID) (Replica, &i.Version, &i.Error, &i.Primary, + &i.ClusterHost, + &i.NATSPort, ) return i, err } const getReplicasUpdatedAfter = `-- name: GetReplicasUpdatedAfter :many -SELECT id, created_at, started_at, stopped_at, updated_at, hostname, region_id, relay_address, database_latency, version, error, "primary" FROM replicas WHERE updated_at > $1 AND stopped_at IS NULL +SELECT id, created_at, started_at, stopped_at, updated_at, hostname, region_id, relay_address, database_latency, version, error, "primary", cluster_host, nats_port FROM replicas WHERE updated_at > $1 AND stopped_at IS NULL ` func (q *sqlQuerier) GetReplicasUpdatedAfter(ctx context.Context, updatedAt time.Time) ([]Replica, error) { @@ -23788,6 +23790,8 @@ func (q *sqlQuerier) GetReplicasUpdatedAfter(ctx context.Context, updatedAt time &i.Version, &i.Error, &i.Primary, + &i.ClusterHost, + &i.NATSPort, ); err != nil { return nil, err } @@ -23813,8 +23817,10 @@ INSERT INTO replicas ( relay_address, version, database_latency, - "primary" -) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) RETURNING id, created_at, started_at, stopped_at, updated_at, hostname, region_id, relay_address, database_latency, version, error, "primary" + "primary", + cluster_host, + nats_port +) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) RETURNING id, created_at, started_at, stopped_at, updated_at, hostname, region_id, relay_address, database_latency, version, error, "primary", cluster_host, nats_port ` type InsertReplicaParams struct { @@ -23828,6 +23834,8 @@ type InsertReplicaParams struct { Version string `db:"version" json:"version"` DatabaseLatency int32 `db:"database_latency" json:"database_latency"` Primary bool `db:"primary" json:"primary"` + ClusterHost string `db:"cluster_host" json:"cluster_host"` + NATSPort int32 `db:"nats_port" json:"nats_port"` } func (q *sqlQuerier) InsertReplica(ctx context.Context, arg InsertReplicaParams) (Replica, error) { @@ -23842,6 +23850,8 @@ func (q *sqlQuerier) InsertReplica(ctx context.Context, arg InsertReplicaParams) arg.Version, arg.DatabaseLatency, arg.Primary, + arg.ClusterHost, + arg.NATSPort, ) var i Replica err := row.Scan( @@ -23857,6 +23867,8 @@ func (q *sqlQuerier) InsertReplica(ctx context.Context, arg InsertReplicaParams) &i.Version, &i.Error, &i.Primary, + &i.ClusterHost, + &i.NATSPort, ) return i, err } @@ -23872,8 +23884,10 @@ UPDATE replicas SET version = $8, error = $9, database_latency = $10, - "primary" = $11 -WHERE id = $1 RETURNING id, created_at, started_at, stopped_at, updated_at, hostname, region_id, relay_address, database_latency, version, error, "primary" + "primary" = $11, + cluster_host = $12, + nats_port = $13 +WHERE id = $1 RETURNING id, created_at, started_at, stopped_at, updated_at, hostname, region_id, relay_address, database_latency, version, error, "primary", cluster_host, nats_port ` type UpdateReplicaParams struct { @@ -23888,6 +23902,8 @@ type UpdateReplicaParams struct { Error string `db:"error" json:"error"` DatabaseLatency int32 `db:"database_latency" json:"database_latency"` Primary bool `db:"primary" json:"primary"` + ClusterHost string `db:"cluster_host" json:"cluster_host"` + NATSPort int32 `db:"nats_port" json:"nats_port"` } func (q *sqlQuerier) UpdateReplica(ctx context.Context, arg UpdateReplicaParams) (Replica, error) { @@ -23903,6 +23919,8 @@ func (q *sqlQuerier) UpdateReplica(ctx context.Context, arg UpdateReplicaParams) arg.Error, arg.DatabaseLatency, arg.Primary, + arg.ClusterHost, + arg.NATSPort, ) var i Replica err := row.Scan( @@ -23918,6 +23936,8 @@ func (q *sqlQuerier) UpdateReplica(ctx context.Context, arg UpdateReplicaParams) &i.Version, &i.Error, &i.Primary, + &i.ClusterHost, + &i.NATSPort, ) return i, err } diff --git a/coderd/database/queries/replicas.sql b/coderd/database/queries/replicas.sql index 5a0b4ac0fe..3652a96879 100644 --- a/coderd/database/queries/replicas.sql +++ b/coderd/database/queries/replicas.sql @@ -15,8 +15,10 @@ INSERT INTO replicas ( relay_address, version, database_latency, - "primary" -) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) RETURNING *; + "primary", + cluster_host, + nats_port +) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) RETURNING *; -- name: UpdateReplica :one UPDATE replicas SET @@ -29,7 +31,9 @@ UPDATE replicas SET version = $8, error = $9, database_latency = $10, - "primary" = $11 + "primary" = $11, + cluster_host = $12, + nats_port = $13 WHERE id = $1 RETURNING *; -- name: DeleteReplicasUpdatedBefore :exec diff --git a/coderd/database/sqlc.yaml b/coderd/database/sqlc.yaml index 3a72a147ef..abc19ee1ae 100644 --- a/coderd/database/sqlc.yaml +++ b/coderd/database/sqlc.yaml @@ -32,6 +32,7 @@ sql: # identifiers follow Go naming conventions. - id - ai + - nats overrides: - column: "api_keys.scopes" go_type: diff --git a/enterprise/coderd/workspaceproxy.go b/enterprise/coderd/workspaceproxy.go index 718aeec38e..530aa89e09 100644 --- a/enterprise/coderd/workspaceproxy.go +++ b/enterprise/coderd/workspaceproxy.go @@ -667,6 +667,8 @@ func (api *API) workspaceProxyRegister(rw http.ResponseWriter, r *http.Request) Error: req.ReplicaError, DatabaseLatency: 0, Primary: false, + ClusterHost: "", // TODO + NATSPort: 0, // TODO }) if err != nil { return xerrors.Errorf("update replica: %w", err) @@ -684,6 +686,8 @@ func (api *API) workspaceProxyRegister(rw http.ResponseWriter, r *http.Request) Version: req.Version, DatabaseLatency: 0, Primary: false, + ClusterHost: "", // TODO + NATSPort: 0, // TODO }) if err != nil { return xerrors.Errorf("insert replica: %w", err) @@ -826,6 +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 }) if err != nil { return xerrors.Errorf("update replica: %w", err) diff --git a/enterprise/replicasync/replicasync.go b/enterprise/replicasync/replicasync.go index e7c067fff8..7ea6dd5818 100644 --- a/enterprise/replicasync/replicasync.go +++ b/enterprise/replicasync/replicasync.go @@ -77,6 +77,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 }) if err != nil { return nil, xerrors.Errorf("insert replica: %w", err) @@ -327,6 +329,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 }) if err != nil { if !errors.Is(err, sql.ErrNoRows) { @@ -346,6 +350,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 }) if err != nil { return xerrors.Errorf("update replica: %w", err) @@ -497,6 +503,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 }) if err != nil { return xerrors.Errorf("update replica: %w", err)