package coderd

import (

"bytes"

"context"

"database/sql"

"encoding/json"

"errors"

"fmt"

"io"

"net/http"

"net/url"

"slices"

"sort"

"strconv"

"strings"

"time"

"github.com/go-chi/chi/v5"

"github.com/google/uuid"

"github.com/sqlc-dev/pqtype"

"golang.org/x/exp/maps"

"golang.org/x/sync/errgroup"

"golang.org/x/xerrors"

"tailscale.com/tailcfg"

"cdr.dev/slog/v3"

"github.com/coder/coder/v2/coderd/agentapi"

"github.com/coder/coder/v2/coderd/agentapi/metadatabatcher"

"github.com/coder/coder/v2/coderd/database"

"github.com/coder/coder/v2/coderd/database/db2sdk"

"github.com/coder/coder/v2/coderd/database/dbauthz"

"github.com/coder/coder/v2/coderd/database/dbtime"

"github.com/coder/coder/v2/coderd/externalauth"

"github.com/coder/coder/v2/coderd/httpapi"

"github.com/coder/coder/v2/coderd/httpmw"

"github.com/coder/coder/v2/coderd/httpmw/loggermw"

"github.com/coder/coder/v2/coderd/jwtutils"

"github.com/coder/coder/v2/coderd/prebuilds"

"github.com/coder/coder/v2/coderd/rbac"

"github.com/coder/coder/v2/coderd/rbac/policy"

"github.com/coder/coder/v2/coderd/telemetry"

maputil "github.com/coder/coder/v2/coderd/util/maps"

"github.com/coder/coder/v2/coderd/wspubsub"

"github.com/coder/coder/v2/coderd/x/chatd"

"github.com/coder/coder/v2/coderd/x/chatd/chatprompt"

"github.com/coder/coder/v2/coderd/x/gitsync"

"github.com/coder/coder/v2/codersdk"

"github.com/coder/coder/v2/codersdk/agentsdk"

"github.com/coder/coder/v2/codersdk/workspacesdk"

"github.com/coder/coder/v2/codersdk/wsjson"

"github.com/coder/coder/v2/tailnet"

"github.com/coder/coder/v2/tailnet/proto"

"github.com/coder/websocket"

)

// @Summary Get workspace agent by ID

// @ID get-workspace-agent-by-id

// @Security CoderSessionToken

// @Produce json

// @Tags Agents

// @Param workspaceagent path string true "Workspace agent ID" format(uuid)

// @Success 200 {object} codersdk.WorkspaceAgent

// @Router /workspaceagents/{workspaceagent} [get]

func (api *API) workspaceAgent(rw http.ResponseWriter, r *http.Request) {

var (

ctx = r.Context()

waws = httpmw.WorkspaceAgentAndWorkspaceParam(r)

dbApps []database.WorkspaceApp

scripts []database.WorkspaceAgentScript

logSources []database.WorkspaceAgentLogSource

)

var eg errgroup.Group

eg.Go(func() (err error) {

dbApps, err = api.Database.GetWorkspaceAppsByAgentID(ctx, waws.WorkspaceAgent.ID)

return err

})

eg.Go(func() (err error) {

//nolint:gocritic // TODO: can we make this not require system restricted?

scripts, err = api.Database.GetWorkspaceAgentScriptsByAgentIDs(dbauthz.AsSystemRestricted(ctx), []uuid.UUID{waws.WorkspaceAgent.ID})

return err

})

eg.Go(func() (err error) {

//nolint:gocritic // TODO: can we make this not require system restricted?

logSources, err = api.Database.GetWorkspaceAgentLogSourcesByAgentIDs(dbauthz.AsSystemRestricted(ctx), []uuid.UUID{waws.WorkspaceAgent.ID})

return err

})

err := eg.Wait()

if httpapi.Is404Error(err) {

httpapi.ResourceNotFound(rw)

return

}

if err != nil {

httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{

Message: "Internal error fetching workspace agent.",

Detail: err.Error(),

})

return

}

appIDs := []uuid.UUID{}

for _, app := range dbApps {

appIDs = append(appIDs, app.ID)

}

// nolint:gocritic // This is a system restricted operation.

