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

fix: fix goroutine leak in log streaming over websocket (#15709)

Browse Source 
fixes #14881

Our handlers for streaming logs don't read from the websocket. We don't allow the client to send us any data, but the websocket library we use requires reading from the websocket to properly handle pings and closing. Not doing so can [can cause the websocket to hang on write](https://github.com/coder/websocket/issues/405), leaking go routines which were noticed in #14881.

This fixes the issue, and in process refactors our log streaming to a encoder/decoder package which provides generic types for sending JSON over websocket.

I'd also like for us to upgrade to the latest https://github.com/coder/websocket but we should also upgrade our tailscale fork before doing so to avoid including two copies of the websocket library.
This commit is contained in:
Spike Curtis
2024-12-03 10:12:30 +04:00
committed by GitHub
parent e4f6c9ac15
commit 148a5a3593
6 changed files with 134 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files
coderd/provisionerjobs.go
+4 -5
View File
@@ -15,6 +15,7 @@ import (
"nhooyr.io/websocket"
"cdr.dev/slog"
"github.com/coder/coder/v2/codersdk/wsjson"
"github.com/coder/coder/v2/coderd/database"
"github.com/coder/coder/v2/coderd/database/db2sdk"
@@ -312,6 +313,7 @@ type logFollower struct {
r *http.Request
rw http.ResponseWriter
conn *websocket.Conn
enc *wsjson.Encoder[codersdk.ProvisionerJobLog]
jobID uuid.UUID
after int64
@@ -391,6 +393,7 @@ func (f *logFollower) follow() {
}
defer f.conn.Close(websocket.StatusNormalClosure, "done")
go httpapi.Heartbeat(f.ctx, f.conn)
f.enc = wsjson.NewEncoder[codersdk.ProvisionerJobLog](f.conn, websocket.MessageText)
// query for logs once right away, so we can get historical data from before
// subscription
@@ -488,11 +491,7 @@ func (f *logFollower) query() error {
return xerrors.Errorf("error fetching logs: %w", err)
}
for _, log := range logs {
logB, err := json.Marshal(convertProvisionerJobLog(log))
if err != nil {
return xerrors.Errorf("error marshaling log: %w", err)
}
err = f.conn.Write(f.ctx, websocket.MessageText, logB)
err := f.enc.Encode(convertProvisionerJobLog(log))
if err != nil {
return xerrors.Errorf("error writing to websocket: %w", err)
}
coderd/workspaceagents.go
+7 -17
View File
@@ -39,6 +39,7 @@ import (
"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"
)
@@ -396,11 +397,9 @@ func (api *API) workspaceAgentLogs(rw http.ResponseWriter, r *http.Request) {
}
go httpapi.Heartbeat(ctx, conn)
ctx, wsNetConn := codersdk.WebsocketNetConn(ctx, conn, websocket.MessageText)
defer wsNetConn.Close() // Also closes conn.
encoder := wsjson.NewEncoder[[]codersdk.WorkspaceAgentLog](conn, websocket.MessageText)
defer encoder.Close(websocket.StatusNormalClosure)
// The Go stdlib JSON encoder appends a newline character after message write.
encoder := json.NewEncoder(wsNetConn)
err = encoder.Encode(convertWorkspaceAgentLogs(logs))
if err != nil {
return
@@ -740,16 +739,8 @@ func (api *API) derpMapUpdates(rw http.ResponseWriter, r *http.Request) {
})
return
}
ctx, nconn := codersdk.WebsocketNetConn(ctx, ws, websocket.MessageBinary)
defer nconn.Close()
// Slurp all packets from the connection into io.Discard so pongs get sent
// by the websocket package. We don't do any reads ourselves so this is
// necessary.
go func() {
_, _ = io.Copy(io.Discard, nconn)
_ = nconn.Close()
}()
encoder := wsjson.NewEncoder[*tailcfg.DERPMap](ws, websocket.MessageBinary)
defer encoder.Close(websocket.StatusGoingAway)
go func(ctx context.Context) {
// TODO(mafredri): Is this too frequent? Use separate ping disconnect timeout?
@@ -767,7 +758,7 @@ func (api *API) derpMapUpdates(rw http.ResponseWriter, r *http.Request) {
err := ws.Ping(ctx)
cancel()
if err != nil {
_ = nconn.Close()
_ = ws.Close(websocket.StatusGoingAway, "ping failed")
return
}
}
@@ -780,9 +771,8 @@ func (api *API) derpMapUpdates(rw http.ResponseWriter, r *http.Request) {
for {
derpMap := api.DERPMap()
if lastDERPMap == nil || !tailnet.CompareDERPMaps(lastDERPMap, derpMap) {
err := json.NewEncoder(nconn).Encode(derpMap)
err := encoder.Encode(derpMap)
if err != nil {
_ = nconn.Close()
return
}
lastDERPMap = derpMap