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