statuses, err := api.Database.GetWorkspaceAppStatusesByAppIDs(dbauthz.AsSystemRestricted(ctx), appIDs)

if err != nil {

httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{

Message: "Internal error fetching workspace app statuses.",

Detail: err.Error(),

})

return

}

apiAgent, err := db2sdk.WorkspaceAgent(

api.DERPMap(), *api.TailnetCoordinator.Load(), waws.WorkspaceAgent, db2sdk.Apps(dbApps, statuses, waws.WorkspaceAgent, waws.OwnerUsername, waws.WorkspaceTable), convertScripts(scripts), convertLogSources(logSources), api.AgentInactiveDisconnectTimeout,

api.DeploymentValues.AgentFallbackTroubleshootingURL.String(),

)

if err != nil {

httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{

Message: "Internal error reading workspace agent.",

Detail: err.Error(),

})

return

}

httpapi.Write(ctx, rw, http.StatusOK, apiAgent)

}

const AgentAPIVersionREST = "1.0"

// @Summary Patch workspace agent logs

// @ID patch-workspace-agent-logs

// @Security CoderSessionToken

// @Accept json

// @Produce json

// @Tags Agents

// @Param request body agentsdk.PatchLogs true "logs"

// @Success 200 {object} codersdk.Response

// @Router /workspaceagents/me/logs [patch]

func (api *API) patchWorkspaceAgentLogs(rw http.ResponseWriter, r *http.Request) {

ctx := r.Context()

workspaceAgent := httpmw.WorkspaceAgent(r)

var req agentsdk.PatchLogs

if !httpapi.Read(ctx, rw, r, &req) {

return

}

if len(req.Logs) == 0 {

httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{

Message: "No logs provided.",

})

return

}

// This is to support the legacy API where the log source ID was

// not provided in the request body. We default to the external

// log source in this case.

if req.LogSourceID == uuid.Nil {

// Use the external log source

externalSources, err := api.Database.InsertWorkspaceAgentLogSources(ctx, database.InsertWorkspaceAgentLogSourcesParams{

WorkspaceAgentID: workspaceAgent.ID,

CreatedAt: dbtime.Now(),

ID: []uuid.UUID{agentsdk.ExternalLogSourceID},

DisplayName: []string{"External"},

Icon: []string{"/emojis/1f310.png"},

})

if database.IsUniqueViolation(err, database.UniqueWorkspaceAgentLogSourcesPkey) {

err = nil

req.LogSourceID = agentsdk.ExternalLogSourceID

}

if err != nil {

httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{

Message: "Failed to create external log source.",

Detail: err.Error(),

})

return

}

if len(externalSources) == 1 {

req.LogSourceID = externalSources[0].ID

}

}

output := make([]string, 0)

level := make([]database.LogLevel, 0)

outputLength := 0

for _, logEntry := range req.Logs {

sanitizedOutput := agentsdk.SanitizeLogOutput(logEntry.Output)

output = append(output, sanitizedOutput)

outputLength += len(sanitizedOutput)

if logEntry.Level == "" {

// Default to "info" to support older agents that didn't have the level field.

logEntry.Level = codersdk.LogLevelInfo

}

parsedLevel := database.LogLevel(logEntry.Level)

if !parsedLevel.Valid() {

httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{

Message: "Invalid log level provided.",

Detail: fmt.Sprintf("invalid log level: %q", logEntry.Level),

})

return

}

level = append(level, parsedLevel)

}

logs, err := api.Database.InsertWorkspaceAgentLogs(ctx, database.InsertWorkspaceAgentLogsParams{

AgentID: workspaceAgent.ID,

CreatedAt: dbtime.Now(),

Output: output,

Level: level,

LogSourceID: req.LogSourceID,

// #nosec G115 - Log output length is limited and fits in int32

OutputLength: int32(outputLength),

})

if err != nil {

if !database.IsWorkspaceAgentLogsLimitError(err) {

httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{

Message: "Failed to upload logs",

Detail: err.Error(),

})

return

}

if workspaceAgent.LogsOverflowed {

httpapi.Write(ctx, rw, http.StatusRequestEntityTooLarge, codersdk.Response{

Message: "Logs limit exceeded",

Detail: err.Error(),

})

return

}

