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