Add flags for HA

coder · kylecarbs · Oct 17, 2022 · Sep 22, 2022 · Sep 23, 2022 · Sep 23, 2022
commit 585bc1dfc81996b9967475de7f4249c93b24aa46
diff --git a/cli/config/file.go b/cli/config/file.go
@@ -13,6 +13,11 @@ func (r Root) Session() File {
 	return File(filepath.Join(string(r), "session"))
 }
 
+// ReplicaID is a unique identifier for the Coder server.
+func (r Root) ReplicaID() File {
+	return File(filepath.Join(string(r), "replica_id"))
+}
+
 func (r Root) URL() File {
 	return File(filepath.Join(string(r), "url"))
 }

diff --git a/cli/deployment/flags.go b/cli/deployment/flags.go
@@ -85,6 +85,13 @@ func Flags() *codersdk.DeploymentFlags {
 			Description: "Addresses for STUN servers to establish P2P connections. Set empty to disable P2P connections.",
 			Default:     []string{"stun.l.google.com:19302"},
 		},
+		DerpServerRelayAddress: &codersdk.StringFlag{
+			Name:        "DERP Server Relay Address",
+			Flag:        "derp-server-relay-address",
+			EnvVar:      "CODER_DERP_SERVER_RELAY_ADDRESS",
+			Description: "An HTTP address that is accessible by other replicas to relay DERP traffic. Required for high availability.",
+			Enterprise:  true,
+		},
 		DerpConfigURL: &codersdk.StringFlag{
 			Name:        "DERP Config URL",
 			Flag:        "derp-config-url",
@@ -123,6 +130,14 @@ func Flags() *codersdk.DeploymentFlags {
 			Description: "The bind address to serve pprof.",
 			Default:     "127.0.0.1:6060",
 		},
+		HighAvailability: &codersdk.BoolFlag{
+			Name:        "High Availability",
+			Flag:        "high-availability",
+			EnvVar:      "CODER_HIGH_AVAILABILITY",
+			Description: "Specifies whether high availability is enabled.",
+			Default:     true,
+			Enterprise:  true,
+		},
 		CacheDir: &codersdk.StringFlag{
 			Name:        "Cache Directory",
 			Flag:        "cache-dir",

diff --git a/cli/root.go b/cli/root.go
@@ -100,7 +100,7 @@ func Core() []*cobra.Command {
 }
 
 func AGPL() []*cobra.Command {
-	all := append(Core(), Server(deployment.Flags(), func(_ context.Context, o *coderd.Options) (*coderd.API, error) {
+	all := append(Core(), Server(deployment.Flags(), func(_ context.Context, _ config.Root, o *coderd.Options) (*coderd.API, error) {
 		return coderd.New(o), nil
 	}))
 	return all

diff --git a/cli/server.go b/cli/server.go
@@ -67,7 +67,7 @@ import (
 )
 
 // nolint:gocyclo
-func Server(dflags *codersdk.DeploymentFlags, newAPI func(context.Context, *coderd.Options) (*coderd.API, error)) *cobra.Command {
+func Server(dflags *codersdk.DeploymentFlags, newAPI func(context.Context, config.Root, *coderd.Options) (*coderd.API, error)) *cobra.Command {
 	root := &cobra.Command{
 		Use:   "server",
 		Short: "Start a Coder server",
@@ -463,7 +463,7 @@ func Server(dflags *codersdk.DeploymentFlags, newAPI func(context.Context, *code
 				), dflags.PromAddress.Value, "prometheus")()
 			}
 
-			coderAPI, err := newAPI(ctx, options)
+			coderAPI, err := newAPI(ctx, config, options)
 			if err != nil {
 				return err
 			}

diff --git a/coderd/coderd.go b/coderd/coderd.go
@@ -77,6 +77,7 @@ type Options struct {
 	AutoImportTemplates  []AutoImportTemplate
 
 	TailnetCoordinator tailnet.Coordinator
+	DERPServer         *derp.Server
 	DERPMap            *tailcfg.DERPMap
 
 	MetricsCacheRefreshInterval time.Duration
@@ -121,6 +122,9 @@ func New(options *Options) *API {
 	if options.TailnetCoordinator == nil {
 		options.TailnetCoordinator = tailnet.NewCoordinator()
 	}
+	if options.DERPServer == nil {
+		options.DERPServer = derp.NewServer(key.NewNode(), tailnet.Logger(options.Logger))
+	}
 	if options.Auditor == nil {
 		options.Auditor = audit.NewNop()
 	}
@@ -160,7 +164,6 @@ func New(options *Options) *API {
 	api.WorkspaceQuotaEnforcer.Store(&options.WorkspaceQuotaEnforcer)
 	api.workspaceAgentCache = wsconncache.New(api.dialWorkspaceAgentTailnet, 0)
 	api.TailnetCoordinator.Store(&options.TailnetCoordinator)
-	api.derpServer = derp.NewServer(key.NewNode(), tailnet.Logger(options.Logger))
 	oauthConfigs := &httpmw.OAuth2Configs{
 		Github: options.GithubOAuth2Config,
 		OIDC:   options.OIDCConfig,
@@ -228,7 +231,7 @@ func New(options *Options) *API {
 	r.Route("/%40{user}/{workspace_and_agent}/apps/{workspaceapp}", apps)
 	r.Route("/@{user}/{workspace_and_agent}/apps/{workspaceapp}", apps)
 	r.Route("/derp", func(r chi.Router) {
-		r.Get("/", derphttp.Handler(api.derpServer).ServeHTTP)
+		r.Get("/", derphttp.Handler(api.DERPServer).ServeHTTP)
 		// This is used when UDP is blocked, and latency must be checked via HTTP(s).
 		r.Get("/latency-check", func(w http.ResponseWriter, r *http.Request) {
 			w.WriteHeader(http.StatusOK)
@@ -540,7 +543,6 @@ type API struct {
 	// RootHandler serves "/"
 	RootHandler chi.Router
 
-	derpServer          *derp.Server
 	metricsCache        *metricscache.Cache
 	siteHandler         http.Handler
 	websocketWaitMutex  sync.Mutex

diff --git a/coderd/coderdtest/coderdtest.go b/coderd/coderdtest/coderdtest.go
@@ -81,6 +81,12 @@ type Options struct {
 	MetricsCacheRefreshInterval time.Duration
 	AgentStatsRefreshInterval   time.Duration
 	DeploymentFlags             *codersdk.DeploymentFlags
+
+	// Overriding the database is heavily discouraged.
+	// It should only be used in cases where multiple Coder
+	// test instances are running against the same database.
+	Database database.Store
+	Pubsub   database.Pubsub
 }
 
 // New constructs a codersdk client connected to an in-memory API instance.
@@ -135,13 +141,14 @@ func NewOptions(t *testing.T, options *Options) (*httptest.Server, context.Cance
 			close(options.AutobuildStats)
 		})
 	}
-
-	db, pubsub := dbtestutil.NewDB(t)
+	if options.Database == nil {
+		options.Database, options.Pubsub = dbtestutil.NewDB(t)
+	}
 
 	ctx, cancelFunc := context.WithCancel(context.Background())
 	lifecycleExecutor := executor.New(
 		ctx,
-		db,
+		options.Database,
 		slogtest.Make(t, nil).Named("autobuild.executor").Leveled(slog.LevelDebug),
 		options.AutobuildTicker,
 	).WithStatsChannel(options.AutobuildStats)
@@ -181,8 +188,8 @@ func NewOptions(t *testing.T, options *Options) (*httptest.Server, context.Cance
 		AppHostname:                    options.AppHostname,
 		Logger:                         slogtest.Make(t, nil).Leveled(slog.LevelDebug),
 		CacheDir:                       t.TempDir(),
-		Database:                       db,
-		Pubsub:                         pubsub,
+		Database:                       options.Database,
+		Pubsub:                         options.Pubsub,
 
 		Auditor:              options.Auditor,
 		AWSCertificates:      options.AWSCertificates,

diff --git a/codersdk/flags.go b/codersdk/flags.go
@@ -19,6 +19,7 @@ type DeploymentFlags struct {
 	DerpServerRegionCode             *StringFlag      `json:"derp_server_region_code" typescript:",notnull"`
 	DerpServerRegionName             *StringFlag      `json:"derp_server_region_name" typescript:",notnull"`
 	DerpServerSTUNAddresses          *StringArrayFlag `json:"derp_server_stun_address" typescript:",notnull"`
+	DerpServerRelayAddress           *StringFlag      `json:"derp_server_relay_address" typescript:",notnull"`
 	DerpConfigURL                    *StringFlag      `json:"derp_config_url" typescript:",notnull"`
 	DerpConfigPath                   *StringFlag      `json:"derp_config_path" typescript:",notnull"`
 	PromEnabled                      *BoolFlag        `json:"prom_enabled" typescript:",notnull"`
@@ -59,6 +60,7 @@ type DeploymentFlags struct {
 	Verbose                          *BoolFlag        `json:"verbose" typescript:",notnull"`
 	AuditLogging                     *BoolFlag        `json:"audit_logging" typescript:",notnull"`
 	BrowserOnly                      *BoolFlag        `json:"browser_only" typescript:",notnull"`
+	HighAvailability                 *BoolFlag        `json:"high_availability" typescript:",notnull"`
 	SCIMAuthHeader                   *StringFlag      `json:"scim_auth_header" typescript:",notnull"`
 	UserWorkspaceQuota               *IntFlag         `json:"user_workspace_quota" typescript:",notnull"`
 }

diff --git a/codersdk/replicas.go b/codersdk/replicas.go
@@ -0,0 +1,22 @@
+package codersdk
+
+import (
+	"time"
+
+	"github.com/google/uuid"
+)
+
+type Replica struct {
+	// ID is the unique identifier for the replica.
+	ID uuid.UUID `json:"id"`
+	// Hostname is the hostname of the replica.
+	Hostname string `json:"hostname"`
+	// CreatedAt is when the replica was first seen.
+	CreatedAt time.Time `json:"created_at"`
+	// Active determines whether the replica is online.
+	Active bool `json:"active"`
+	// RelayAddress is the accessible address to relay DERP connections.
+	RelayAddress string `json:"relay_address"`
+	// Error is the error.
+	Error string `json:"error"`
+}
diff --git a/enterprise/cli/server.go b/enterprise/cli/server.go
@@ -3,8 +3,12 @@ package cli
 import (
 	"context"
 
+	"github.com/google/uuid"
 	"github.com/spf13/cobra"
 
+	"cdr.dev/slog"
+
+	"github.com/coder/coder/cli/config"
 	"github.com/coder/coder/cli/deployment"
 	"github.com/coder/coder/enterprise/coderd"
 
@@ -14,14 +18,29 @@ import (
 
 func server() *cobra.Command {
 	dflags := deployment.Flags()
-	cmd := agpl.Server(dflags, func(ctx context.Context, options *agplcoderd.Options) (*agplcoderd.API, error) {
+	cmd := agpl.Server(dflags, func(ctx context.Context, cfg config.Root, options *agplcoderd.Options) (*agplcoderd.API, error) {
+		replicaIDRaw, err := cfg.ReplicaID().Read()
+		if err != nil {
+			replicaIDRaw = uuid.NewString()
+		}
+		replicaID, err := uuid.Parse(replicaIDRaw)
+		if err != nil {
+			options.Logger.Warn(ctx, "failed to parse replica id", slog.Error(err), slog.F("replica_id", replicaIDRaw))
+			replicaID = uuid.New()
+		}
 		o := &coderd.Options{
 			AuditLogging:       dflags.AuditLogging.Value,
 			BrowserOnly:        dflags.BrowserOnly.Value,
 			SCIMAPIKey:         []byte(dflags.SCIMAuthHeader.Value),
 			UserWorkspaceQuota: dflags.UserWorkspaceQuota.Value,
-			RBACEnabled:        true,
-			Options:            options,
+			RBAC:               true,
+			HighAvailability:   dflags.HighAvailability.Value,
+
+			ReplicaID:              replicaID,
+			DERPServerRelayAddress: dflags.DerpServerRelayAddress.Value,
+			DERPServerRegionID:     dflags.DerpServerRegionID.Value,
+
+			Options: options,
 		}
 		api, err := coderd.New(ctx, o)
 		if err != nil {

diff --git a/enterprise/coderd/coderd.go b/enterprise/coderd/coderd.go
@@ -11,6 +11,7 @@ import (
 
 	"github.com/cenkalti/backoff/v4"
 	"github.com/go-chi/chi/v5"
+	"github.com/google/uuid"
 
 	"cdr.dev/slog"
 	"github.com/coder/coder/coderd"
@@ -24,6 +25,8 @@ import (
 	"github.com/coder/coder/enterprise/audit/backends"
 	"github.com/coder/coder/enterprise/coderd/license"
 	"github.com/coder/coder/enterprise/highavailability"
+	"github.com/coder/coder/enterprise/highavailability/derpmesh"
+	"github.com/coder/coder/enterprise/highavailability/replica"
 	"github.com/coder/coder/tailnet"
 )
 
@@ -43,6 +46,7 @@ func New(ctx context.Context, options *Options) (*API, error) {
 		Options:                options,
 		cancelEntitlementsLoop: cancelFunc,
 	}
+
 	oauthConfigs := &httpmw.OAuth2Configs{
 		Github: options.GithubOAuth2Config,
 		OIDC:   options.OIDCConfig,
@@ -113,7 +117,27 @@ func New(ctx context.Context, options *Options) (*API, error) {
 		})
 	}
 
-	err := api.updateEntitlements(ctx)
+	// If high availability is disabled and multiple replicas appear, show an error.
+	// If high availability is enabled and the built-in DERP is but the DERP relay isn't set, show an error.
+	// We need to block meshing if high availability is disabled, because the meshing code would just work.
+	// SetAddresses([]string{})
+
+	api.AGPL.RootHandler.Route("/replicas", func(r chi.Router) {
+
+	})
+
+	var err error
+	api.replica, err = replica.New(ctx, options.Logger, options.Database, options.Pubsub, replica.Options{
+		ID:           options.ReplicaID,
+		RelayAddress: options.DERPServerRelayAddress,
+		RegionID:     int32(options.DERPServerRegionID),
+	})
+	if err != nil {
+		return nil, xerrors.Errorf("initialize replica: %w", err)
+	}
+	api.derpMesh = derpmesh.New(options.Logger, api.DERPServer)
+
+	err = api.updateEntitlements(ctx)
 	if err != nil {
 		return nil, xerrors.Errorf("update entitlements: %w", err)
 	}
@@ -125,12 +149,18 @@ func New(ctx context.Context, options *Options) (*API, error) {
 type Options struct {
 	*coderd.Options
 
-	RBACEnabled  bool
+	RBAC         bool
 	AuditLogging bool
 	// Whether to block non-browser connections.
 	BrowserOnly        bool
 	SCIMAPIKey         []byte
 	UserWorkspaceQuota int
+	HighAvailability   bool
+
+	// Used for high availability.
+	DERPServerRelayAddress string
+	DERPServerRegionID     int
+	ReplicaID              uuid.UUID
 
 	EntitlementsUpdateInterval time.Duration
 	Keys                       map[string]ed25519.PublicKey
@@ -140,13 +170,20 @@ type API struct {
 	AGPL *coderd.API
 	*Options
 
+	// Detects multiple Coder replicas running at the same time.
+	replica *replica.Server
+	// Meshes DERP connections from multiple replicas.
+	derpMesh *derpmesh.Mesh
+
 	cancelEntitlementsLoop func()
 	entitlementsMu         sync.RWMutex
 	entitlements           codersdk.Entitlements
 }
 
 func (api *API) Close() error {
 	api.cancelEntitlementsLoop()
+	_ = api.replica.Close()
+	_ = api.derpMesh.Close()
 	return api.AGPL.Close()
 }
 
@@ -155,11 +192,12 @@ func (api *API) updateEntitlements(ctx context.Context) error {
 	defer api.entitlementsMu.Unlock()
 
 	entitlements, err := license.Entitlements(ctx, api.Database, api.Logger, api.Keys, map[string]bool{
-		codersdk.FeatureAuditLog:       api.AuditLogging,
-		codersdk.FeatureBrowserOnly:    api.BrowserOnly,
-		codersdk.FeatureSCIM:           len(api.SCIMAPIKey) != 0,
-		codersdk.FeatureWorkspaceQuota: api.UserWorkspaceQuota != 0,
-		codersdk.FeatureTemplateRBAC:   api.RBACEnabled,
+		codersdk.FeatureAuditLog:         api.AuditLogging,
+		codersdk.FeatureBrowserOnly:      api.BrowserOnly,
+		codersdk.FeatureSCIM:             len(api.SCIMAPIKey) != 0,
+		codersdk.FeatureWorkspaceQuota:   api.UserWorkspaceQuota != 0,
+		codersdk.FeatureHighAvailability: api.HighAvailability,
+		codersdk.FeatureTemplateRBAC:     api.RBAC,
 	})
 	if err != nil {
 		return err
@@ -210,13 +248,23 @@ func (api *API) updateEntitlements(ctx context.Context) error {
 		if enabled {
 			haCoordinator, err := highavailability.NewCoordinator(api.Logger, api.Pubsub)
 			if err != nil {
-				api.Logger.Error(ctx, "unable to setup HA tailnet coordinator", slog.Error(err))
+				api.Logger.Error(ctx, "unable to set up high availability coordinator", slog.Error(err))
 				// If we try to setup the HA coordinator and it fails, nothing
 				// is actually changing.
 				changed = false
 			} else {
 				coordinator = haCoordinator
 			}
+
+			api.replica.SetCallback(func() {
+				addresses := make([]string, 0)
+				for _, replica := range api.replica.Regional() {
+					addresses = append(addresses, replica.RelayAddress)
+				}
+				api.derpMesh.SetAddresses(addresses)
+			})
+		} else {
+			api.derpMesh.SetAddresses([]string{})
 		}
 
 		// Recheck changed in case the HA coordinator failed to set up.

diff --git a/enterprise/coderd/coderdenttest/coderdenttest.go b/enterprise/coderd/coderdenttest/coderdenttest.go
@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/golang-jwt/jwt/v4"
+	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -62,10 +63,14 @@ func NewWithAPI(t *testing.T, options *Options) (*codersdk.Client, io.Closer, *c
 	}
 	srv, cancelFunc, oop := coderdtest.NewOptions(t, options.Options)
 	coderAPI, err := coderd.New(context.Background(), &coderd.Options{
-		RBACEnabled:                true,
+		RBAC:                       true,
 		AuditLogging:               options.AuditLogging,
 		BrowserOnly:                options.BrowserOnly,
 		SCIMAPIKey:                 options.SCIMAPIKey,
+		DERPServerRelayAddress:     oop.AccessURL.String(),
+		DERPServerRegionID:         1,
+		HighAvailability:           true,
+		ReplicaID:                  uuid.New(),
 		UserWorkspaceQuota:         options.UserWorkspaceQuota,
 		Options:                    oop,
 		EntitlementsUpdateInterval: options.EntitlementsUpdateInterval,