err := api.Database.UpdateWorkspaceAgentLogOverflowByID(ctx, database.UpdateWorkspaceAgentLogOverflowByIDParams{

ID: workspaceAgent.ID,

LogsOverflowed: true,

})

if err != nil {

// We don't want to return here, because the agent will retry

// on failure and this isn't a huge deal. The overflow state

// is just a hint to the user that the logs are incomplete.

api.Logger.Warn(ctx, "failed to update workspace agent log overflow", slog.Error(err))

}

workspace, err := api.Database.GetWorkspaceByAgentID(ctx, workspaceAgent.ID)

if err != nil {

httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{

Message: "Failed to get workspace.",

Detail: err.Error(),

})

return

}

api.publishWorkspaceUpdate(ctx, workspace.OwnerID, wspubsub.WorkspaceEvent{

Kind: wspubsub.WorkspaceEventKindAgentLogsOverflow,

WorkspaceID: workspace.ID,

AgentID: &workspaceAgent.ID,

})

httpapi.Write(ctx, rw, http.StatusRequestEntityTooLarge, codersdk.Response{

Message: "Logs limit exceeded",

})

return

}

lowestLogID := logs[0].ID

// Publish by the lowest log ID inserted so the

// log stream will fetch everything from that point.

api.publishWorkspaceAgentLogsUpdate(ctx, workspaceAgent.ID, agentsdk.LogsNotifyMessage{

CreatedAfter: lowestLogID - 1,

})

if workspaceAgent.LogsLength == 0 {

// If these are the first logs being appended, we publish a UI update

// to notify the UI that logs are now available.

workspace, err := api.Database.GetWorkspaceByAgentID(ctx, workspaceAgent.ID)

if err != nil {

httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{

Message: "Failed to get workspace.",

Detail: err.Error(),

})

return

}

api.publishWorkspaceUpdate(ctx, workspace.OwnerID, wspubsub.WorkspaceEvent{

Kind: wspubsub.WorkspaceEventKindAgentFirstLogs,

WorkspaceID: workspace.ID,

AgentID: &workspaceAgent.ID,

})

}

httpapi.Write(ctx, rw, http.StatusOK, nil)

}

// @Summary Patch workspace agent app status

// @ID patch-workspace-agent-app-status

// @Security CoderSessionToken

// @Accept json

// @Produce json

// @Tags Agents

// @Param request body agentsdk.PatchAppStatus true "app status"

// @Success 200 {object} codersdk.Response

// @Router /workspaceagents/me/app-status [patch]

// @Deprecated Use UpdateAppStatus on the Agent API instead.

func (api *API) patchWorkspaceAgentAppStatus(rw http.ResponseWriter, r *http.Request) {

ctx := r.Context()

workspaceAgent := httpmw.WorkspaceAgent(r)

var req agentsdk.PatchAppStatus

if !httpapi.Read(ctx, rw, r, &req) {

return

}

workspace, err := api.Database.GetWorkspaceByAgentID(ctx, workspaceAgent.ID)

if err != nil {

httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{

Message: "Failed to get workspace.",

Detail: err.Error(),

})

return

}

// This functionality has been moved to the AppsAPI in the agentapi. We keep this HTTP handler around for back

// compatibility with older agents. We'll translate the request into the protobuf so there is only one primary

// implementation.

cachedWs := &agentapi.CachedWorkspaceFields{}

cachedWs.UpdateValues(workspace)

appAPI := &agentapi.AppsAPI{

AgentID: workspaceAgent.ID,

Database: api.Database,

Log: api.Logger,

Workspace: cachedWs,

AgentFn: func(ctx context.Context) (database.WorkspaceAgent, error) {

return api.Database.GetWorkspaceAgentByID(ctx, workspaceAgent.ID)

},

PublishWorkspaceUpdateFn: func(ctx context.Context, agentID uuid.UUID, kind wspubsub.WorkspaceEventKind) error {

api.publishWorkspaceUpdate(ctx, workspace.OwnerID, wspubsub.WorkspaceEvent{

Kind: kind,

WorkspaceID: workspace.ID,

AgentID: &agentID,

})

return nil

},

NotificationsEnqueuer: api.NotificationsEnqueuer,

Clock: api.Clock,

}

