feat: add health check monitoring to workspace apps (#4114)

2026-09-24 15:04:27 +08:00 · 2022-09-23 15:51:04 -04:00
parent f160830226
commit 4c8be34d81
64 changed files with 1592 additions and 509 deletions
@@ -0,0 +1,155 @@
+package codersdk
+
+import (
+	"context"
+	"encoding/binary"
+	"encoding/json"
+	"net"
+	"net/netip"
+	"strconv"
+	"time"
+
+	"golang.org/x/crypto/ssh"
+	"golang.org/x/xerrors"
+	"tailscale.com/ipn/ipnstate"
+	"tailscale.com/net/speedtest"
+	"tailscale.com/tailcfg"
+
+	"github.com/coder/coder/tailnet"
+)
+
+var (
+	// TailnetIP is a static IPv6 address with the Tailscale prefix that is used to route
+	// connections from clients to this node. A dynamic address is not required because a Tailnet
+	// client only dials a single agent at a time.
+	TailnetIP                  = netip.MustParseAddr("fd7a:115c:a1e0:49d6:b259:b7ac:b1b2:48f4")
+	TailnetSSHPort             = 1
+	TailnetReconnectingPTYPort = 2
+	TailnetSpeedtestPort       = 3
+)
+
+// ReconnectingPTYRequest is sent from the client to the server
+// to pipe data to a PTY.
+// @typescript-ignore ReconnectingPTYRequest
+type ReconnectingPTYRequest struct {
+	Data   string `json:"data"`
+	Height uint16 `json:"height"`
+	Width  uint16 `json:"width"`
+}
+
+// @typescript-ignore AgentConn
+type AgentConn struct {
+	*tailnet.Conn
+	CloseFunc func()
+}
+
+func (c *AgentConn) Ping() (time.Duration, error) {
+	errCh := make(chan error, 1)
+	durCh := make(chan time.Duration, 1)
+	c.Conn.Ping(TailnetIP, tailcfg.PingICMP, func(pr *ipnstate.PingResult) {
+		if pr.Err != "" {
+			errCh <- xerrors.New(pr.Err)
+			return
+		}
+		durCh <- time.Duration(pr.LatencySeconds * float64(time.Second))
+	})
+	select {
+	case err := <-errCh:
+		return 0, err
+	case dur := <-durCh:
+		return dur, nil
+	}
+}
+
+func (c *AgentConn) CloseWithError(_ error) error {
+	return c.Close()
+}
+
+func (c *AgentConn) Close() error {
+	if c.CloseFunc != nil {
+		c.CloseFunc()
+	}
+	return c.Conn.Close()
+}
+
+// @typescript-ignore ReconnectingPTYInit
+type ReconnectingPTYInit struct {
+	ID      string
+	Height  uint16
+	Width   uint16
+	Command string
+}
+
+func (c *AgentConn) ReconnectingPTY(id string, height, width uint16, command string) (net.Conn, error) {
+	conn, err := c.DialContextTCP(context.Background(), netip.AddrPortFrom(TailnetIP, uint16(TailnetReconnectingPTYPort)))
+	if err != nil {
+		return nil, err
+	}
+	data, err := json.Marshal(ReconnectingPTYInit{
+		ID:      id,
+		Height:  height,
+		Width:   width,
+		Command: command,
+	})
+	if err != nil {
+		_ = conn.Close()
+		return nil, err
+	}
+	data = append(make([]byte, 2), data...)
+	binary.LittleEndian.PutUint16(data, uint16(len(data)-2))
+
+	_, err = conn.Write(data)
+	if err != nil {
+		_ = conn.Close()
+		return nil, err
+	}
+	return conn, nil
+}
+
+func (c *AgentConn) SSH() (net.Conn, error) {
+	return c.DialContextTCP(context.Background(), netip.AddrPortFrom(TailnetIP, uint16(TailnetSSHPort)))
+}
+
+// SSHClient calls SSH to create a client that uses a weak cipher
+// for high throughput.
+func (c *AgentConn) SSHClient() (*ssh.Client, error) {
+	netConn, err := c.SSH()
+	if err != nil {
+		return nil, xerrors.Errorf("ssh: %w", err)
+	}
+	sshConn, channels, requests, err := ssh.NewClientConn(netConn, "localhost:22", &ssh.ClientConfig{
+		// SSH host validation isn't helpful, because obtaining a peer
+		// connection already signifies user-intent to dial a workspace.
+		// #nosec
+		HostKeyCallback: ssh.InsecureIgnoreHostKey(),
+	})
+	if err != nil {
+		return nil, xerrors.Errorf("ssh conn: %w", err)
+	}
+	return ssh.NewClient(sshConn, channels, requests), nil
+}
+
+func (c *AgentConn) Speedtest(direction speedtest.Direction, duration time.Duration) ([]speedtest.Result, error) {
+	speedConn, err := c.DialContextTCP(context.Background(), netip.AddrPortFrom(TailnetIP, uint16(TailnetSpeedtestPort)))
+	if err != nil {
+		return nil, xerrors.Errorf("dial speedtest: %w", err)
+	}
+	results, err := speedtest.RunClientWithConn(direction, duration, speedConn)
+	if err != nil {
+		return nil, xerrors.Errorf("run speedtest: %w", err)
+	}
+	return results, err
+}
+
+func (c *AgentConn) DialContext(ctx context.Context, network string, addr string) (net.Conn, error) {
+	if network == "unix" {
+		return nil, xerrors.New("network must be tcp or udp")
+	}
+	_, rawPort, _ := net.SplitHostPort(addr)
+	port, _ := strconv.Atoi(rawPort)
+	ipp := netip.AddrPortFrom(TailnetIP, uint16(port))
+	if network == "udp" {
+		return c.Conn.DialContextUDP(ctx, ipp)
+	}
+	return c.Conn.DialContextTCP(ctx, ipp)
+}
@@ -21,20 +21,22 @@ import (

 	"cdr.dev/slog"

-	"github.com/coder/coder/agent"
 	"github.com/coder/coder/tailnet"
 	"github.com/coder/retry"
 )

+// @typescript-ignore GoogleInstanceIdentityToken
 type GoogleInstanceIdentityToken struct {
 	JSONWebToken string `json:"json_web_token" validate:"required"`
 }

+// @typescript-ignore AWSInstanceIdentityToken
 type AWSInstanceIdentityToken struct {
 	Signature string `json:"signature" validate:"required"`
 	Document  string `json:"document" validate:"required"`
 }

+// @typescript-ignore ReconnectingPTYRequest
 type AzureInstanceIdentityToken struct {
 	Signature string `json:"signature" validate:"required"`
 	Encoding  string `json:"encoding" validate:"required"`
@@ -42,20 +44,31 @@ type AzureInstanceIdentityToken struct {

 // WorkspaceAgentAuthenticateResponse is returned when an instance ID
 // has been exchanged for a session token.
+// @typescript-ignore WorkspaceAgentAuthenticateResponse
 type WorkspaceAgentAuthenticateResponse struct {
 	SessionToken string `json:"session_token"`
 }

 // WorkspaceAgentConnectionInfo returns required information for establishing
 // a connection with a workspace.
+// @typescript-ignore WorkspaceAgentConnectionInfo
 type WorkspaceAgentConnectionInfo struct {
 	DERPMap *tailcfg.DERPMap `json:"derp_map"`
 }

+// @typescript-ignore PostWorkspaceAgentVersionRequest
 type PostWorkspaceAgentVersionRequest struct {
 	Version string `json:"version"`
 }

+// @typescript-ignore WorkspaceAgentMetadata
+type WorkspaceAgentMetadata struct {
+	DERPMap              *tailcfg.DERPMap  `json:"derpmap"`
+	EnvironmentVariables map[string]string `json:"environment_variables"`
+	StartupScript        string            `json:"startup_script"`
+	Directory            string            `json:"directory"`
+}
+
 // AuthWorkspaceGoogleInstanceIdentity uses the Google Compute Engine Metadata API to
 // fetch a signed JWT, and exchange it for a session token for a workspace agent.
 //
@@ -185,16 +198,16 @@ func (c *Client) AuthWorkspaceAzureInstanceIdentity(ctx context.Context) (Worksp
 }

 // WorkspaceAgentMetadata fetches metadata for the currently authenticated workspace agent.
-func (c *Client) WorkspaceAgentMetadata(ctx context.Context) (agent.Metadata, error) {
+func (c *Client) WorkspaceAgentMetadata(ctx context.Context) (WorkspaceAgentMetadata, error) {
 	res, err := c.Request(ctx, http.MethodGet, "/api/v2/workspaceagents/me/metadata", nil)
 	if err != nil {
-		return agent.Metadata{}, err
+		return WorkspaceAgentMetadata{}, err
 	}
 	defer res.Body.Close()
 	if res.StatusCode != http.StatusOK {
-		return agent.Metadata{}, readBodyAsError(res)
+		return WorkspaceAgentMetadata{}, readBodyAsError(res)
 	}
-	var agentMetadata agent.Metadata
+	var agentMetadata WorkspaceAgentMetadata
 	return agentMetadata, json.NewDecoder(res.Body).Decode(&agentMetadata)
 }

@@ -228,7 +241,7 @@ func (c *Client) ListenWorkspaceAgentTailnet(ctx context.Context) (net.Conn, err
 	return websocket.NetConn(ctx, conn, websocket.MessageBinary), nil
 }

-func (c *Client) DialWorkspaceAgentTailnet(ctx context.Context, logger slog.Logger, agentID uuid.UUID) (*agent.Conn, error) {
+func (c *Client) DialWorkspaceAgentTailnet(ctx context.Context, logger slog.Logger, agentID uuid.UUID) (*AgentConn, error) {
 	res, err := c.Request(ctx, http.MethodGet, fmt.Sprintf("/api/v2/workspaceagents/%s/connection", agentID), nil)
 	if err != nil {
 		return nil, err
@@ -325,7 +338,7 @@ func (c *Client) DialWorkspaceAgentTailnet(ctx context.Context, logger slog.Logg
 		_ = conn.Close()
 		return nil, err
 	}
-	return &agent.Conn{
+	return &AgentConn{
 		Conn: conn,
 		CloseFunc: func() {
 			cancelFunc()
@@ -348,6 +361,34 @@ func (c *Client) WorkspaceAgent(ctx context.Context, id uuid.UUID) (WorkspaceAge
 	return workspaceAgent, json.NewDecoder(res.Body).Decode(&workspaceAgent)
 }

+// MyWorkspaceAgent returns the requesting agent.
+func (c *Client) WorkspaceAgentApps(ctx context.Context) ([]WorkspaceApp, error) {
+	res, err := c.Request(ctx, http.MethodGet, "/api/v2/workspaceagents/me/apps", nil)
+	if err != nil {
+		return nil, err
+	}
+	defer res.Body.Close()
+	if res.StatusCode != http.StatusOK {
+		return nil, readBodyAsError(res)
+	}
+	var workspaceApps []WorkspaceApp
+	return workspaceApps, json.NewDecoder(res.Body).Decode(&workspaceApps)
+}
+
+// PostWorkspaceAgentAppHealth updates the workspace agent app health status.
+func (c *Client) PostWorkspaceAgentAppHealth(ctx context.Context, req PostWorkspaceAppHealthsRequest) error {
+	res, err := c.Request(ctx, http.MethodPost, "/api/v2/workspaceagents/me/app-health", req)
+	if err != nil {
+		return err
+	}
+	defer res.Body.Close()
+	if res.StatusCode != http.StatusOK {
+		return readBodyAsError(res)
+	}
+
+	return nil
+}
+
 func (c *Client) PostWorkspaceAgentVersion(ctx context.Context, version string) error {
 	// Phone home and tell the mothership what version we're on.
 	versionReq := PostWorkspaceAgentVersionRequest{Version: version}
@@ -392,12 +433,22 @@ func (c *Client) WorkspaceAgentReconnectingPTY(ctx context.Context, agentID, rec
 	return websocket.NetConn(ctx, conn, websocket.MessageBinary), nil
 }

+// Stats records the Agent's network connection statistics for use in
+// user-facing metrics and debugging.
+// Each member value must be written and read with atomic.
+// @typescript-ignore AgentStats
+type AgentStats struct {
+	NumConns int64 `json:"num_comms"`
+	RxBytes  int64 `json:"rx_bytes"`
+	TxBytes  int64 `json:"tx_bytes"`
+}
+
 // AgentReportStats begins a stat streaming connection with the Coder server.
 // It is resilient to network failures and intermittent coderd issues.
 func (c *Client) AgentReportStats(
 	ctx context.Context,
 	log slog.Logger,
-	stats func() *agent.Stats,
+	stats func() *AgentStats,
 ) (io.Closer, error) {
 	serverURL, err := c.URL.Parse("/api/v2/workspaceagents/me/report-stats")
 	if err != nil {
@@ -4,6 +4,15 @@ import (
 	"github.com/google/uuid"
 )

+type WorkspaceAppHealth string
+
+const (
+	WorkspaceAppHealthDisabled     WorkspaceAppHealth = "disabled"
+	WorkspaceAppHealthInitializing WorkspaceAppHealth = "initializing"
+	WorkspaceAppHealthHealthy      WorkspaceAppHealth = "healthy"
+	WorkspaceAppHealthUnhealthy    WorkspaceAppHealth = "unhealthy"
+)
+
 type WorkspaceApp struct {
 	ID uuid.UUID `json:"id"`
 	// Name is a unique identifier attached to an agent.
@@ -12,4 +21,22 @@ type WorkspaceApp struct {
 	// Icon is a relative path or external URL that specifies
 	// an icon to be displayed in the dashboard.
 	Icon string `json:"icon,omitempty"`
+	// Healthcheck specifies the configuration for checking app health.
+	Healthcheck Healthcheck        `json:"healthcheck"`
+	Health      WorkspaceAppHealth `json:"health"`
+}
+
+type Healthcheck struct {
+	// URL specifies the url to check for the app health.
+	URL string `json:"url"`
+	// Interval specifies the seconds between each health check.
+	Interval int32 `json:"interval"`
+	// Threshold specifies the number of consecutive failed health checks before returning "unhealthy".
+	Threshold int32 `json:"threshold"`
+}
+
+// @typescript-ignore PostWorkspaceAppHealthsRequest
+type PostWorkspaceAppHealthsRequest struct {
+	// Healths is a map of the workspace app name and the health of the app.
+	Healths map[string]WorkspaceAppHealth
 }