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