Something went wrong. Try again.
Monorepo for Tangled forked from tangled.org/core
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402//go:build linux
package microvm
import ( "context" "encoding/hex" "errors" "fmt" "hash/fnv" "io" "log/slog" "net" "regexp" "strings" "sync" "time"
"tangled.org/core/spindle/agentproto" agentv1 "tangled.org/core/spindle/agentproto/gen" "tangled.org/core/spindle/models")
const debugAcceptTimeout = 15 * time.Second
type debugTarget struct { cid uint32 agent *AgentSession failedStep *agentv1.ExecStart knot string repoDid string maxAliveAt time.Time connected chan struct{} // closed when the user first ssh's in, ending the grace window released chan struct{} // closed when the user exits the debug shell, to tear down early evicted chan struct{} // closed when a newer failure takes over the hold holdRepo string actor string // asks the mill on executors, nil means the engine's own authorizer decides keyCheck func(ctx context.Context, key []byte) bool}
func (t debugTarget) sharesHoldWith(other debugTarget) bool { return (t.holdRepo != "" && t.holdRepo == other.holdRepo) || (t.actor != "" && t.actor == other.actor)}
func isClosed(ch chan struct{}) bool { select { case <-ch: return true default: return false }}
var debugHandleSafe = regexp.MustCompile(`[^a-zA-Z0-9_.-]`)
func newDebugHandle(wid models.WorkflowId) string { h := fnv.New32a() _, _ = io.WriteString(h, wid.String()) token := hex.EncodeToString(h.Sum(nil)) name := strings.Trim(debugHandleSafe.ReplaceAllString(wid.Name, "-"), "-") if name == "" { return token } return name + "-" + token}
func (e *Engine) registerDebugTarget(wid models.WorkflowId, t debugTarget) { e.debugMu.Lock() defer e.debugMu.Unlock() e.debug[newDebugHandle(wid)] = t}
// a held VM keeps its whole seat, so each repo and each pusher get one at a// time. a newer failure replaces a hold nobody has connected to yet, and// false means someone is already debugging onefunc (e *Engine) claimDebugHold(wid models.WorkflowId, t debugTarget) bool { e.debugMu.Lock() defer e.debugMu.Unlock() var replaced []string for handle, held := range e.debug { if !held.sharesHoldWith(t) { continue } if isClosed(held.connected) { return false } replaced = append(replaced, handle) } for _, handle := range replaced { close(e.debug[handle].evicted) delete(e.debug, handle) } e.debug[newDebugHandle(wid)] = t return true}
func (e *Engine) unregisterDebugTarget(wid models.WorkflowId) { e.debugMu.Lock() defer e.debugMu.Unlock() delete(e.debug, newDebugHandle(wid))}
// ends the grace window so we hold the VM until the shell exits instead.// false means a newer failure replaced this hold while the shell was openingfunc (e *Engine) markDebugConnected(handle string) bool { e.debugMu.Lock() defer e.debugMu.Unlock() t, ok := e.debug[handle] if !ok { return false } if !isClosed(t.connected) { close(t.connected) } return true}
func (e *Engine) releaseDebugTarget(handle string) { e.debugMu.Lock() defer e.debugMu.Unlock() t, ok := e.debug[handle] if !ok { return } delete(e.debug, handle) close(t.released)}
func (e *Engine) lookupDebugTarget(handle string) (debugTarget, bool) { e.debugMu.Lock() defer e.debugMu.Unlock() t, ok := e.debug[handle] return t, ok}
func (e *Engine) RepoForJob(jobID string) (knot, repoDid string, ok bool) { t, found := e.lookupDebugTarget(jobID) if !found || t.knot == "" || t.repoDid == "" { return "", "", false } return t.knot, t.repoDid, true}
func (e *Engine) OpenDebugSession(ctx context.Context, jobID, term string, rows, cols int) (*DebugSession, error) { target, ok := e.lookupDebugTarget(jobID) if !ok { return nil, fmt.Errorf("no live microVM for job %q", jobID) }
ln, port, err := listenRandomVsockPort(ctx) if err != nil { return nil, fmt.Errorf("listen for debug shell: %w", err) } filtered := &cidFilteredVsockListener{Listener: ln, cid: target.cid, logger: e.l}
if err := target.agent.OpenDebugShell(&agentv1.OpenDebugShell{ VsockPort: port, Term: term, Rows: clampDim(rows), Cols: clampDim(cols), FailedStep: target.failedStep, ShellPrelude: depsSourcePrelude(), }); err != nil { _ = ln.Close() return nil, fmt.Errorf("ask guest to open debug shell: %w", err) }
conn, err := acceptWithTimeout(ctx, filtered, debugAcceptTimeout) if err != nil { _ = ln.Close() return nil, fmt.Errorf("accept debug shell connection: %w", err) }
if !e.markDebugConnected(jobID) { _ = conn.Close() _ = ln.Close() return nil, fmt.Errorf("a newer failed run replaced the microVM for job %q", jobID) }
return newDebugSession(conn, ln, e.l), nil}
func (e *Engine) writeDebugHint(wid models.WorkflowId, idx int, wfLogger models.WorkflowLogger, held bool) { if wfLogger == nil { wfLogger = models.NullLogger{} } step := Step{name: "Debug shell", kind: models.StepKindSystem}
wfLogger.ControlWriter(idx, step, models.StepStatusStart).Write([]byte{0}) defer wfLogger.ControlWriter(idx, step, models.StepStatusEnd).Write([]byte{0})
out := wfLogger.DataWriter(idx, "stdout") if !held { fmt.Fprintln(out, "Workflow failed, but someone is already debugging another failed run from this repo or pusher, so this VM won't be kept.") return } ssh := e.cfg.MicroVMPipelines.DebugSSH cmd := debugSSHCommand(ssh.ListenAddr, e.cfg.Server.Hostname, ssh.Host, ssh.JumpHost, newDebugHandle(wid)) fmt.Fprintf(out, "Workflow failed, connect within %s to debug until shell exit or workflow timeout:\n", ssh.GracePeriod) fmt.Fprintf(out, " %s\n", cmd)}
func (e *Engine) maybeRetainForDebug(ctx context.Context, wid models.WorkflowId) { handle := newDebugHandle(wid) target, ok := e.lookupDebugTarget(handle) if !ok { return }
grace := e.cfg.MicroVMPipelines.DebugSSH.GracePeriod e.l.Info("retaining failed microVM for debug", "workflow", wid, "grace", grace.String())
maxAlive := time.NewTimer(time.Until(target.maxAliveAt)) defer maxAlive.Stop() graceTimer := time.NewTimer(grace) defer graceTimer.Stop()
// wait for the user to ssh in within the grace window select { case <-ctx.Done(): return case <-maxAlive.C: e.l.Info("debug retention hit max VM lifetime; tearing down microVM", "workflow", wid) return case <-graceTimer.C: e.l.Info("nobody ssh'd in within grace; tearing down microVM", "workflow", wid) return case <-target.evicted: e.l.Info("a newer failed run took the debug hold; tearing down microVM", "workflow", wid) return case <-target.connected: e.l.Info("debug shell connected; holding microVM until exit", "workflow", wid) }
// connected: hold the VM until the user exits or it hits its max lifetime select { case <-ctx.Done(): case <-maxAlive.C: e.l.Info("debug session hit max VM lifetime; tearing down microVM", "workflow", wid) case <-target.released: e.l.Info("debug shell exited; tearing down microVM", "workflow", wid) }}
func debugSSHCommand(listenAddr, hostname, debugHost, jumpHost, jobID string) string { host, port := hostname, "" if debugHost != "" { host = debugHost } if _, p, err := net.SplitHostPort(listenAddr); err == nil { port = p } args := []string{"ssh", "-tt"} if jumpHost != "" { args = append(args, "-J", jumpHost) } if port != "" && port != "22" { args = append(args, "-p", port) } args = append(args, jobID+"@"+host) return strings.Join(args, " ")}
// bridges an interactive shell over the agentproto vsock.// Read to it gets the shell output from guest, Write sends the keyboard input from user.type DebugSession struct { conn net.Conn ln net.Listener enc *agentproto.Encoder dec *agentproto.Decoder l *slog.Logger
out chan []byte leftover []byte closeOne sync.Once
stateMu sync.Mutex state debugSessionResult}
type debugSessionResult struct { exit *agentv1.ExecExit readError string}
func newDebugSession(conn net.Conn, ln net.Listener, l *slog.Logger) *DebugSession { d := &DebugSession{ conn: conn, ln: ln, enc: agentproto.NewEncoder(conn), dec: agentproto.NewDecoder(conn), l: l, out: make(chan []byte, 16), } go d.readLoop() return d}
func (d *DebugSession) readLoop() { defer close(d.out) for { msg, err := d.dec.Decode() if err != nil { if !errors.Is(err, io.EOF) { d.l.Debug("debug shell decode ended", "error", err) d.stateMu.Lock() d.state.readError = err.Error() d.stateMu.Unlock() } return } if p := msg.PtyData; p != nil && len(p.Data) > 0 { d.out <- p.Data } else if p := msg.ExecExit; p != nil { d.stateMu.Lock() d.state.exit = p d.stateMu.Unlock() return } }}
func (d *DebugSession) Read(p []byte) (int, error) { if len(d.leftover) == 0 { chunk, ok := <-d.out if !ok { return 0, io.EOF } d.leftover = chunk } n := copy(p, d.leftover) d.leftover = d.leftover[n:] return n, nil}
func (d *DebugSession) Write(p []byte) (int, error) { if err := d.enc.Encode(&agentproto.Message{ Id: "pty", PtyData: &agentv1.PtyData{Data: append([]byte(nil), p...)}, }); err != nil { return 0, err } return len(p), nil}
func (d *DebugSession) Resize(rows, cols int) error { return d.enc.Encode(&agentproto.Message{ Id: "pty", PtyResize: &agentv1.PtyResize{Rows: clampDim(rows), Cols: clampDim(cols)}, })}
func (d *DebugSession) result() debugSessionResult { d.stateMu.Lock() defer d.stateMu.Unlock() return d.state}
func (d *DebugSession) Close() error { var err error d.closeOne.Do(func() { err = d.conn.Close() if d.ln != nil { _ = d.ln.Close() } }) return err}
func acceptWithTimeout(ctx context.Context, ln net.Listener, timeout time.Duration) (net.Conn, error) { ctx, cancel := context.WithTimeout(ctx, timeout) defer cancel()
type result struct { conn net.Conn err error } ch := make(chan result, 1) go func() { conn, err := ln.Accept() ch <- result{conn, err} }()
select { case <-ctx.Done(): return nil, ctx.Err() case r := <-ch: return r.conn, r.err }}
func clampDim(v int) uint32 { if v < 1 { return 1 } if v > 65535 { return 65535 } return uint32(v)}