protoReq, err := agentsdk.ProtoFromPatchAppStatus(req)

if err != nil {

httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{

Message: "Failed to parse request.",

Detail: err.Error(),

})

return

}

_, err = appAPI.UpdateAppStatus(r.Context(), protoReq)

if err != nil {

sdkErr := new(codersdk.Error)

if xerrors.As(err, &sdkErr) {

httpapi.Write(ctx, rw, sdkErr.StatusCode(), sdkErr.Response)

return

}

httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{

Message: "Failed to update app status.",

Detail: err.Error(),

})

return

}

httpapi.Write(ctx, rw, http.StatusOK, nil)

}

// workspaceAgentLogs returns the logs associated with a workspace agent

//

// @Summary Get logs by workspace agent

// @ID get-logs-by-workspace-agent

// @Security CoderSessionToken

// @Produce json

// @Tags Agents

// @Param workspaceagent path string true "Workspace agent ID" format(uuid)

// @Param before query int false "Before log id"

// @Param after query int false "After log id"

// @Param follow query bool false "Follow log stream"

// @Param no_compression query bool false "Disable compression for WebSocket connection"

// @Param format query string false "Log output format. Accepted: 'json' (default), 'text' (plain text with RFC3339 timestamps and ANSI colors). Not supported with follow=true." Enums(json,text)

// @Success 200 {array} codersdk.WorkspaceAgentLog

// @Router /workspaceagents/{workspaceagent}/logs [get]

func (api *API) workspaceAgentLogs(rw http.ResponseWriter, r *http.Request) {

// This mostly copies how provisioner job logs are streamed!

var (

ctx = r.Context()

waws = httpmw.WorkspaceAgentAndWorkspaceParam(r)

logger = api.Logger.With(slog.F("workspace_agent_id", waws.WorkspaceAgent.ID))

follow = r.URL.Query().Has("follow")

afterRaw = r.URL.Query().Get("after")

noCompression = r.URL.Query().Has("no_compression")

format = r.URL.Query().Get("format")

)

// Validate format parameter.

if format == "" {

format = "json"

}

if format != "json" && format != "text" {

httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{

Message: "Invalid format parameter.",

Detail: "Allowed values are \"json\" and \"text\".",

})

return

}

// Text format is not supported with streaming.

if format == "text" && follow {

httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{

Message: "Text format is not supported with follow mode.",

Detail: "Use format=json or omit the follow parameter.",

})

return

}

var after int64

// Only fetch logs created after the time provided.

if afterRaw != "" {

var err error

after, err = strconv.ParseInt(afterRaw, 10, 64)

if err != nil || after < 0 {

httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{

Message: "Query param \"after\" must be an integer greater than or equal to zero.",

Validations: []codersdk.ValidationError{

{Field: "after", Detail: "Must be an integer greater than or equal to zero"},

},

})

return

}

}

logs, err := api.Database.GetWorkspaceAgentLogsAfter(ctx, database.GetWorkspaceAgentLogsAfterParams{

AgentID: waws.WorkspaceAgent.ID,

CreatedAfter: after,

})

if errors.Is(err, sql.ErrNoRows) {

err = nil

}

if err != nil {

httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{

Message: "Internal error fetching provisioner logs.",

Detail: err.Error(),

})

return

}

if logs == nil {

logs = []database.WorkspaceAgentLog{}

}

if !follow {

if format == "text" {

sids, err := api.Database.GetWorkspaceAgentLogSourcesByAgentIDs(ctx, []uuid.UUID{waws.WorkspaceAgent.ID})

if err != nil {

httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{

Message: "Internal error fetching workspace agent log sources.",

Detail: err.Error(),

})

return

}

lsids := make(map[uuid.UUID]string, len(sids))

for _, sid := range sids {

lsids[sid.ID] = sid.DisplayName

}

rw.Header().Set("Content-Type", "text/plain; charset=utf-8")

rw.WriteHeader(http.StatusOK)

for _, log := range logs {

_, _ = rw.Write([]byte(db2sdk.WorkspaceAgentLog(log).Text(waws.WorkspaceAgent.Name, lsids[log.LogSourceID])))

_, _ = rw.Write([]byte("\n"))

}

return

}

