opensource/coder
3
0
Fork 0
mirror of https://github.com/coder/coder.git synced 2026-09-24 15:04:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

chore: modify replicasync to handle NATS explicitly (#26666)

Browse Source 
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.
This commit is contained in:
Spike Curtis
2026-06-26 08:36:32 -04:00
committed by GitHub
parent 9f211ce5ae
commit 98e1ce133c
16 changed files with 219 additions and 132 deletions
Download Patch File Download Diff File Expand all files Collapse all files
enterprise/cli/server.go
+13 -2
View File
@@ -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(),
enterprise/coderd/coderd.go
+4 -1
View File
@@ -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
enterprise/coderd/coderd_test.go
+9 -3
View File
@@ -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)
enterprise/coderd/workspaceproxy.go
+6 -6
View File
@@ -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)
enterprise/replicasync/replicasync.go
+24 -10
View File
@@ -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)
enterprise/replicasync/replicasync_test.go
+19 -8
View File
@@ -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