opensource/coder
3
0
Fork 0
mirror of https://github.com/coder/coder.git synced 2026-09-21 12:44:32 +08:00
Code Issues Packages Projects Releases Wiki Activity

chore: add replica_host and nats_port to replicas table (#26665)

Browse Source 
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.
This commit is contained in:
Spike Curtis
2026-06-24 15:04:04 -04:00
committed by GitHub
parent e0305cfa59
commit e8bd5004a2
10 changed files with 83 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
coderd/database/check_constraint.go
Generated
+1
View File
@@ -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
coderd/database/dump.sql
Generated
+10 -1
View File
@@ -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
coderd/database/migrations/000530_relay_host_nats_port.down.sql
+2
View File
@@ -0,0 +1,2 @@
ALTER TABLE replicas DROP COLUMN nats_port;
ALTER TABLE replicas DROP COLUMN cluster_host;
coderd/database/migrations/000530_relay_host_nats_port.up.sql
+5
View File
@@ -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.';
coderd/database/models.go
Generated
+17 -12
View File
@@ -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 {
coderd/database/queries.sql.go
Generated
+26 -6
View File
@@ -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
}
coderd/database/queries/replicas.sql
+7 -3
View File
@@ -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
coderd/database/sqlc.yaml
+1
View File
@@ -32,6 +32,7 @@ sql:
# identifiers follow Go naming conventions.
- id
- ai
- nats
overrides:
- column: "api_keys.scopes"
go_type:
enterprise/coderd/workspaceproxy.go
+6
View File
@@ -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)
enterprise/replicasync/replicasync.go
+8
View File
@@ -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)