httpapi.Write(ctx, rw, http.StatusOK, convertWorkspaceAgentLogs(logs))

return

}

api.WebsocketWaitMutex.Lock()

api.WebsocketWaitGroup.Add(1)

api.WebsocketWaitMutex.Unlock()

defer api.WebsocketWaitGroup.Done()

opts := &websocket.AcceptOptions{}

// Allow client to request no compression. This is useful for buggy

// clients or if there's a client/server incompatibility. This is

// needed with e.g. coder/websocket and Safari (confirmed in 16.5).

//

// See:

// * https://github.com/nhooyr/websocket/issues/218

// * https://github.com/gobwas/ws/issues/169

if noCompression {

opts.CompressionMode = websocket.CompressionDisabled

}

conn, err := websocket.Accept(rw, r, opts)

if err != nil {

httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{

Message: "Failed to accept websocket.",

Detail: err.Error(),

})

return

}

ctx, cancel := context.WithCancel(ctx)

defer cancel()

go httpapi.HeartbeatClose(ctx, api.Logger, cancel, conn)

encoder := wsjson.NewEncoder[[]codersdk.WorkspaceAgentLog](conn, websocket.MessageText)

defer encoder.Close(websocket.StatusNormalClosure)

err = encoder.Encode(convertWorkspaceAgentLogs(logs))

if err != nil {

return

}

lastSentLogID := after

if len(logs) > 0 {

lastSentLogID = logs[len(logs)-1].ID

}

workspaceNotifyCh := make(chan struct{}, 1)

notifyCh := make(chan struct{}, 1)

// Allow us to immediately check if we missed any logs

// between initial fetch and subscribe.

notifyCh <- struct{}{}

// Subscribe to workspace to detect new builds.

closeSubscribeWorkspace, err := api.Pubsub.SubscribeWithErr(wspubsub.WorkspaceEventChannel(waws.WorkspaceTable.OwnerID),

wspubsub.HandleWorkspaceEvent(

func(_ context.Context, e wspubsub.WorkspaceEvent, err error) {

if err != nil {

return

}

if e.Kind == wspubsub.WorkspaceEventKindStateChange && e.WorkspaceID == waws.WorkspaceTable.ID {

select {

case workspaceNotifyCh <- struct{}{}:

default:

}

}

}))

if err != nil {

httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{

Message: "Failed to subscribe to workspace for log streaming.",

Detail: err.Error(),

})

return

}

defer closeSubscribeWorkspace()

// Subscribe early to prevent missing log events.

closeSubscribe, err := api.Pubsub.Subscribe(agentsdk.LogsNotifyChannel(waws.WorkspaceAgent.ID), func(_ context.Context, _ []byte) {

// The message is not important, we're tracking lastSentLogID manually.

select {

case notifyCh <- struct{}{}:

default:

}

})

if err != nil {

httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{

Message: "Failed to subscribe to agent for log streaming.",

Detail: err.Error(),

})

return

}

defer closeSubscribe()

// Buffer size controls the log prefetch capacity.

bufferedLogs := make(chan []database.WorkspaceAgentLog, 8)

// Check at least once per minute in case we didn't receive a pubsub message.

recheckInterval := time.Minute

t := time.NewTicker(recheckInterval)

defer t.Stop()

// Log the request immediately instead of after it completes.

if rl := loggermw.RequestLoggerFromContext(ctx); rl != nil {

rl.WriteLog(ctx, http.StatusAccepted)

}

