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