alice.pds.demo.boltless.dev
This repository has no description
1package microvm
2
3import (
4 "context"
5 "encoding/json"
6 "errors"
7 "fmt"
8 "io"
9 "log/slog"
10 "net/http"
11 "os"
12 "path/filepath"
13 "slices"
14 "sync"
15 "sync/atomic"
16 "time"
17
18 "gopkg.in/yaml.v3"
19
20 "tangled.org/core/api/tangled"
21 "tangled.org/core/log"
22 "tangled.org/core/spindle/agentproto"
23 agentv1 "tangled.org/core/spindle/agentproto/gen"
24 "tangled.org/core/spindle/config"
25 "tangled.org/core/spindle/db"
26 "tangled.org/core/spindle/engine"
27 "tangled.org/core/spindle/models"
28 "tangled.org/core/spindle/secrets"
29)
30
31const (
32 guestWorkDir = "/workspace/repo"
33 guestBasePATH = "/run/current-system/sw/bin:/nix/var/nix/profiles/default/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"
34 guestDevShellEnvPath = "/run/spindle/devshell-env.sh"
35 activationStepAction = "activate-config"
36 agentAcceptTimeout = 2 * time.Minute
37 agentHandshakeTimeout = 30 * time.Second
38 cacheDrainTimeout = 5 * time.Minute
39 vmShutdownTimeout = 10 * time.Second
40 guestTimeoutGrace = 5 * time.Second
41)
42
43type cleanupFunc func(context.Context) error
44
45type Engine struct {
46 l *slog.Logger
47 cfg *config.Config
48 db *db.DB
49 agentMu sync.Mutex
50 agent *agentHub
51 scheduler *engine.ResourceScheduler[Resources]
52 cgroupParent *CgroupParent
53
54 cleanupMu sync.Mutex
55 cleanup map[string][]cleanupFunc
56
57 debugMu sync.Mutex
58 debug map[string]debugTarget
59}
60
61type Step struct {
62 name string
63 kind models.StepKind
64 command string
65 environment map[string]string
66 action string
67 config manifestConfig
68 configKey string
69}
70
71func (s Step) Name() string { return s.name }
72func (s Step) Command() string { return s.command }
73func (s Step) Kind() models.StepKind { return s.kind }
74
75func New(ctx context.Context, cfg *config.Config, d *db.DB) (*Engine, error) {
76 l := log.FromContext(ctx).With("component", "engine.microvm")
77 budget, max, agingThreshold := newVMBudgetConfig(cfg.MicroVMPipelines)
78 l.Info("initialized microVM workflow budget", "budget", budget.String(), "maxWorkflow", max.String(), "agingThreshold", agingThreshold)
79
80 var cgroupParent *CgroupParent
81 var err error
82 if cfg.MicroVMPipelines.EnableCgroups {
83 cgroupParent, err = initCgroupParent(cfg.MicroVMPipelines.CgroupParent, cfg.MicroVMPipelines.CgroupSupervisorMemoryMinMiB, l)
84 if err != nil {
85 return nil, err
86 }
87 }
88
89 e := &Engine{
90 l: l,
91 cfg: cfg,
92 db: d,
93 scheduler: engine.NewResourceScheduler(budget, max, agingThreshold),
94 cgroupParent: cgroupParent,
95 cleanup: make(map[string][]cleanupFunc),
96 debug: make(map[string]debugTarget),
97 }
98
99 if cfg.MicroVMPipelines.DebugSSH.ListenAddr != "" {
100 go e.serveDebugSSH(ctx)
101 }
102
103 return e, nil
104}
105
106func (e *Engine) ensureAgentHub() (*agentHub, error) {
107 e.agentMu.Lock()
108 defer e.agentMu.Unlock()
109
110 if e.agent != nil {
111 return e.agent, nil
112 }
113
114 port := e.cfg.MicroVMPipelines.AgentPort
115 if port == 0 {
116 port = agentproto.DefaultPort
117 }
118 agent, err := newAgentHub(port, e.l)
119 if err != nil {
120 return nil, err
121 }
122 e.agent = agent
123 return agent, nil
124}
125
126func (e *Engine) InitWorkflow(twf tangled.Pipeline_Workflow, tpl tangled.Pipeline) (*models.Workflow, error) {
127 swf := &models.Workflow{}
128 var dwf manifestWorkflow
129
130 if err := engine.DescribeManifestError(twf.Raw, manifestWorkflow{}); err != nil {
131 return nil, err
132 }
133 if err := yaml.Unmarshal([]byte(twf.Raw), &dwf); err != nil {
134 return nil, err
135 }
136
137 for _, dstep := range dwf.Steps {
138 swf.Steps = append(swf.Steps, Step{
139 name: dstep.Name,
140 kind: models.StepKindUser,
141 command: dstep.Command,
142 environment: dstep.Environment,
143 })
144 }
145 swf.Name = twf.Name
146 swf.Environment = dwf.Environment
147
148 if tpl.TriggerMetadata != nil {
149 if clone := models.BuildCloneStep(twf, *tpl.TriggerMetadata, e.cfg.Server.Dev); clone.Command() != "" {
150 swf.Steps = append([]models.Step{clone}, swf.Steps...)
151 }
152 }
153
154 imageSpec, imageSpecPath, imageName, err := e.resolveImage(dwf.Image)
155 if err != nil {
156 return nil, err
157 }
158 configKey := ""
159 config := manifestConfig{
160 Services: dwf.Services,
161 Virtualisation: dwf.Virtualisation,
162 Dependencies: dwf.Dependencies,
163 Registry: dwf.Registry,
164 }
165 if config.Enabled() {
166 if !imageSpec.SupportsConfigActivation() {
167 return nil, fmt.Errorf(
168 "microVM image %q is not a NixOS image: services, virtualisation, dependencies and registry workflow options require a NixOS image",
169 imageName,
170 )
171 }
172 var err error
173 configKey, err = buildConfigKey(imageSpec, config)
174 if err != nil {
175 return nil, fmt.Errorf("build config key: %w", err)
176 }
177 activationStep := Step{
178 name: "NixOS config activation",
179 kind: models.StepKindSystem,
180 command: "activate nixos config",
181 action: activationStepAction,
182 config: config,
183 configKey: configKey,
184 }
185
186 insertAt := 0
187 if len(swf.Steps) > 0 && swf.Steps[0].Kind() == models.StepKindSystem {
188 insertAt = 1
189 }
190 swf.Steps = append(swf.Steps, nil)
191 copy(swf.Steps[insertAt+1:], swf.Steps[insertAt:])
192 swf.Steps[insertAt] = activationStep
193 }
194
195 cacheURLs, cacheKeys, err := workflowCaches(dwf.Caches)
196 if err != nil {
197 return nil, err
198 }
199
200 swf.Data = &workflowState{
201 ImageSpec: imageSpec,
202 ImageSpecPath: imageSpecPath,
203 Config: config,
204 ConfigKey: configKey,
205 Image: imageName,
206 CacheReadURLs: cacheURLs,
207 CacheTrustedPublicKeys: cacheKeys,
208 NixOSToplevelCache: newNixOSToplevelCacheStore(e.db),
209 }
210 return swf, nil
211}
212
213func (e *Engine) SetupWorkflow(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, wfLogger models.WorkflowLogger) (err error) {
214 l := e.l.With("workflow", wid)
215 setupStep := Step{name: "microVM setup", kind: models.StepKindSystem}
216
217 wfLogger.ControlWriter(-1, setupStep, models.StepStatusStart).Write([]byte{0})
218 defer wfLogger.ControlWriter(-1, setupStep, models.StepStatusEnd).Write([]byte{0})
219
220 category := "Failed to setup VM"
221 defer func() {
222 if err != nil {
223 err = fmt.Errorf("%s:\n%w", category, err)
224 }
225 }()
226
227 state, ok := wf.Data.(*workflowState)
228 if !ok || state == nil {
229 return fmt.Errorf("workflow state is not initialized")
230 }
231
232 cid, err := AllocateCID()
233 if err != nil {
234 return err
235 }
236 agent, err := e.ensureAgentHub()
237 if err != nil {
238 return err
239 }
240 connCh, unregister, err := agent.expect(cid)
241 if err != nil {
242 return err
243 }
244 defer unregister()
245
246 workDirBase := e.cfg.MicroVMPipelines.OverlayDir
247 if workDirBase == "" {
248 workDirBase = os.TempDir()
249 }
250 workDir, err := os.MkdirTemp(workDirBase, "spindle-microvm-"+wid.String()+"-*")
251 if err != nil {
252 return fmt.Errorf("create workflow microVM directory: %w", err)
253 }
254 state.WorkDir = workDir
255
256 setupDone := false
257 defer func() {
258 if setupDone {
259 return
260 }
261 if detail := vmCrashLog(state.VM); detail != "" {
262 l.Error("microVM setup failed", "detail", detail)
263 }
264 if err := e.cleanupState(context.Background(), wid, state); err != nil {
265 l.Error("failed to cleanup failed setup", "error", err)
266 }
267 }()
268
269 upstreams, err := BuildCacheUpstreams(e.cfg.NixCache.ReadURLs, state.CacheReadURLs)
270 if err != nil {
271 return err
272 }
273 readCache, err := StartReadCacheProxy(ctx, cid, upstreams, l)
274 if err != nil {
275 return err
276 }
277 state.ReadCache = readCache
278 stagingDir := filepath.Join(workDir, "upload-cache")
279 uploadCache, err := StartUploadCacheProxy(ctx, cid, e.cfg.NixCache.UploadURL, upstreams, stagingDir, l)
280 if err != nil {
281 return err
282 }
283 state.UploadCache = uploadCache
284 dnsProxy, err := StartDNSProxy(ctx, cid, l)
285 if err != nil {
286 return err
287 }
288 state.DNSProxy = dnsProxy
289
290 port := e.cfg.MicroVMPipelines.AgentPort
291 if port == 0 {
292 port = agentproto.DefaultPort
293 }
294 state.ImageSpec.BootArgs = fmt.Sprintf("%s shuttle.vsock_port=%d", state.ImageSpec.BootArgs, port)
295
296 fmt.Fprintf(wfLogger.DataWriter(-1, "stdout"), "starting microVM image %s\n", state.Image)
297 l.Info("starting microVM workflow", "image", state.Image, "imageSpec", state.ImageSpecPath, "cid", cid, "workDir", workDir)
298
299 var vm VMHandle
300 vm, err = StartVM(ctx, VMConfig{
301 Image: state.ImageSpec,
302 CID: cid,
303 EnableKVM: e.cfg.MicroVMPipelines.EnableKVM,
304 WorkDir: workDir,
305 Cgroup: e.cgroupLimits(wid, state.ImageSpec),
306 Dev: e.cfg.Server.Dev,
307 }, l)
308 if err != nil {
309 return err
310 }
311 state.VM = vm
312 state.StartedAt = time.Now()
313
314 category = "Failed to connect to agent"
315
316 acceptCtx, cancelAccept := context.WithTimeout(ctx, agentAcceptTimeout)
317 defer cancelAccept()
318 conn, err := waitAgentConn(acceptCtx, connCh)
319 if err != nil {
320 return err
321 }
322
323 agentSession := NewAgentSession(conn, l)
324 initCtx, cancelInit := context.WithTimeout(ctx, agentHandshakeTimeout)
325 defer cancelInit()
326 if err := agentSession.Init(initCtx, &agentv1.Init{
327 JobId: wid.String(),
328 CacheTrustedPublicKeys: append(slices.Clone(e.cfg.NixCache.TrustedPublicKeys), state.CacheTrustedPublicKeys...),
329 CacheReadProxyPort: readCache.Port(),
330 CacheUploadProxyPort: uploadCache.Port(),
331 DnsProxyPort: dnsProxy.Port(),
332 }); err != nil {
333 _ = agentSession.Close()
334 return err
335 }
336 state.Agent = agentSession
337 state.CID = cid
338 wf.Data = state
339
340 e.registerCleanup(wid, func(ctx context.Context) error {
341 return e.cleanupState(ctx, wid, state)
342 })
343 setupDone = true
344
345 fmt.Fprintf(wfLogger.DataWriter(-1, "stdout"),
346 "agent connected; serial log: %s\n", vm.Logs().Serial,
347 )
348 return nil
349}
350
351func applyDepsSource(command string) string {
352 return fmt.Sprintf(
353 // check if it exists because not all images have this
354 `if [ -f %s ]; then . %s; export PATH="$PATH:%s"; fi; %s`,
355 guestDevShellEnvPath, guestDevShellEnvPath, guestBasePATH, command,
356 )
357}
358
359func (e *Engine) RunStep(ctx context.Context, wid models.WorkflowId, w *models.Workflow, idx int, secrets []secrets.UnlockedSecret, wfLogger models.WorkflowLogger) error {
360 state, ok := w.Data.(*workflowState)
361 if !ok || state == nil || state.Agent == nil {
362 return fmt.Errorf("microVM workflow is not connected to agent")
363 }
364
365 stderr := wfLogger.DataWriter(idx, "stderr")
366
367 execCtx, vmExited, cancelWatch := watchVMExit(ctx, state.VM)
368 defer cancelWatch()
369
370 step := w.Steps[idx]
371 if s, ok := step.(Step); ok && s.action == activationStepAction {
372 err := e.activateConfig(execCtx, wid, state, s, wfLogger.DataWriter(idx, "stdout"))
373 return e.classifyStepError(ctx, wid, step, state, stderr, vmExited, "Failed to activate config", err)
374 }
375 env := []string{
376 "HOME=/workspace",
377 "LOGNAME=" + guestWorkflowUser,
378 "PATH=" + guestBasePATH,
379 "USER=" + guestWorkflowUser,
380 }
381 for k, v := range w.Environment {
382 env = append(env, k+"="+v)
383 }
384 for _, s := range secrets {
385 env = append(env, s.Key+"="+s.Value)
386 }
387 if s, ok := step.(Step); ok {
388 for k, v := range s.environment {
389 env = append(env, k+"="+v)
390 }
391 }
392
393 stdout := wfLogger.DataWriter(idx, "stdout")
394 exitCode, err := state.Agent.Exec(execCtx, AgentExec{
395 ID: fmt.Sprintf("%s-%d", wid.String(), idx),
396 ExecStart: &agentv1.ExecStart{
397 Argv: []string{state.ImageSpec.Shell, "-lc", applyDepsSource(step.Command())},
398 Env: env,
399 Cwd: guestWorkDir,
400 User: guestWorkflowUser,
401 // timeout not set here, Exec will fill it
402 },
403 Stdout: stdout,
404 Stderr: stderr,
405 })
406 if err != nil {
407 return e.classifyStepError(ctx, wid, step, state, stderr, vmExited, "User step error", err)
408 }
409
410 if exitCode != 0 {
411 e.l.Debug("step exited non-zero", "workflow", wid, "step", step.Name(), "exitCode", exitCode)
412 e.registerDebugTarget(wid, debugTarget{
413 cid: state.CID,
414 agent: state.Agent,
415 knot: w.Environment["TANGLED_REPO_KNOT"],
416 repoDid: w.Environment["TANGLED_REPO_REPO_DID"],
417 wfLogger: wfLogger,
418 stepCount: len(w.Steps),
419 maxAliveAt: state.StartedAt.Add(e.WorkflowTimeout()),
420 connected: make(chan struct{}),
421 released: make(chan struct{}),
422 })
423 return fmt.Errorf("User step error: exited with code %d", exitCode)
424 }
425 return nil
426}
427
428// reads the vm serial logs so we report the tail of that as an error instead of
429// just "guest agent connection lost: EOF"
430func (e *Engine) classifyStepError(ctx context.Context, wid models.WorkflowId, step models.Step, state *workflowState, stderr io.Writer, vmExited *atomic.Bool, category string, err error) error {
431 if err == nil {
432 return nil
433 }
434 l := e.l.With("workflow", wid, "step", step.Name())
435
436 if vmExited != nil && vmExited.Load() {
437 reason := "microVM exited unexpectedly"
438 oom := state.VM != nil && state.VM.OOMKilled()
439 if oom {
440 reason = "microVM killed by OOM (cgroup memory limit exceeded)"
441 }
442 if detail := vmCrashLog(state.VM); detail != "" {
443 fmt.Fprintf(stderr, "%s:\n%s\n", reason, detail)
444 l.Error(reason, "oom", oom, "detail", detail)
445 } else {
446 fmt.Fprintln(stderr, reason)
447 l.Error(reason, "oom", oom)
448 }
449 return fmt.Errorf("%s:\n%w", category, errors.New(reason+"; see workflow logs for serial output"))
450 }
451
452 if errors.Is(err, errGuestTimedOut) || ctx.Err() != nil {
453 l.Debug("step timed out", "guestReported", errors.Is(err, errGuestTimedOut))
454 return engine.ErrTimedOut
455 }
456
457 // the agent connection dropped while qemu stayed up (eg. the guest kernel
458 // OOM-killed the agent or a guest panic), so surface serial logs, those
459 // will be more helpful.
460 if detail := vmCrashLog(state.VM); detail != "" {
461 fmt.Fprintf(stderr, "step failed (%v):\n%s\n", err, detail)
462 l.Error("step failed", "error", err, "detail", detail)
463 } else {
464 l.Error("step failed", "error", err)
465 }
466 return fmt.Errorf("%s:\n%w", category, err)
467}
468
469func (e *Engine) activateConfig(ctx context.Context, wid models.WorkflowId, state *workflowState, step Step, out io.Writer) error {
470 cfg := step.config
471 if !cfg.Enabled() {
472 return nil
473 }
474
475 configKey := step.configKey
476 if configKey == "" {
477 configKey = state.ConfigKey
478 }
479
480 userConfigJSON, err := json.Marshal(cfg)
481 if err != nil {
482 return fmt.Errorf("encode user config: %w", err)
483 }
484
485 var cachedToplevel string
486 if configKey != "" {
487 if record, ok, err := state.NixOSToplevelCache.Lookup(configKey); err != nil {
488 return err
489 } else if ok {
490 // todo(dawn): we should probably use gc roots to eliminate TOCTOU
491 // the spindle will have to manage the gc roots, and for remote we have to
492 // ssh in to the host and add / remove gc root.
493 // we need to have this check anyway since the only check http caches can
494 // use is this one, since we cant manage gc roots there...
495 if e.anyCacheHasPath(ctx, state, record.Toplevel) {
496 cachedToplevel = record.Toplevel
497 fmt.Fprintf(out, "realizing cached NixOS config %s\n", cachedToplevel)
498 }
499 }
500 }
501 if cachedToplevel == "" {
502 fmt.Fprintf(out, "building NixOS config from user config\n")
503 }
504
505 baseHash, err := BaseConfigHash(state.ImageSpec)
506 if err != nil {
507 return fmt.Errorf("calculate base config hash: %w", err)
508 }
509
510 result, err := state.Agent.ActivateConfig(ctx, fmt.Sprintf("%s-config", wid.String()), &agentv1.ActivateConfig{
511 ConfigKey: configKey,
512 BaseConfigHash: baseHash,
513 UserConfig: string(userConfigJSON),
514 Toplevel: cachedToplevel,
515 }, out)
516 if err != nil {
517 return err
518 }
519 fmt.Fprintf(out, "activated NixOS config toplevel %s\n", result.Toplevel)
520
521 if cachedToplevel != "" || configKey == "" {
522 return nil
523 }
524 if e.cfg.NixCache.UploadURL == "" {
525 e.l.Warn("not committing config cache metadata: no upload URL configured", "workflow", wid, "configKey", configKey, "toplevel", result.Toplevel)
526 return nil
527 }
528
529 if err := e.drainNixCache(ctx, state); err != nil {
530 // a partial upload would leave the cache unable to realize this toplevel,
531 // so skip the metadata commit rather than poison it with an un-realizable
532 // key. the config still activated fine, so don't fail the workflow.
533 e.l.Warn("cache drain failed; skipping config cache metadata commit", "workflow", wid, "configKey", configKey, "toplevel", result.Toplevel, "error", err)
534 return nil
535 }
536 if err := state.NixOSToplevelCache.Commit(configKey, result.Toplevel); err != nil {
537 return err
538 }
539 fmt.Fprintf(out, "committed config cache metadata %s -> %s\n", configKey, result.Toplevel)
540 return nil
541}
542
543func (e *Engine) anyCacheHasPath(ctx context.Context, state *workflowState, storePath string) bool {
544 upstreams, err := BuildCacheUpstreams(e.cfg.NixCache.ReadURLs, state.CacheReadURLs)
545 if err != nil {
546 e.l.Warn("config cache check: build upstreams failed; treating as absent", "path", storePath, "error", err)
547 return false
548 }
549 if len(upstreams) == 0 {
550 return false
551 }
552 hash, _, err := parseStorePath(storePath)
553 if err != nil {
554 e.l.Warn("config cache check: invalid toplevel path; treating as absent", "path", storePath, "error", err)
555 return false
556 }
557 req, err := http.NewRequestWithContext(ctx, http.MethodHead, "http://upstream/"+hash+".narinfo", nil)
558 if err != nil {
559 e.l.Warn("config cache check: build request failed; treating as absent", "path", storePath, "error", err)
560 return false
561 }
562 resp, err := newNarinfoExistenceTransport(upstreams, e.l).RoundTrip(req)
563 if err != nil {
564 e.l.Warn("config cache check: narinfo probe failed; treating as absent", "path", storePath, "error", err)
565 return false
566 }
567 defer resp.Body.Close()
568 _, _ = io.Copy(io.Discard, resp.Body)
569 return resp.StatusCode == http.StatusOK
570}
571
572func (e *Engine) DestroyWorkflow(ctx context.Context, wid models.WorkflowId) error {
573 if e.cfg.MicroVMPipelines.DebugSSH.Enabled {
574 // keep a failed VM alive for the grace period before we tear it down
575 e.maybeRetainForDebug(ctx, wid)
576 }
577
578 fns := e.drainCleanups(wid)
579
580 var cleanupErr error
581 for i := len(fns) - 1; i >= 0; i-- {
582 if err := fns[i](ctx); err != nil {
583 e.l.Error("failed to cleanup workflow resource", "workflowId", wid, "error", err)
584 cleanupErr = errors.Join(cleanupErr, err)
585 }
586 }
587 return cleanupErr
588}
589
590func (e *Engine) WorkflowTimeout() time.Duration {
591 d, err := time.ParseDuration(e.cfg.MicroVMPipelines.WorkflowTimeout)
592 if err != nil {
593 d = 5 * time.Minute
594 }
595 return d + guestTimeoutGrace
596}
597
598func (e *Engine) registerCleanup(wid models.WorkflowId, fn cleanupFunc) {
599 e.cleanupMu.Lock()
600 defer e.cleanupMu.Unlock()
601 key := wid.String()
602 e.cleanup[key] = append(e.cleanup[key], fn)
603}
604
605func (e *Engine) drainCleanups(wid models.WorkflowId) []cleanupFunc {
606 e.cleanupMu.Lock()
607 defer e.cleanupMu.Unlock()
608 key := wid.String()
609 fns := e.cleanup[key]
610 delete(e.cleanup, key)
611 return fns
612}
613
614func (e *Engine) cgroupLimits(wid models.WorkflowId, spec ImageSpec) CgroupLimits {
615 cfg := e.cfg.MicroVMPipelines
616 return CgroupLimits{
617 Enabled: cfg.EnableCgroups,
618 Parent: e.cgroupParent,
619 Name: "workflow-" + wid.String(),
620 MemoryMaxMiB: resourcesForImage(spec).MemoryMiB,
621 SwapMaxMiB: cfg.CgroupSwapMaxMiB,
622 PidsMax: cfg.CgroupPidsMax,
623 }
624}