go func() {

defer func() {

logger.Debug(ctx, "end log streaming loop")

close(bufferedLogs)

}()

logger.Debug(ctx, "start log streaming loop", slog.F("last_sent_log_id", lastSentLogID))

keepGoing := true

for keepGoing {

var (

debugTriggeredBy string

onlyCheckLatestBuild bool

)

select {

case <-ctx.Done():

return

case <-t.C:

debugTriggeredBy = "timer"

case <-workspaceNotifyCh:

debugTriggeredBy = "workspace"

onlyCheckLatestBuild = true

case <-notifyCh:

debugTriggeredBy = "log"

t.Reset(recheckInterval)

}

agents, err := api.Database.GetWorkspaceAgentsInLatestBuildByWorkspaceID(ctx, waws.WorkspaceTable.ID)

if err != nil && !xerrors.Is(err, sql.ErrNoRows) {

if xerrors.Is(err, context.Canceled) {

return

}

logger.Warn(ctx, "failed to get workspace agents in latest build", slog.Error(err))

continue

}

// If the agent is no longer in the latest build, we can stop after

// checking once.

keepGoing = slices.ContainsFunc(agents, func(agent database.WorkspaceAgent) bool { return agent.ID == waws.WorkspaceAgent.ID })

logger.Debug(

ctx,

"checking for new logs",

slog.F("triggered_by", debugTriggeredBy),

slog.F("only_check_latest_build", onlyCheckLatestBuild),

slog.F("keep_going", keepGoing),

slog.F("last_sent_log_id", lastSentLogID),

slog.F("workspace_has_agents", len(agents) > 0),

)

if onlyCheckLatestBuild && keepGoing {

continue

}

logs, err := api.Database.GetWorkspaceAgentLogsAfter(ctx, database.GetWorkspaceAgentLogsAfterParams{

AgentID: waws.WorkspaceAgent.ID,

CreatedAfter: lastSentLogID,

})

if err != nil {

if xerrors.Is(err, context.Canceled) {

return

}

logger.Warn(ctx, "failed to get workspace agent logs after", slog.Error(err))

continue

}

if len(logs) == 0 {

// Just keep listening - more logs might come in the future!

continue

}

select {

case <-ctx.Done():

return

case bufferedLogs <- logs:

lastSentLogID = logs[len(logs)-1].ID

}

}

}()

defer func() {

// Ensure that we don't return until the goroutine has exited.

//nolint:revive // Consume channel to wait until it's closed.

for range bufferedLogs {

}

}()

for {

select {

case <-ctx.Done():

logger.Debug(ctx, "job logs context canceled")

return

case logs, ok := <-bufferedLogs:

if !ok {

select {

case <-ctx.Done():

logger.Debug(ctx, "job logs context canceled")

default:

logger.Debug(ctx, "reached the end of published logs")

}

return

}

err = encoder.Encode(convertWorkspaceAgentLogs(logs))

if err != nil {

return

}

}

}

}

// @Summary Get listening ports for workspace agent

// @ID get-listening-ports-for-workspace-agent

// @Security CoderSessionToken

// @Produce json

// @Tags Agents

// @Param workspaceagent path string true "Workspace agent ID" format(uuid)

// @Success 200 {object} codersdk.WorkspaceAgentListeningPortsResponse

// @Router /workspaceagents/{workspaceagent}/listening-ports [get]

func (api *API) workspaceAgentListeningPorts(rw http.ResponseWriter, r *http.Request) {

ctx := r.Context()

waws := httpmw.WorkspaceAgentAndWorkspaceParam(r)

// If the agent is unreachable, the request will hang. Assume that if we

// don't get a response after 30s that the agent is unreachable.

ctx, cancel := context.WithTimeout(ctx, 30*time.Second)

defer cancel()

apiAgent, err := db2sdk.WorkspaceAgent(

api.DERPMap(), *api.TailnetCoordinator.Load(), waws.WorkspaceAgent, nil, nil, nil, api.AgentInactiveDisconnectTimeout,

api.DeploymentValues.AgentFallbackTroubleshootingURL.String(),

)

if err != nil {

httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{

Message: "Internal error reading workspace agent.",

Detail: err.Error(),

})

return

}

if apiAgent.Status != codersdk.WorkspaceAgentConnected {

httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{

Message: fmt.Sprintf("Agent state is %q, it must be in the %q state.", apiAgent.Status, codersdk.WorkspaceAgentConnected),

})

return

}

agentConn, release, err := api.agentProvider.AgentConn(ctx, waws.WorkspaceAgent.ID)

if err != nil {

httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{

Message: "Internal error dialing workspace agent.",

Detail: err.Error(),

})

return

}

defer release()

