alice.pds.demo.boltless.dev
This repository has no description
1package microvm
2
3import (
4 "context"
5 "encoding/hex"
6 "errors"
7 "fmt"
8 "hash/fnv"
9 "io"
10 "log/slog"
11 "net"
12 "regexp"
13 "strings"
14 "sync"
15 "time"
16
17 "tangled.org/core/spindle/agentproto"
18 agentv1 "tangled.org/core/spindle/agentproto/gen"
19 "tangled.org/core/spindle/models"
20)
21
22const debugAcceptTimeout = 15 * time.Second
23
24type debugTarget struct {
25 cid uint32
26 agent *AgentSession
27 knot string
28 repoDid string
29 wfLogger models.WorkflowLogger
30 maxAliveAt time.Time
31 stepCount int // index to emit the debug step at
32 connected chan struct{} // closed when the user first ssh's in, ending the grace window
33 released chan struct{} // closed when the user exits the debug shell, to tear down early
34}
35
36var debugHandleSafe = regexp.MustCompile(`[^a-zA-Z0-9_.-]`)
37
38func newDebugHandle(wid models.WorkflowId) string {
39 h := fnv.New32a()
40 _, _ = io.WriteString(h, wid.String())
41 token := hex.EncodeToString(h.Sum(nil))
42 name := strings.Trim(debugHandleSafe.ReplaceAllString(wid.Name, "-"), "-")
43 if name == "" {
44 return token
45 }
46 return name + "-" + token
47}
48
49func (e *Engine) registerDebugTarget(wid models.WorkflowId, t debugTarget) {
50 e.debugMu.Lock()
51 defer e.debugMu.Unlock()
52 e.debug[newDebugHandle(wid)] = t
53}
54
55func (e *Engine) unregisterDebugTarget(wid models.WorkflowId) {
56 e.debugMu.Lock()
57 defer e.debugMu.Unlock()
58 delete(e.debug, newDebugHandle(wid))
59}
60
61// ends the grace window so we hold the VM until the shell exits instead.
62func (e *Engine) markDebugConnected(handle string) {
63 e.debugMu.Lock()
64 defer e.debugMu.Unlock()
65 if t, ok := e.debug[handle]; ok && t.connected != nil {
66 select {
67 case <-t.connected: // already signalled
68 default:
69 close(t.connected)
70 }
71 }
72}
73
74func (e *Engine) releaseDebugTarget(handle string) {
75 e.debugMu.Lock()
76 defer e.debugMu.Unlock()
77 t, ok := e.debug[handle]
78 if !ok {
79 return
80 }
81 delete(e.debug, handle)
82 if t.released != nil {
83 close(t.released)
84 }
85}
86
87func (e *Engine) lookupDebugTarget(handle string) (debugTarget, bool) {
88 e.debugMu.Lock()
89 defer e.debugMu.Unlock()
90 t, ok := e.debug[handle]
91 return t, ok
92}
93
94func (e *Engine) RepoForJob(jobID string) (knot, repoDid string, ok bool) {
95 t, found := e.lookupDebugTarget(jobID)
96 if !found || t.knot == "" || t.repoDid == "" {
97 return "", "", false
98 }
99 return t.knot, t.repoDid, true
100}
101
102func (e *Engine) OpenDebugSession(ctx context.Context, jobID, term string, rows, cols int) (*DebugSession, error) {
103 target, ok := e.lookupDebugTarget(jobID)
104 if !ok {
105 return nil, fmt.Errorf("no live microVM for job %q", jobID)
106 }
107
108 ln, port, err := listenRandomVsockPort(ctx)
109 if err != nil {
110 return nil, fmt.Errorf("listen for debug shell: %w", err)
111 }
112 filtered := &cidFilteredVsockListener{Listener: ln, cid: target.cid, logger: e.l}
113
114 if err := target.agent.OpenDebugShell(&agentv1.OpenDebugShell{
115 VsockPort: port,
116 Term: term,
117 Rows: clampDim(rows),
118 Cols: clampDim(cols),
119 }); err != nil {
120 _ = ln.Close()
121 return nil, fmt.Errorf("ask guest to open debug shell: %w", err)
122 }
123
124 conn, err := acceptWithTimeout(ctx, filtered, debugAcceptTimeout)
125 if err != nil {
126 _ = ln.Close()
127 return nil, fmt.Errorf("accept debug shell connection: %w", err)
128 }
129
130 e.markDebugConnected(jobID)
131
132 return newDebugSession(conn, ln, e.l), nil
133}
134
135func (e *Engine) maybeRetainForDebug(ctx context.Context, wid models.WorkflowId) {
136 handle := newDebugHandle(wid)
137 target, ok := e.lookupDebugTarget(handle)
138 if !ok {
139 // no target registered means the workflow didn't fail; nothing to retain
140 return
141 }
142
143 wfLogger := target.wfLogger
144 if wfLogger == nil {
145 wfLogger = models.NullLogger{}
146 }
147 step := Step{name: "Debug shell", kind: models.StepKindSystem}
148 idx := target.stepCount
149
150 wfLogger.ControlWriter(idx, step, models.StepStatusStart).Write([]byte{0})
151 defer wfLogger.ControlWriter(idx, step, models.StepStatusEnd).Write([]byte{0})
152
153 ssh := e.cfg.MicroVMPipelines.DebugSSH
154 grace := ssh.GracePeriod
155
156 cmd := debugSSHCommand(ssh.ListenAddr, e.cfg.Server.Hostname, handle)
157 out := wfLogger.DataWriter(idx, "stdout")
158 fmt.Fprintf(out, "Workflow failed, connect within %s to debug until shell exit or workflow timeout:\n", grace)
159 fmt.Fprintf(out, " %s\n", cmd)
160 e.l.Info("retaining failed microVM for debug", "workflow", wid, "grace", grace.String())
161
162 maxAlive := time.NewTimer(time.Until(target.maxAliveAt))
163 defer maxAlive.Stop()
164 graceTimer := time.NewTimer(grace)
165 defer graceTimer.Stop()
166
167 // wait for the user to ssh in within the grace window
168 select {
169 case <-ctx.Done():
170 return
171 case <-maxAlive.C:
172 e.l.Info("debug retention hit max VM lifetime; tearing down microVM", "workflow", wid)
173 return
174 case <-graceTimer.C:
175 e.l.Info("nobody ssh'd in within grace; tearing down microVM", "workflow", wid)
176 return
177 case <-target.connected:
178 e.l.Info("debug shell connected; holding microVM until exit", "workflow", wid)
179 }
180
181 // connected: hold the VM until the user exits or it hits its max lifetime
182 select {
183 case <-ctx.Done():
184 case <-maxAlive.C:
185 e.l.Info("debug session hit max VM lifetime; tearing down microVM", "workflow", wid)
186 case <-target.released:
187 e.l.Info("debug shell exited; tearing down microVM", "workflow", wid)
188 }
189}
190
191func debugSSHCommand(listenAddr, hostname, jobID string) string {
192 host, port := hostname, ""
193 if h, p, err := net.SplitHostPort(listenAddr); err == nil {
194 port = p
195 if h != "" && h != "0.0.0.0" && h != "::" {
196 host = h
197 }
198 }
199 cmd := "ssh -tt "
200 if port != "" && port != "22" {
201 cmd += "-p " + port + " "
202 }
203 return cmd + jobID + "@" + host
204}
205
206// bridges an interactive shell over the agentproto vsock.
207// Read to it gets the shell output from guest, Write sends the keyboard input from user.
208type DebugSession struct {
209 conn net.Conn
210 ln net.Listener
211 enc *agentproto.Encoder
212 dec *agentproto.Decoder
213 l *slog.Logger
214
215 out chan []byte
216 leftover []byte
217 exitCode int
218 closeOne sync.Once
219}
220
221func newDebugSession(conn net.Conn, ln net.Listener, l *slog.Logger) *DebugSession {
222 d := &DebugSession{
223 conn: conn,
224 ln: ln,
225 enc: agentproto.NewEncoder(conn),
226 dec: agentproto.NewDecoder(conn),
227 l: l,
228 out: make(chan []byte, 16),
229 }
230 go d.readLoop()
231 return d
232}
233
234func (d *DebugSession) readLoop() {
235 defer close(d.out)
236 for {
237 msg, err := d.dec.Decode()
238 if err != nil {
239 if !errors.Is(err, io.EOF) {
240 d.l.Debug("debug shell decode ended", "error", err)
241 }
242 return
243 }
244 if p := msg.PtyData; p != nil && len(p.Data) > 0 {
245 d.out <- p.Data
246 } else if p := msg.ExecExit; p != nil {
247 d.exitCode = int(p.ExitCode)
248 return
249 }
250 }
251}
252
253func (d *DebugSession) Read(p []byte) (int, error) {
254 if len(d.leftover) == 0 {
255 chunk, ok := <-d.out
256 if !ok {
257 return 0, io.EOF
258 }
259 d.leftover = chunk
260 }
261 n := copy(p, d.leftover)
262 d.leftover = d.leftover[n:]
263 return n, nil
264}
265
266func (d *DebugSession) Write(p []byte) (int, error) {
267 if err := d.enc.Encode(&agentproto.Message{
268 Id: "pty",
269 PtyData: &agentv1.PtyData{Data: append([]byte(nil), p...)},
270 }); err != nil {
271 return 0, err
272 }
273 return len(p), nil
274}
275
276func (d *DebugSession) Resize(rows, cols int) error {
277 return d.enc.Encode(&agentproto.Message{
278 Id: "pty",
279 PtyResize: &agentv1.PtyResize{Rows: clampDim(rows), Cols: clampDim(cols)},
280 })
281}
282
283func (d *DebugSession) ExitCode() int { return d.exitCode }
284
285func (d *DebugSession) Close() error {
286 var err error
287 d.closeOne.Do(func() {
288 err = d.conn.Close()
289 if d.ln != nil {
290 _ = d.ln.Close()
291 }
292 })
293 return err
294}
295
296func acceptWithTimeout(ctx context.Context, ln net.Listener, timeout time.Duration) (net.Conn, error) {
297 ctx, cancel := context.WithTimeout(ctx, timeout)
298 defer cancel()
299
300 type result struct {
301 conn net.Conn
302 err error
303 }
304 ch := make(chan result, 1)
305 go func() {
306 conn, err := ln.Accept()
307 ch <- result{conn, err}
308 }()
309
310 select {
311 case <-ctx.Done():
312 return nil, ctx.Err()
313 case r := <-ch:
314 return r.conn, r.err
315 }
316}
317
318func clampDim(v int) uint32 {
319 if v < 1 {
320 return 1
321 }
322 if v > 65535 {
323 return 65535
324 }
325 return uint32(v)
326}