alice.pds.demo.boltless.dev
This repository has no description
1//go:build linux
2
3package microvm
4
5import (
6 "context"
7 "crypto/rand"
8 "encoding/binary"
9 "errors"
10 "fmt"
11 "io"
12 "log/slog"
13 "maps"
14 "math"
15 "net"
16 "os"
17 "os/exec"
18 "path/filepath"
19 "regexp"
20 "slices"
21 "strings"
22 "sync/atomic"
23 "time"
24
25 "tangled.org/core/spindle/models"
26)
27
28const (
29 minGuestCID = 3
30 vmCrashLogTailBytes = 4096
31)
32
33func AllocateCID() (uint32, error) {
34 var data [4]byte
35 if _, err := rand.Read(data[:]); err != nil {
36 return 0, fmt.Errorf("allocate guest CID: %w", err)
37 }
38 return minGuestCID + binary.BigEndian.Uint32(data[:])%60000, nil
39}
40
41func prepareWorkDir(workDir string) error {
42 if workDir == "" {
43 return fmt.Errorf("microvm work directory is required")
44 }
45 if err := os.MkdirAll(workDir, 0o755); err != nil {
46 return fmt.Errorf("create microvm work directory: %w", err)
47 }
48 return nil
49}
50
51func prepareVolumes(ctx context.Context, workDir string, volumes []Volume, mkfsExt4 string) (map[string]string, error) {
52 paths := make(map[string]string, len(volumes))
53 for _, volume := range volumes {
54 if volume.ReadOnly {
55 return nil, fmt.Errorf("read-only microvm volume %q is not supported yet", volume.Image)
56 }
57 if volume.FSType != "ext4" {
58 return nil, fmt.Errorf("microvm volume %q uses unsupported fsType %q", volume.Image, volume.FSType)
59 }
60 if volume.ImageType != "" && volume.ImageType != "raw" {
61 return nil, fmt.Errorf("microvm volume %q uses unsupported imageType %q", volume.Image, volume.ImageType)
62 }
63
64 path := filepath.Join(workDir, filepath.Base(volume.Image))
65 if err := createSparseFile(path, volume.SizeMiB); err != nil {
66 return nil, err
67 }
68 noJournal := volume.MountPoint == "/workspace"
69 if err := runMkfsExt4(ctx, mkfsExt4, path, noJournal); err != nil {
70 return nil, err
71 }
72 paths[volume.Image] = path
73 }
74 return paths, nil
75}
76
77func createSparseFile(path string, sizeMiB int64) error {
78 if sizeMiB <= 0 {
79 return fmt.Errorf("sparse file %q size must be positive", path)
80 }
81 if sizeMiB > math.MaxInt64/(1024*1024) {
82 return fmt.Errorf("sparse file %q size is too large", path)
83 }
84 file, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE|os.O_EXCL, 0o600)
85 if err != nil {
86 return fmt.Errorf("create sparse file %q: %w", path, err)
87 }
88 defer file.Close()
89
90 if err := file.Truncate(sizeMiB * 1024 * 1024); err != nil {
91 return fmt.Errorf("resize sparse file %q: %w", path, err)
92 }
93 return nil
94}
95
96func runMkfsExt4(ctx context.Context, mkfsExt4, path string, noJournal bool) error {
97 if mkfsExt4 == "" {
98 return fmt.Errorf("mkfs.ext4 path is required")
99 }
100 args := []string{"-F"}
101 if noJournal {
102 args = append(args, "-O", "^has_journal")
103 }
104 args = append(args, path)
105
106 cmd := exec.CommandContext(ctx, mkfsExt4, args...)
107 output, err := cmd.CombinedOutput()
108 if err != nil {
109 return fmt.Errorf("mkfs.ext4 %q: %w: %s", path, err, strings.TrimSpace(string(output)))
110 }
111 return nil
112}
113
114func createParentedFile(path string) (*os.File, error) {
115 if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
116 return nil, fmt.Errorf("create log directory: %w", err)
117 }
118 file, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0o644)
119 if err != nil {
120 return nil, fmt.Errorf("create log file %q: %w", path, err)
121 }
122 return file, nil
123}
124
125type VMLogs struct {
126 Serial string
127 Extra map[string]string
128}
129
130type VMHandle interface {
131 Shutdown(ctx context.Context) error
132 WaitContext(ctx context.Context) error
133 Close() error
134 Logs() VMLogs
135 CID() uint32
136 WorkDir() string
137 OOMKilled() bool
138}
139
140type VMConfig struct {
141 Image ImageSpec
142 CID uint32
143 EnableKVM bool
144 WorkDir string
145 Cgroup CgroupLimits
146
147 BootTimeout time.Duration
148 MkfsExt4 string
149 Dev bool
150}
151
152type workflowState struct {
153 ImageSpec ImageSpec
154 ImageSpecPath string
155 Config manifestConfig
156 ConfigKey string
157 Image string
158 CacheReadURLs []string
159 CacheTrustedPublicKeys []string
160 VM VMHandle
161 Agent *AgentSession
162 ReadCache *ReadCacheProxy
163 UploadCache *UploadCacheProxy
164 DNSProxy *DNSProxy
165 WorkDir string
166 NixOSToplevelCache nixosToplevelCacheStore
167}
168
169func (e *Engine) cleanupState(ctx context.Context, wid models.WorkflowId, state *workflowState) error {
170 if state == nil {
171 return nil
172 }
173
174 ctx = context.WithoutCancel(ctx)
175
176 var err error
177 // todo(dawn): expose this error to the user as a warning
178 if drainErr := e.drainNixCache(ctx, state); drainErr != nil {
179 e.l.Warn("cache drain failed during cleanup; continuing", "workflow", wid, "error", drainErr)
180 }
181 err = errors.Join(err, e.shutdownVM(ctx, wid, state))
182 err = errors.Join(err, closeIO(&state.Agent))
183 err = errors.Join(err, closeIO(&state.ReadCache))
184 err = errors.Join(err, closeIO(&state.UploadCache))
185 err = errors.Join(err, closeIO(&state.DNSProxy))
186 err = errors.Join(err, removeWorkDir(state))
187 return err
188}
189
190func (e *Engine) drainNixCache(ctx context.Context, state *workflowState) error {
191 if e.cfg.NixCache.UploadURL == "" {
192 return nil
193 }
194
195 drainCtx, cancel := context.WithTimeout(ctx, cacheDrainTimeout)
196 defer cancel()
197
198 if state.Agent != nil {
199 if _, err := state.Agent.Drain(drainCtx); err != nil {
200 return fmt.Errorf("drain guest nix cache uploads: %w", err)
201 }
202 }
203 return nil
204}
205
206func (e *Engine) shutdownVM(ctx context.Context, wid models.WorkflowId, state *workflowState) error {
207 if state.VM == nil {
208 return nil
209 }
210 if vmExited(state.VM) {
211 return closeIO(&state.VM)
212 }
213
214 var poweroffErr error
215
216 if state.Agent != nil {
217 gracefulCtx, cancel := context.WithTimeout(ctx, vmShutdownTimeout)
218 var poweredOff bool
219 poweredOff, poweroffErr = e.poweroffViaAgent(gracefulCtx, wid, state)
220 cancel()
221
222 if poweredOff {
223 return closeIO(&state.VM)
224 }
225 if vmExited(state.VM) {
226 return closeIO(&state.VM)
227 }
228 }
229
230 fallbackCtx, cancel := context.WithTimeout(ctx, vmShutdownTimeout)
231 defer cancel()
232
233 shutdownErr := state.VM.Shutdown(fallbackCtx)
234 if shutdownErr != nil && !vmExited(state.VM) {
235 e.l.Warn("microVM shutdown fallback failed", "workflow", wid, "error", shutdownErr)
236 return errors.Join(poweroffErr, shutdownErr, closeIO(&state.VM))
237 }
238
239 return closeIO(&state.VM)
240}
241
242func vmExited(vm VMHandle) bool {
243 ctx, cancel := context.WithCancel(context.Background())
244 cancel()
245 // a cancelled wait means the process is still live
246 // any other result means it exited
247 return !errors.Is(vm.WaitContext(ctx), context.Canceled)
248}
249
250func (e *Engine) poweroffViaAgent(ctx context.Context, wid models.WorkflowId, state *workflowState) (bool, error) {
251 if err := state.Agent.Poweroff(ctx); err != nil {
252 e.l.Warn("agent poweroff request failed", "workflow", wid, "error", err)
253 return false, err
254 }
255
256 if err := state.VM.WaitContext(ctx); err != nil {
257 e.l.Warn("agent poweroff did not stop microVM", "workflow", wid, "error", err)
258 return false, nil
259 }
260
261 return true, nil
262}
263
264// helper for closing io interfaces, sets to nil to prevent double-close
265func closeIO[T io.Closer](field *T) error {
266 closer := *field
267 var zero T
268 *field = zero
269 if any(closer) == any(zero) {
270 return nil
271 }
272 return closer.Close()
273}
274
275func removeWorkDir(state *workflowState) error {
276 if state.WorkDir == "" {
277 return nil
278 }
279
280 err := os.RemoveAll(state.WorkDir)
281 state.WorkDir = ""
282 return err
283}
284
285// returns a context derived from ctx that is cancelled either when ctx itself
286// is cancelled or when the microVM exits on its own. the returned flag reports
287// whether the VM exited (as opposed to ctx being cancelled for another reason,
288// e.g. the workflow timeout), letting callers tell a crash apart from a
289// timeout. cancel must be called to release the watcher goroutine.
290func watchVMExit(ctx context.Context, vm VMHandle) (context.Context, *atomic.Bool, context.CancelFunc) {
291 exited := &atomic.Bool{}
292 watchCtx, cancel := context.WithCancel(ctx)
293 if vm == nil {
294 return watchCtx, exited, cancel
295 }
296 go func() {
297 _ = vm.WaitContext(watchCtx) // returns when VM exits or watchCtx is cancelled
298 if watchCtx.Err() == nil {
299 exited.Store(true)
300 cancel() // don't forget to cancel the watchCtx...
301 }
302 }()
303 return watchCtx, exited, cancel
304}
305
306func VMCrashLog(vm VMHandle) string {
307 if vm == nil {
308 return ""
309 }
310 logs := vm.Logs()
311
312 var b strings.Builder
313 if tail := tailFile(logs.Serial, vmCrashLogTailBytes); tail != "" {
314 fmt.Fprintf(&b, "==== serial log ====\n%s\n", tail)
315 }
316 for _, name := range slices.Sorted(maps.Keys(logs.Extra)) {
317 if tail := tailFile(logs.Extra[name], vmCrashLogTailBytes); tail != "" {
318 fmt.Fprintf(&b, "==== %s log ====\n%s\n", name, tail)
319 }
320 }
321 return strings.TrimRight(b.String(), "\n")
322}
323
324func tailFile(path string, max int64) string {
325 if path == "" {
326 return ""
327 }
328 f, err := os.Open(path)
329 if err != nil {
330 return ""
331 }
332 defer f.Close()
333 if info, err := f.Stat(); err == nil && info.Size() > max {
334 if _, err := f.Seek(-max, io.SeekEnd); err != nil {
335 return ""
336 }
337 }
338 data, err := io.ReadAll(f)
339 if err != nil {
340 return ""
341 }
342 return strings.TrimSpace(string(data))
343}
344
345func waitAgentConn(ctx context.Context, connCh <-chan net.Conn) (net.Conn, error) {
346 select {
347 case conn := <-connCh:
348 if conn == nil {
349 return nil, fmt.Errorf("agent connection closed before setup")
350 }
351 return conn, nil
352 case <-ctx.Done():
353 return nil, fmt.Errorf("waiting for agent: %w", ctx.Err())
354 }
355}
356
357func StartVM(ctx context.Context, cfg VMConfig, logger *slog.Logger) (VMHandle, error) {
358 if logger == nil {
359 logger = slog.Default()
360 }
361
362 runner, err := runnerFor(cfg.Image.RunnerType)
363 if err != nil {
364 return nil, err
365 }
366 if err := cfg.Image.Validate(); err != nil {
367 return nil, err
368 }
369 if err := cfg.Image.validateImageFiles(); err != nil {
370 return nil, err
371 }
372 if err := runner.Validate(cfg.Image, cfg.EnableKVM); err != nil {
373 return nil, err
374 }
375
376 if err := prepareWorkDir(cfg.WorkDir); err != nil {
377 return nil, err
378 }
379
380 mkfsExt4 := cfg.MkfsExt4
381 if mkfsExt4 == "" {
382 mkfsExt4, err = exec.LookPath("mkfs.ext4")
383 if err != nil {
384 return nil, fmt.Errorf("mkfs.ext4 command not found in PATH: %w", err)
385 }
386 }
387 volumePaths, err := prepareVolumes(ctx, cfg.WorkDir, cfg.Image.Volumes, mkfsExt4)
388 if err != nil {
389 return nil, err
390 }
391
392 return runner.Start(ctx, cfg, volumePaths, logger)
393}
394
395// checks serial log for ooms or kernel panic
396// this is very linux specific! but these strings are stable in linux itself, see mm/oom_kill.c and kernel/panic.c
397func ParseCrashLog(detail string) (error, bool) {
398 if strings.Contains(detail, "Out of memory:") {
399 // we can show process name where possible
400 re := regexp.MustCompile(`Out of memory: Killed process \d+ \(([^)]+)\)`)
401 matches := re.FindStringSubmatch(detail)
402 if len(matches) > 1 {
403 return fmt.Errorf("guest out of memory (process '%s' killed by guest kernel OOM)", matches[1]), true
404 }
405 return errors.New("guest out of memory (OOM killer invoked)"), true
406 }
407 if strings.Contains(detail, "Kernel panic") {
408 return errors.New("guest kernel panic"), true
409 }
410 return nil, false
411}