portsResponse, err := agentConn.ListeningPorts(ctx)

if err != nil {

httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{

Message: "Internal error fetching listening ports.",

Detail: err.Error(),

})

return

}

// Get a list of ports that are in-use by applications.

apps, err := api.Database.GetWorkspaceAppsByAgentID(ctx, waws.WorkspaceAgent.ID)

if xerrors.Is(err, sql.ErrNoRows) {

apps = []database.WorkspaceApp{}

err = nil

}

if err != nil {

httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{

Message: "Internal error fetching workspace apps.",

Detail: err.Error(),

})

return

}

appPorts := make(map[uint16]struct{}, len(apps))

for _, app := range apps {

if !app.Url.Valid || app.Url.String == "" {

continue

}

u, err := url.Parse(app.Url.String)

if err != nil {

continue

}

port := u.Port()

if port == "" {

continue

}

portNum, err := strconv.ParseUint(port, 10, 16)

if err != nil {

continue

}

if portNum < 1 || portNum > 65535 {

continue

}

appPorts[uint16(portNum)] = struct{}{}

}

// Filter out ports that are globally blocked, in-use by applications, or

// common non-HTTP ports such as databases, FTP, SSH, etc.

filteredPorts := make([]codersdk.WorkspaceAgentListeningPort, 0, len(portsResponse.Ports))

for _, port := range portsResponse.Ports {

if port.Port < workspacesdk.AgentMinimumListeningPort {

continue

}

if _, ok := appPorts[port.Port]; ok {

continue

}

if _, ok := workspacesdk.AgentIgnoredListeningPorts[port.Port]; ok {

continue

}

filteredPorts = append(filteredPorts, port)

}

portsResponse.Ports = filteredPorts

httpapi.Write(ctx, rw, http.StatusOK, portsResponse)

}

// @Summary Watch workspace agent for container updates.

// @ID watch-workspace-agent-for-container-updates

// @Security CoderSessionToken

// @Produce json

// @Tags Agents

// @Param workspaceagent path string true "Workspace agent ID" format(uuid)

// @Success 200 {object} codersdk.WorkspaceAgentListContainersResponse

// @Router /workspaceagents/{workspaceagent}/containers/watch [get]

func (api *API) watchWorkspaceAgentContainers(rw http.ResponseWriter, r *http.Request) {

var (

ctx = r.Context()

waws = httpmw.WorkspaceAgentAndWorkspaceParam(r)

logger = api.Logger.Named("agent_container_watcher").With(slog.F("agent_id", waws.WorkspaceAgent.ID))

)

// If the agent is unreachable, the request will hang. Assume that if we

// don't get a response after 30s that the agent is unreachable.

dialCtx, dialCancel := context.WithTimeout(ctx, 30*time.Second)

defer dialCancel()

apiAgent, err := db2sdk.WorkspaceAgent(

api.DERPMap(),

*api.TailnetCoordinator.Load(),

waws.WorkspaceAgent,

nil,

nil,

nil,

api.AgentInactiveDisconnectTimeout,

api.DeploymentValues.AgentFallbackTroubleshootingURL.String(),

)

if err != nil {

httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{

Message: "Internal error reading workspace agent.",

Detail: err.Error(),

})

return

}

if apiAgent.Status != codersdk.WorkspaceAgentConnected {

httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{

Message: fmt.Sprintf("Agent state is %q, it must be in the %q state.", apiAgent.Status, codersdk.WorkspaceAgentConnected),

})

return

}

agentConn, release, err := api.agentProvider.AgentConn(dialCtx, waws.WorkspaceAgent.ID)

if err != nil {

httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{

Message: "Internal error dialing workspace agent.",

Detail: err.Error(),

})

return

}

defer release()

containersCh, closer, err := agentConn.WatchContainers(ctx, logger)

if err != nil {

httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{

Message: "Internal error watching agent's containers.",

Detail: err.Error(),

})

return

}

defer closer.Close()

conn, err := websocket.Accept(rw, r, nil)

if err != nil {

httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{

Message: "Failed to upgrade connection to websocket.",

Detail: err.Error(),

})

return

}

ctx, cancel := context.WithCancel(r.Context())

