spindle/microvm: watch & report vm crashes / exits properly · oscillatory.net/core@d29480a

+13

spindle/engines/microvm/agent.go

··· 317 317 _ = s.conn.SetReadDeadline(time.Time{}) 318 318 } 319 319 320 + // a blocked vsock read wont wake up just from the ctx being cancelled, 321 + // only a deadline will wake it up, so if the VM crashes mid-step the read would 322 + // hang until workflow timeout. so we will set a deadline in the past to cancel it. 323 + // 324 + // we set a deadline here instead of closing the connection, this is the long-lived 325 + // connection that everything reuses, so we only really want to interrupt it for this 326 + // current read. this also lands as a timeout error which the netErr.Timeout() check 327 + // below maps to ctx.Err() correctly 328 + stop := context.AfterFunc(ctx, func() { 329 + _ = s.conn.SetReadDeadline(time.Now()) 330 + }) 331 + defer stop() 332 + 320 333 msg, err := s.dec.Decode() 321 334 if err != nil { 322 335 var netErr net.Error

+32 -7

spindle/engines/microvm/engine.go

··· 10 10 "os" 11 11 "slices" 12 12 "sync" 13 + "sync/atomic" 13 14 "time" 14 15 15 16 "gopkg.in/yaml.v3" ··· 281 282 return fmt.Errorf("microVM workflow is not connected to agent") 282 283 } 283 284 285 + stderr := wfLogger.DataWriter(idx, "stderr") 286 + 287 + execCtx, vmExited, cancelWatch := watchVMExit(ctx, state.VM) 288 + defer cancelWatch() 289 + 284 290 step := w.Steps[idx] 285 291 if s, ok := step.(Step); ok && s.action == activationStepAction { 286 - return e.activateConfig(ctx, wid, state, s, wfLogger.DataWriter(idx, "stdout")) 292 + err := e.activateConfig(execCtx, wid, state, s, wfLogger.DataWriter(idx, "stdout")) 293 + return e.classifyStepError(ctx, state, stderr, vmExited, err) 287 294 } 288 295 env := []string{ 289 296 "HOME=/workspace", ··· 304 311 } 305 312 306 313 stdout := wfLogger.DataWriter(idx, "stdout") 307 - stderr := wfLogger.DataWriter(idx, "stderr") 308 - exitCode, err := state.Agent.Exec(ctx, AgentExec{ 314 + exitCode, err := state.Agent.Exec(execCtx, AgentExec{ 309 315 ID: fmt.Sprintf("%s-%d", wid.String(), idx), 310 316 ExecStart: &agentv1.ExecStart{ 311 317 Argv: []string{guestShell, "-lc", step.Command()}, ··· 318 324 Stderr: stderr, 319 325 }) 320 326 if err != nil { 321 - if ctx.Err() != nil { 322 - return engine.ErrTimedOut 323 - } 324 - return err 327 + return e.classifyStepError(ctx, state, stderr, vmExited, err) 325 328 } 326 329 327 330 if exitCode != 0 { 328 331 return engine.ErrWorkflowFailed 329 332 } 330 333 return nil 334 + } 335 + 336 + // reads the vm serial logs so we report the tail of that as an error instead of 337 + // just "guest agent connection lost: EOF" 338 + func (e *Engine) classifyStepError(ctx context.Context, state *workflowState, stderr io.Writer, vmExited *atomic.Bool, err error) error { 339 + if err == nil { 340 + return nil 341 + } 342 + if vmExited != nil && vmExited.Load() { 343 + if detail := vmCrashLog(state.VM); detail != "" { 344 + fmt.Fprintf(stderr, "microVM exited unexpectedly:\n%s\n", detail) 345 + e.l.Error("microVM exited unexpectedly during step", "detail", detail) 346 + } else { 347 + fmt.Fprintln(stderr, "microVM exited unexpectedly") 348 + e.l.Error("microVM exited unexpectedly during step") 349 + } 350 + return errors.New("microVM exited unexpectedly; see workflow logs for serial/qemu output") 351 + } 352 + if ctx.Err() != nil { 353 + return engine.ErrTimedOut 354 + } 355 + return err 331 356 } 332 357 333 358 func (e *Engine) activateConfig(ctx context.Context, wid models.WorkflowId, state *workflowState, step Step, out io.Writer) error {

+66 -1

spindle/engines/microvm/vm.go

··· 6 6 "encoding/binary" 7 7 "errors" 8 8 "fmt" 9 + "io" 9 10 "log/slog" 10 11 "math" 11 12 "net" ··· 13 14 "os/exec" 14 15 "path/filepath" 15 16 "strings" 17 + "sync/atomic" 16 18 "time" 17 19 18 20 "tangled.org/core/spindle/models" 19 21 ) 20 22 21 - const minGuestCID = 3 23 + const ( 24 + minGuestCID = 3 25 + vmCrashLogTailBytes = 4096 26 + ) 22 27 23 28 func AllocateCID() (uint32, error) { 24 29 var data [4]byte ··· 306 311 err := os.RemoveAll(state.WorkDir) 307 312 state.WorkDir = "" 308 313 return err 314 + } 315 + 316 + // returns a context derived from ctx that is cancelled either when ctx itself 317 + // is cancelled or when the microVM exits on its own. the returned flag reports 318 + // whether the VM exited (as opposed to ctx being cancelled for another reason, 319 + // e.g. the workflow timeout), letting callers tell a crash apart from a 320 + // timeout. cancel must be called to release the watcher goroutine. 321 + func watchVMExit(ctx context.Context, vm VMHandle) (context.Context, *atomic.Bool, context.CancelFunc) { 322 + exited := &atomic.Bool{} 323 + watchCtx, cancel := context.WithCancel(ctx) 324 + if vm == nil { 325 + return watchCtx, exited, cancel 326 + } 327 + go func() { 328 + _ = vm.WaitContext(watchCtx) // returns when VM exits or watchCtx is cancelled 329 + if watchCtx.Err() == nil { 330 + exited.Store(true) 331 + cancel() // don't forget to cancel the watchCtx... 332 + } 333 + }() 334 + return watchCtx, exited, cancel 335 + } 336 + 337 + func vmCrashLog(vm VMHandle) string { 338 + if vm == nil { 339 + return "" 340 + } 341 + logs := vm.Logs() 342 + 343 + var b strings.Builder 344 + if tail := tailFile(logs.Serial, vmCrashLogTailBytes); tail != "" { 345 + fmt.Fprintf(&b, "==== serial log ====\n%s\n", tail) 346 + } 347 + if qemuLog := logs.Extra["qemu"]; qemuLog != "" { 348 + if tail := tailFile(qemuLog, vmCrashLogTailBytes); tail != "" { 349 + fmt.Fprintf(&b, "==== qemu log ====\n%s\n", tail) 350 + } 351 + } 352 + return strings.TrimRight(b.String(), "\n") 353 + } 354 + 355 + func tailFile(path string, max int64) string { 356 + if path == "" { 357 + return "" 358 + } 359 + f, err := os.Open(path) 360 + if err != nil { 361 + return "" 362 + } 363 + defer f.Close() 364 + if info, err := f.Stat(); err == nil && info.Size() > max { 365 + if _, err := f.Seek(-max, io.SeekEnd); err != nil { 366 + return "" 367 + } 368 + } 369 + data, err := io.ReadAll(f) 370 + if err != nil { 371 + return "" 372 + } 373 + return strings.TrimSpace(string(data)) 309 374 } 310 375 311 376 func waitAgentConn(ctx context.Context, connCh <-chan net.Conn) (net.Conn, error) {

Configure Feed

Configure Feed