chore: Proxy health status checks + endpoint (#7233)

* chore: Implement workspace proxy health check cron

At a given interval will check the reachability of workspace proxies.

* Proxyhealth is an enterprise feature
* Start proxyhealth go routine on enterprise coder
This commit is contained in:
Steven Masley
2023-04-24 10:25:35 -05:00
committed by GitHub
parent 63e68c11d1
commit 3129741e08
13 changed files with 912 additions and 60 deletions
+35 -7
View File
@@ -24,6 +24,7 @@ import (
"github.com/coder/coder/coderd/schedule"
"github.com/coder/coder/codersdk"
"github.com/coder/coder/enterprise/coderd/license"
"github.com/coder/coder/enterprise/coderd/proxyhealth"
"github.com/coder/coder/enterprise/derpmesh"
"github.com/coder/coder/enterprise/replicasync"
"github.com/coder/coder/enterprise/tailnet"
@@ -52,9 +53,11 @@ func New(ctx context.Context, options *Options) (*API, error) {
}
ctx, cancelFunc := context.WithCancel(ctx)
api := &API{
AGPL: coderd.New(options.Options),
Options: options,
cancelEntitlementsLoop: cancelFunc,
ctx: ctx,
cancel: cancelFunc,
AGPL: coderd.New(options.Options),
Options: options,
}
api.AGPL.Options.SetUserGroups = api.setUserGroups
@@ -226,6 +229,24 @@ func New(ctx context.Context, options *Options) (*API, error) {
}
api.derpMesh = derpmesh.New(options.Logger.Named("derpmesh"), api.DERPServer, meshTLSConfig)
if api.AGPL.Experiments.Enabled(codersdk.ExperimentMoons) {
// Proxy health is a moon feature.
api.proxyHealth, err = proxyhealth.New(&proxyhealth.Options{
Interval: time.Second * 5,
DB: api.Database,
Logger: options.Logger.Named("proxyhealth"),
Client: api.HTTPClient,
Prometheus: api.PrometheusRegistry,
})
if err != nil {
return nil, xerrors.Errorf("initialize proxy health: %w", err)
}
go api.proxyHealth.Run(ctx)
// Force the initial loading of the cache. Do this in a go routine in case
// the calls to the workspace proxies hang and this takes some time.
go api.forceWorkspaceProxyHealthUpdate(ctx)
}
err = api.updateEntitlements(ctx)
if err != nil {
return nil, xerrors.Errorf("update entitlements: %w", err)
@@ -249,6 +270,7 @@ type Options struct {
DERPServerRegionID int
EntitlementsUpdateInterval time.Duration
ProxyHealthInterval time.Duration
Keys map[string]ed25519.PublicKey
}
@@ -256,18 +278,24 @@ type API struct {
AGPL *coderd.API
*Options
// ctx is canceled immediately on shutdown, it can be used to abort
// interruptible tasks.
ctx context.Context
cancel context.CancelFunc
// Detects multiple Coder replicas running at the same time.
replicaManager *replicasync.Manager
// Meshes DERP connections from multiple replicas.
derpMesh *derpmesh.Mesh
// proxyHealth checks the reachability of all workspace proxies.
proxyHealth *proxyhealth.ProxyHealth
cancelEntitlementsLoop func()
entitlementsMu sync.RWMutex
entitlements codersdk.Entitlements
entitlementsMu sync.RWMutex
entitlements codersdk.Entitlements
}
func (api *API) Close() error {
api.cancelEntitlementsLoop()
api.cancel()
_ = api.replicaManager.Close()
_ = api.derpMesh.Close()
return api.AGPL.Close()
@@ -0,0 +1,292 @@
package proxyhealth
import (
"context"
"encoding/json"
"fmt"
"net/http"
"strings"
"sync"
"sync/atomic"
"time"
"github.com/google/uuid"
"github.com/prometheus/client_golang/prometheus"
"golang.org/x/sync/errgroup"
"golang.org/x/xerrors"
"cdr.dev/slog"
"github.com/coder/coder/coderd/database"
"github.com/coder/coder/coderd/database/dbauthz"
"github.com/coder/coder/coderd/prometheusmetrics"
"github.com/coder/coder/codersdk"
)
type Status string
const (
// Unknown should never be returned by the proxy health check.
Unknown Status = "unknown"
// Healthy means the proxy access url is reachable and returns a healthy
// status code.
Healthy Status = "ok"
// Unreachable means the proxy access url is not responding.
Unreachable Status = "unreachable"
// Unhealthy means the proxy access url is responding, but there is some
// problem with the proxy. This problem may or may not be preventing functionality.
Unhealthy Status = "unhealthy"
// Unregistered means the proxy has not registered a url yet. This means
// the proxy was created with the cli, but has not yet been started.
Unregistered Status = "unregistered"
)
type Options struct {
// Interval is the interval at which the proxy health is checked.
Interval time.Duration
DB database.Store
Logger slog.Logger
Client *http.Client
Prometheus *prometheus.Registry
}
// ProxyHealth runs a go routine that periodically checks the health of all
// workspace proxies. This information is stored in memory, so each coderd
// replica has its own view of the health of the proxies. These views should be
// consistent, and if they are not, it indicates a problem.
type ProxyHealth struct {
db database.Store
interval time.Duration
logger slog.Logger
client *http.Client
cache *atomic.Pointer[map[uuid.UUID]ProxyStatus]
// PromMetrics
healthCheckDuration prometheus.Histogram
healthCheckResults *prometheusmetrics.CachedGaugeVec
}
func New(opts *Options) (*ProxyHealth, error) {
if opts.Interval <= 0 {
opts.Interval = time.Minute
}
if opts.DB == nil {
return nil, xerrors.Errorf("db is required")
}
if opts.Prometheus == nil {
opts.Prometheus = prometheus.NewRegistry()
}
client := opts.Client
if client == nil {
client = http.DefaultClient
}
// Set a timeout on the client, so we don't wait forever for a healthz response.
tmp := *client
tmp.Timeout = time.Second * 5
client = &tmp
// Prometheus metrics
healthCheckDuration := prometheus.NewHistogram(prometheus.HistogramOpts{
Namespace: "coderd",
Subsystem: "proxyhealth",
Name: "health_check_duration_seconds",
Help: "Histogram for duration of proxy health collection in seconds.",
Buckets: []float64{0.001, 0.005, 0.010, 0.025, 0.050, 0.100, 0.500, 1, 5, 10, 30},
})
opts.Prometheus.MustRegister(healthCheckDuration)
healthCheckResults := prometheusmetrics.NewCachedGaugeVec(prometheus.NewGaugeVec(
prometheus.GaugeOpts{
Namespace: "coderd",
Subsystem: "proxyhealth",
Name: "health_check_results",
Help: "This endpoint returns a number to indicate the health status. " +
"-3 (unknown), -2 (Unreachable), -1 (Unhealthy), 0 (Unregistered), 1 (Healthy)",
}, []string{"proxy_id"}))
opts.Prometheus.MustRegister(healthCheckResults)
return &ProxyHealth{
db: opts.DB,
interval: opts.Interval,
logger: opts.Logger,
client: client,
cache: &atomic.Pointer[map[uuid.UUID]ProxyStatus]{},
healthCheckDuration: healthCheckDuration,
healthCheckResults: healthCheckResults,
}, nil
}
// Run will block until the context is canceled. It will periodically check the
// health of all proxies and store the results in the cache.
func (p *ProxyHealth) Run(ctx context.Context) {
ticker := time.NewTicker(p.interval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case now := <-ticker.C:
statuses, err := p.runOnce(ctx, now)
if err != nil {
p.logger.Error(ctx, "proxy health check failed", slog.Error(err))
continue
}
// Store the statuses in the cache.
p.cache.Store(&statuses)
}
}
}
// ForceUpdate runs a single health check and updates the cache. If the health
// check fails, the cache is not updated and an error is returned. This is useful
// to trigger an update when a proxy is created or deleted.
func (p *ProxyHealth) ForceUpdate(ctx context.Context) error {
statuses, err := p.runOnce(ctx, time.Now())
if err != nil {
return err
}
// Store the statuses in the cache.
p.cache.Store(&statuses)
return nil
}
// HealthStatus returns the current health status of all proxies stored in the
// cache.
func (p *ProxyHealth) HealthStatus() map[uuid.UUID]ProxyStatus {
ptr := p.cache.Load()
if ptr == nil {
return map[uuid.UUID]ProxyStatus{}
}
return *ptr
}
type ProxyStatus struct {
// ProxyStatus includes the value of the proxy at the time of checking. This is
// useful to know as it helps determine if the proxy checked has different values
// then the proxy in hand. AKA if the proxy was updated, and the status was for
// an older proxy.
Proxy database.WorkspaceProxy
Status Status
Report codersdk.ProxyHealthReport
CheckedAt time.Time
}
// runOnce runs the health check for all workspace proxies. If there is an
// unexpected error, an error is returned. Expected errors will mark a proxy as
// unreachable.
func (p *ProxyHealth) runOnce(ctx context.Context, now time.Time) (map[uuid.UUID]ProxyStatus, error) {
// Record from the given time.
defer p.healthCheckDuration.Observe(time.Since(now).Seconds())
//nolint:gocritic // Proxy health is a system service.
proxies, err := p.db.GetWorkspaceProxies(dbauthz.AsSystemRestricted(ctx))
if err != nil {
return nil, xerrors.Errorf("get workspace proxies: %w", err)
}
// Just use a mutex to protect map writes.
var statusMu sync.Mutex
proxyStatus := map[uuid.UUID]ProxyStatus{}
grp, gctx := errgroup.WithContext(ctx)
// Arbitrary parallelism limit.
grp.SetLimit(5)
for _, proxy := range proxies {
if proxy.Deleted {
// Ignore deleted proxies.
continue
}
// Each proxy needs to have a status set. Make a local copy for the
// call to be run async.
proxy := proxy
status := ProxyStatus{
Proxy: proxy,
CheckedAt: now,
Status: Unknown,
}
grp.Go(func() error {
if proxy.Url == "" {
// Empty URL means the proxy has not registered yet.
// When the proxy is started, it will update the url.
statusMu.Lock()
defer statusMu.Unlock()
p.healthCheckResults.WithLabelValues(prometheusmetrics.VectorOperationSet, 0, proxy.ID.String())
status.Status = Unregistered
proxyStatus[proxy.ID] = status
return nil
}
// Try to hit the healthz-report endpoint for a comprehensive health check.
reqURL := fmt.Sprintf("%s/healthz-report", strings.TrimSuffix(proxy.Url, "/"))
req, err := http.NewRequestWithContext(gctx, http.MethodGet, reqURL, nil)
if err != nil {
return xerrors.Errorf("new request: %w", err)
}
req = req.WithContext(gctx)
resp, err := p.client.Do(req)
if err == nil {
defer resp.Body.Close()
}
// A switch statement felt easier to categorize the different cases than
// if else statements or nested if statements.
switch {
case err == nil && resp.StatusCode == http.StatusOK:
err := json.NewDecoder(resp.Body).Decode(&status.Report)
if err != nil {
// If we cannot read the report, mark the proxy as unhealthy.
status.Report.Errors = []string{fmt.Sprintf("failed to decode health report: %s", err.Error())}
status.Status = Unhealthy
break
}
if len(status.Report.Errors) > 0 {
status.Status = Unhealthy
break
}
status.Status = Healthy
case err == nil && resp.StatusCode != http.StatusOK:
// Unhealthy as we did reach the proxy but it got an unexpected response.
status.Status = Unhealthy
status.Report.Errors = []string{fmt.Sprintf("unexpected status code %d", resp.StatusCode)}
case err != nil:
// Request failed, mark the proxy as unreachable.
status.Status = Unreachable
status.Report.Errors = []string{fmt.Sprintf("request to proxy failed: %s", err.Error())}
default:
// This should never happen
status.Status = Unknown
}
// Set the prometheus metric correctly.
switch status.Status {
case Healthy:
p.healthCheckResults.WithLabelValues(prometheusmetrics.VectorOperationSet, 1, proxy.ID.String())
case Unhealthy:
p.healthCheckResults.WithLabelValues(prometheusmetrics.VectorOperationSet, -1, proxy.ID.String())
case Unreachable:
p.healthCheckResults.WithLabelValues(prometheusmetrics.VectorOperationSet, -2, proxy.ID.String())
default:
// Unknown
p.healthCheckResults.WithLabelValues(prometheusmetrics.VectorOperationSet, -3, proxy.ID.String())
}
statusMu.Lock()
defer statusMu.Unlock()
proxyStatus[proxy.ID] = status
return nil
})
}
err = grp.Wait()
if err != nil {
return nil, xerrors.Errorf("group run: %w", err)
}
p.healthCheckResults.Commit()
return proxyStatus, nil
}
@@ -0,0 +1,174 @@
package proxyhealth_test
import (
"context"
"net"
"net/http"
"net/http/httptest"
"testing"
"github.com/stretchr/testify/require"
"golang.org/x/xerrors"
"cdr.dev/slog/sloggers/slogtest"
"github.com/coder/coder/coderd/database"
"github.com/coder/coder/coderd/database/dbfake"
"github.com/coder/coder/coderd/database/dbgen"
"github.com/coder/coder/coderd/httpapi"
"github.com/coder/coder/codersdk"
"github.com/coder/coder/enterprise/coderd/proxyhealth"
"github.com/coder/coder/testutil"
)
func insertProxy(t *testing.T, db database.Store, url string) database.WorkspaceProxy {
t.Helper()
ctx, cancel := context.WithTimeout(context.Background(), testutil.WaitShort)
defer cancel()
proxy, _ := dbgen.WorkspaceProxy(t, db, database.WorkspaceProxy{})
_, err := db.RegisterWorkspaceProxy(ctx, database.RegisterWorkspaceProxyParams{
Url: url,
WildcardHostname: "",
ID: proxy.ID,
})
require.NoError(t, err, "failed to update proxy")
return proxy
}
func TestProxyHealth_Unregistered(t *testing.T) {
t.Parallel()
db := dbfake.New()
proxies := []database.WorkspaceProxy{
insertProxy(t, db, ""),
insertProxy(t, db, ""),
}
ph, err := proxyhealth.New(&proxyhealth.Options{
Interval: 0,
DB: db,
Logger: slogtest.Make(t, nil),
})
require.NoError(t, err, "failed to create proxy health")
ctx, cancel := context.WithTimeout(context.Background(), testutil.WaitShort)
defer cancel()
err = ph.ForceUpdate(ctx)
require.NoError(t, err, "failed to force update")
for _, p := range proxies {
require.Equal(t, ph.HealthStatus()[p.ID].Status, proxyhealth.Unregistered, "expect unregistered proxy")
}
}
func TestProxyHealth_Unhealthy(t *testing.T) {
t.Parallel()
db := dbfake.New()
srvBadReport := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
httpapi.Write(context.Background(), w, http.StatusOK, codersdk.ProxyHealthReport{
Errors: []string{"We have a problem!"},
})
}))
defer srvBadReport.Close()
srvBadCode := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusBadRequest)
}))
defer srvBadCode.Close()
proxies := []database.WorkspaceProxy{
// Same url for both, just checking multiple proxies are checked.
insertProxy(t, db, srvBadReport.URL),
insertProxy(t, db, srvBadCode.URL),
}
ph, err := proxyhealth.New(&proxyhealth.Options{
Interval: 0,
DB: db,
Logger: slogtest.Make(t, nil),
Client: srvBadReport.Client(),
})
require.NoError(t, err, "failed to create proxy health")
ctx, cancel := context.WithTimeout(context.Background(), testutil.WaitShort)
defer cancel()
err = ph.ForceUpdate(ctx)
require.NoError(t, err, "failed to force update")
for _, p := range proxies {
require.Equal(t, ph.HealthStatus()[p.ID].Status, proxyhealth.Unhealthy, "expect reachable proxy")
}
}
func TestProxyHealth_Reachable(t *testing.T) {
t.Parallel()
db := dbfake.New()
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
httpapi.Write(context.Background(), w, http.StatusOK, codersdk.ProxyHealthReport{
Warnings: []string{"No problems, just a warning"},
})
}))
defer srv.Close()
proxies := []database.WorkspaceProxy{
// Same url for both, just checking multiple proxies are checked.
insertProxy(t, db, srv.URL),
insertProxy(t, db, srv.URL),
}
ph, err := proxyhealth.New(&proxyhealth.Options{
Interval: 0,
DB: db,
Logger: slogtest.Make(t, nil),
Client: srv.Client(),
})
require.NoError(t, err, "failed to create proxy health")
ctx, cancel := context.WithTimeout(context.Background(), testutil.WaitShort)
defer cancel()
err = ph.ForceUpdate(ctx)
require.NoError(t, err, "failed to force update")
for _, p := range proxies {
require.Equal(t, ph.HealthStatus()[p.ID].Status, proxyhealth.Healthy, "expect reachable proxy")
}
}
func TestProxyHealth_Unreachable(t *testing.T) {
t.Parallel()
db := dbfake.New()
cli := &http.Client{
Transport: &http.Transport{
DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) {
return nil, xerrors.New("Always fail")
},
},
}
proxies := []database.WorkspaceProxy{
// example.com is a real domain, but the client should always fail.
insertProxy(t, db, "https://example.com"),
insertProxy(t, db, "https://random.example.com"),
}
ph, err := proxyhealth.New(&proxyhealth.Options{
Interval: 0,
DB: db,
Logger: slogtest.Make(t, nil),
Client: cli,
})
require.NoError(t, err, "failed to create proxy health")
ctx, cancel := context.WithTimeout(context.Background(), testutil.WaitShort)
defer cancel()
err = ph.ForceUpdate(ctx)
require.NoError(t, err, "failed to force update")
for _, p := range proxies {
require.Equal(t, ph.HealthStatus()[p.ID].Status, proxyhealth.Unreachable, "expect unreachable proxy")
}
}
+53 -23
View File
@@ -1,15 +1,18 @@
package coderd
import (
"context"
"crypto/sha256"
"database/sql"
"fmt"
"net/http"
"net/url"
"time"
"github.com/google/uuid"
"golang.org/x/xerrors"
"cdr.dev/slog"
agpl "github.com/coder/coder/coderd"
"github.com/coder/coder/coderd/audit"
"github.com/coder/coder/coderd/database"
@@ -19,9 +22,18 @@ import (
"github.com/coder/coder/coderd/workspaceapps"
"github.com/coder/coder/codersdk"
"github.com/coder/coder/cryptorand"
"github.com/coder/coder/enterprise/coderd/proxyhealth"
"github.com/coder/coder/enterprise/wsproxy/wsproxysdk"
)
// forceWorkspaceProxyHealthUpdate forces an update of the proxy health.
// This is useful when a proxy is created or deleted. Errors will be logged.
func (api *API) forceWorkspaceProxyHealthUpdate(ctx context.Context) {
if err := api.proxyHealth.ForceUpdate(ctx); err != nil {
api.Logger.Error(ctx, "force proxy health update", slog.Error(err))
}
}
// @Summary Delete workspace proxy
// @ID delete-workspace-proxy
// @Security CoderSessionToken
@@ -62,6 +74,9 @@ func (api *API) deleteWorkspaceProxy(rw http.ResponseWriter, r *http.Request) {
httpapi.Write(ctx, rw, http.StatusOK, codersdk.Response{
Message: "Proxy has been deleted!",
})
// Update the proxy health cache to remove this proxy.
go api.forceWorkspaceProxyHealthUpdate(api.ctx)
}
// @Summary Create workspace proxy
@@ -122,9 +137,16 @@ func (api *API) postWorkspaceProxy(rw http.ResponseWriter, r *http.Request) {
aReq.New = proxy
httpapi.Write(ctx, rw, http.StatusCreated, codersdk.CreateWorkspaceProxyResponse{
Proxy: convertProxy(proxy),
Proxy: convertProxy(proxy, proxyhealth.ProxyStatus{
Proxy: proxy,
CheckedAt: time.Now(),
Status: proxyhealth.Unregistered,
}),
ProxyToken: fullToken,
})
// Update the proxy health cache to include this new proxy.
go api.forceWorkspaceProxyHealthUpdate(api.ctx)
}
// nolint:revive
@@ -158,28 +180,8 @@ func (api *API) workspaceProxies(rw http.ResponseWriter, r *http.Request) {
return
}
httpapi.Write(ctx, rw, http.StatusOK, convertProxies(proxies))
}
func convertProxies(p []database.WorkspaceProxy) []codersdk.WorkspaceProxy {
resp := make([]codersdk.WorkspaceProxy, 0, len(p))
for _, proxy := range p {
resp = append(resp, convertProxy(proxy))
}
return resp
}
func convertProxy(p database.WorkspaceProxy) codersdk.WorkspaceProxy {
return codersdk.WorkspaceProxy{
ID: p.ID,
Name: p.Name,
Icon: p.Icon,
URL: p.Url,
WildcardHostname: p.WildcardHostname,
CreatedAt: p.CreatedAt,
UpdatedAt: p.UpdatedAt,
Deleted: p.Deleted,
}
statues := api.proxyHealth.HealthStatus()
httpapi.Write(ctx, rw, http.StatusOK, convertProxies(proxies, statues))
}
// @Summary Issue signed workspace app token
@@ -295,6 +297,8 @@ func (api *API) workspaceProxyRegister(rw http.ResponseWriter, r *http.Request)
httpapi.Write(ctx, rw, http.StatusCreated, wsproxysdk.RegisterWorkspaceProxyResponse{
AppSecurityKey: api.AppSecurityKey.String(),
})
go api.forceWorkspaceProxyHealthUpdate(api.ctx)
}
// reconnectingPTYSignedToken issues a signed app token for use when connecting
@@ -392,3 +396,29 @@ func (api *API) reconnectingPTYSignedToken(rw http.ResponseWriter, r *http.Reque
SignedToken: tokenStr,
})
}
func convertProxies(p []database.WorkspaceProxy, statuses map[uuid.UUID]proxyhealth.ProxyStatus) []codersdk.WorkspaceProxy {
resp := make([]codersdk.WorkspaceProxy, 0, len(p))
for _, proxy := range p {
resp = append(resp, convertProxy(proxy, statuses[proxy.ID]))
}
return resp
}
func convertProxy(p database.WorkspaceProxy, status proxyhealth.ProxyStatus) codersdk.WorkspaceProxy {
return codersdk.WorkspaceProxy{
ID: p.ID,
Name: p.Name,
Icon: p.Icon,
URL: p.Url,
WildcardHostname: p.WildcardHostname,
CreatedAt: p.CreatedAt,
UpdatedAt: p.UpdatedAt,
Deleted: p.Deleted,
Status: codersdk.WorkspaceProxyStatus{
Status: codersdk.ProxyHealthStatus(status.Status),
Report: status.Report,
CheckedAt: status.CheckedAt,
},
}
}
+1 -1
View File
@@ -60,7 +60,7 @@ func TestWorkspaceProxyCRUD(t *testing.T) {
proxies, err := client.WorkspaceProxies(ctx)
require.NoError(t, err)
require.Len(t, proxies, 1)
require.Equal(t, proxyRes.Proxy, proxies[0])
require.Equal(t, proxyRes.Proxy.ID, proxies[0].ID)
require.NotEmpty(t, proxyRes.ProxyToken)
})