defer cancel()

// Here we close the websocket for reading, so that the websocket library will handle pings and

// close frames.

_ = conn.CloseRead(context.Background())

ctx, wsNetConn := codersdk.WebsocketNetConn(ctx, conn, websocket.MessageText)

defer wsNetConn.Close()

go httpapi.HeartbeatClose(ctx, logger, cancel, conn)

encoder := json.NewEncoder(wsNetConn)

for {

select {

case <-api.ctx.Done():

return

case <-ctx.Done():

return

case containers, ok := <-containersCh:

if !ok {

return

}

if err := encoder.Encode(containers); err != nil {

api.Logger.Error(ctx, "encode containers", slog.Error(err))

return

}

}

}

}

// @Summary Get running containers for workspace agent

// @ID get-running-containers-for-workspace-agent

// @Security CoderSessionToken

// @Produce json

// @Tags Agents

// @Param workspaceagent path string true "Workspace agent ID" format(uuid)

// @Param label query string true "Labels" format(key=value)

// @Success 200 {object} codersdk.WorkspaceAgentListContainersResponse

// @Router /workspaceagents/{workspaceagent}/containers [get]

func (api *API) workspaceAgentListContainers(rw http.ResponseWriter, r *http.Request) {

ctx := r.Context()

waws := httpmw.WorkspaceAgentAndWorkspaceParam(r)

labelParam, ok := r.URL.Query()["label"]

if !ok {

labelParam = []string{}

}

labels := make(map[string]string, len(labelParam)/2)

for _, label := range labelParam {

kvs := strings.Split(label, "=")

if len(kvs) != 2 {

httpapi.Write(r.Context(), rw, http.StatusBadRequest, codersdk.Response{

Message: "Invalid label format",

Detail: "Labels must be in the format key=value",

})

return

}

labels[kvs[0]] = kvs[1]

}

// If the agent is unreachable, the request will hang. Assume that if we

// don't get a response after 30s that the agent is unreachable.

ctx, cancel := context.WithTimeout(ctx, 30*time.Second)

defer cancel()

apiAgent, err := db2sdk.WorkspaceAgent(

api.DERPMap(),

*api.TailnetCoordinator.Load(),

waws.WorkspaceAgent,

nil,

nil,

nil,

api.AgentInactiveDisconnectTimeout,

api.DeploymentValues.AgentFallbackTroubleshootingURL.String(),

)

if err != nil {

httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{

Message: "Internal error reading workspace agent.",

Detail: err.Error(),

})

return

}

if apiAgent.Status != codersdk.WorkspaceAgentConnected {

httpapi.Write(ctx, rw, http.StatusBadRequest, codersdk.Response{

Message: fmt.Sprintf("Agent state is %q, it must be in the %q state.", apiAgent.Status, codersdk.WorkspaceAgentConnected),

})

return

}

agentConn, release, err := api.agentProvider.AgentConn(ctx, waws.WorkspaceAgent.ID)

if err != nil {

httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{

Message: "Internal error dialing workspace agent.",

Detail: err.Error(),

})

return

}

defer release()

// Get a list of containers that the agent is able to detect

cts, err := agentConn.ListContainers(ctx)

if err != nil {

if errors.Is(err, context.Canceled) {

httpapi.Write(ctx, rw, http.StatusRequestTimeout, codersdk.Response{

Message: "Failed to fetch containers from agent.",

Detail: "Request timed out.",

})

return

}

// If the agent returns a codersdk.Error, we can return that directly.

if cerr, ok := codersdk.AsError(err); ok {

httpapi.Write(ctx, rw, cerr.StatusCode(), cerr.Response)

return

}

httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{

Message: "Internal error fetching containers.",

Detail: err.Error(),

})

return

}

// Filter in-place by labels

cts.Containers = slices.DeleteFunc(cts.Containers, func(ct codersdk.WorkspaceAgentContainer) bool {

return !maputil.Subset(labels, ct.Labels)

})

httpapi.Write(ctx, rw, http.StatusOK, cts)

}

// @Summary Delete devcontainer for workspace agent

// @ID delete-devcontainer-for-workspace-agent

// @Security CoderSessionToken

// @Tags Agents