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