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