From 4b16037d0bc7e7da710a2ef27ac593c13139c86d Mon Sep 17 00:00:00 2001 From: dawn <90008@gaze.systems> Date: Fri, 3 Apr 2026 16:21:10 +0300 Subject: [PATCH] spindle/qemu,nix: use qemu guest agent to execute commands instead of ssh Signed-off-by: dawn <90008@gaze.systems> --- cmd/spindle/main.go | 20 +- flake.nix | 2 +- spindle/config/config.go | 2 +- spindle/engines/qemu/bakers/alpine.go | 41 ++- spindle/engines/qemu/bakers/baker.go | 38 ++- spindle/engines/qemu/bakers/nbd.go | 54 +++- spindle/engines/qemu/bakers/ubuntu.go | 96 +++++- spindle/engines/qemu/cloudinit.go | 15 +- spindle/engines/qemu/engine.go | 415 +++++++++++++++----------- spindle/engines/qemu/logs.go | 35 --- spindle/engines/qemu/models.go | 25 +- spindle/engines/qemu/setup.go | 23 +- 12 files changed, 480 insertions(+), 286 deletions(-) delete mode 100644 spindle/engines/qemu/logs.go diff --git a/cmd/spindle/main.go b/cmd/spindle/main.go index 2c58dc62..2fbb2b69 100644 --- a/cmd/spindle/main.go +++ b/cmd/spindle/main.go @@ -9,7 +9,6 @@ import ( "github.com/urfave/cli/v3" tlog "tangled.org/core/log" "tangled.org/core/spindle" - "tangled.org/core/spindle/config" "tangled.org/core/spindle/engines/qemu" "tangled.org/core/spindle/engines/qemu/bakers" ) @@ -52,8 +51,18 @@ func SetupQemuCommand() *cli.Command { return &cli.Command{ Name: "setup-qemu", Usage: "downloads and prepares default qemu engine images", + Flags: []cli.Flag{ + &cli.StringFlag{ + Name: "image-dir", + Usage: "directory to store images", + }, + }, Action: func(ctx context.Context, cmd *cli.Command) error { - return qemu.SetupDefaultImages(ctx) + imageDir := cmd.String("image-dir") + if imageDir == "" { + imageDir = os.Getenv("SPINDLE_QEMU_PIPELINES_IMAGE_DIR") + } + return qemu.SetupDefaultImages(ctx, imageDir) }, } } @@ -63,15 +72,12 @@ func BakeImageCommand() *cli.Command { Name: "bake-image", Usage: "(needs root) prepare a downloaded qemu image (e.g. extract kernel/initrd etc.)", Action: func(ctx context.Context, cmd *cli.Command) error { - cfg, err := config.Load(ctx) - if err != nil { - return fmt.Errorf("loading config: %w", err) - } + imageDir := os.Getenv("SPINDLE_QEMU_PIPELINES_IMAGE_DIR") name := cmd.Args().First() if name == "" { return fmt.Errorf("name argument (e.g. 'alpine') is required") } - return bakers.PrepareImage(ctx, cfg, name) + return bakers.PrepareImage(ctx, imageDir, name) }, } } diff --git a/flake.nix b/flake.nix index 8e3ad4ec..f82ce1f7 100644 --- a/flake.nix +++ b/flake.nix @@ -215,8 +215,8 @@ pkgs.worker-build pkgs.cargo-generate pkgs.qemu - pkgs.cloud-utils pkgs.cdrkit + pkgs.parted (fenix.packages.${system}.combine [ fenix.packages.${system}.stable.cargo fenix.packages.${system}.stable.rustc diff --git a/spindle/config/config.go b/spindle/config/config.go index f5669357..dd6d0ea0 100644 --- a/spindle/config/config.go +++ b/spindle/config/config.go @@ -59,7 +59,7 @@ type S3 struct { } type QemuPipelines struct { - ImageDir string `env:"IMAGE_DIR, required"` + ImageDir string `env:"IMAGE_DIR, required"` // SPINDLE_QEMU_PIPELINES_IMAGE_DIR; also used in cmd/spindle/main.go OverlayDir string `env:"OVERLAY_DIR, default="` // where qemu snapshot overlays will live DefaultImage string `env:"DEFAULT_IMAGE, default=ubuntu-24.04"` EnableKVM bool `env:"ENABLE_KVM, default=true"` diff --git a/spindle/engines/qemu/bakers/alpine.go b/spindle/engines/qemu/bakers/alpine.go index 7e05a635..46a8bf4f 100644 --- a/spindle/engines/qemu/bakers/alpine.go +++ b/spindle/engines/qemu/bakers/alpine.go @@ -37,7 +37,12 @@ func copyFile(src, dst string) error { func (p *AlpineBaker) Prepare(ctx context.Context, dir string) error { l := log.FromContext(ctx) - diskPath := filepath.Join(dir, "disk.qcow2") + diskPath := DiskPath(dir) + + // check that we have all the required files + if _, err := os.Stat(diskPath); os.IsNotExist(err) { + return fmt.Errorf("missing '%s' in %s", DiskName, dir) + } nbd, err := p.nbd.Connect(ctx, l, diskPath) if err != nil { @@ -52,20 +57,32 @@ func (p *AlpineBaker) Prepare(ctx context.Context, dir string) error { } defer nbd.Unmount(mntDir) - resolvDest := filepath.Join(mntDir, "etc/resolv.conf") - if err := os.WriteFile(resolvDest, nil, 0644); err != nil { - return fmt.Errorf("creating resolv.conf mount point: %w", err) + cleanup, err := bindMountResolvConf(ctx, nbd, mntDir) + if err != nil { + return err } - if out, err := exec.CommandContext(ctx, "mount", "--bind", "/etc/resolv.conf", resolvDest).CombinedOutput(); err != nil { - return fmt.Errorf("bind mount resolv.conf: %s: %w", string(out), err) + defer cleanup() + + run := func(command string, args ...string) ([]byte, error) { + fullArgs := append([]string{mntDir, command}, args...) + cmd := exec.CommandContext(ctx, "chroot", fullArgs...) + out, err := cmd.CombinedOutput() + if err != nil { + return nil, fmt.Errorf("%s: %s: %w", command, string(out), err) + } + return out, nil } - defer nbd.Unmount(resolvDest) l.Info("installing packages...") - if out, err := exec.CommandContext(ctx, "chroot", mntDir, "/sbin/apk", "add", "--no-cache", "git", "curl").CombinedOutput(); err != nil { + if out, err := run("/sbin/apk", "add", "--no-cache", "git", "curl", "qemu-guest-agent"); err != nil { return fmt.Errorf("installing packages: %s: %w", string(out), err) } + l.Info("enabling qemu-guest-agent...") + if out, err := run("/sbin/rc-update", "add", "qemu-guest-agent", "default"); err != nil { + return fmt.Errorf("enabling qemu-guest-agent: %s: %w", string(out), err) + } + l.Info("extracting kernel and initrd...") bootDir := filepath.Join(mntDir, "boot") kernelSource := filepath.Join(bootDir, "vmlinuz-virt") @@ -77,16 +94,16 @@ func (p *AlpineBaker) Prepare(ctx context.Context, dir string) error { return fmt.Errorf("initrd 'initramfs-virt' not found in %s", bootDir) } - if err := copyFile(kernelSource, filepath.Join(dir, "kernel")); err != nil { + if err := copyFile(kernelSource, KernelPath(dir)); err != nil { return fmt.Errorf("extracting kernel: %w", err) } - if err := copyFile(initrdSource, filepath.Join(dir, "initrd")); err != nil { + if err := copyFile(initrdSource, InitrdPath(dir)); err != nil { return fmt.Errorf("extracting initrd: %w", err) } - configPath := filepath.Join(dir, "config.json") + configPath := ConfigPath(dir) metadata := ImageMetadata{ - Cmdline: "root=/dev/vda rootfstype=ext4 rw console=ttyS0 modules=sd-mod,usb-storage,ext4,virtio_pci,virtio_blk", + Cmdline: "root=/dev/vda rootfstype=ext4 rw console=ttyS0 modules=ext4,virtio_pci,virtio_blk", Shell: "/bin/ash", } b, err := json.MarshalIndent(metadata, "", " ") diff --git a/spindle/engines/qemu/bakers/baker.go b/spindle/engines/qemu/bakers/baker.go index 3469159b..293b4fbf 100644 --- a/spindle/engines/qemu/bakers/baker.go +++ b/spindle/engines/qemu/bakers/baker.go @@ -9,9 +9,24 @@ import ( "strings" "tangled.org/core/log" - "tangled.org/core/spindle/config" ) +const ( + DiskName = "disk" + KernelName = "kernel" + InitrdName = "initrd" + ConfigName = "config.json" + UserDataName = "user-data.yml" + SeedISOName = "cloud-init.iso" +) + +func DiskPath(dir string) string { return filepath.Join(dir, DiskName) } +func KernelPath(dir string) string { return filepath.Join(dir, KernelName) } +func InitrdPath(dir string) string { return filepath.Join(dir, InitrdName) } +func ConfigPath(dir string) string { return filepath.Join(dir, ConfigName) } +func UserDataPath(dir string) string { return filepath.Join(dir, UserDataName) } +func SeedISOPath(dir string) string { return filepath.Join(dir, SeedISOName) } + type ImageMetadata struct { Cmdline string `json:"cmdline"` Shell string `json:"shell"` @@ -25,14 +40,14 @@ var ( nbd = &NBDManager{} preparers = map[string]ImageBaker{ "alpine": &AlpineBaker{nbd: nbd}, - "ubuntu": &UbuntuBaker{}, + "ubuntu": &UbuntuBaker{nbd: nbd}, } ) -func PrepareImage(ctx context.Context, cfg *config.Config, name string) error { +func PrepareImage(ctx context.Context, imageDir string, name string) error { name = strings.ToLower(name) - imageDir := filepath.Join(cfg.QemuPipelines.ImageDir, name) + imageDir = filepath.Join(imageDir, name) if _, err := os.Stat(imageDir); os.IsNotExist(err) { return fmt.Errorf("image directory %s does not exist; did you run 'spindle setup'?", imageDir) } @@ -40,7 +55,7 @@ func PrepareImage(ctx context.Context, cfg *config.Config, name string) error { l := log.FromContext(ctx) l.Info("preparing image", "name", name) - diskPath := filepath.Join(imageDir, "disk.qcow2") + diskPath := DiskPath(imageDir) // todo(dawn): this should be configurable // cloud-init will expand the partitions itself! so we don't have to do anything else l.Info("resizing disk to 8G...", "path", diskPath) @@ -57,3 +72,16 @@ func PrepareImage(ctx context.Context, cfg *config.Config, name string) error { return nil } + +// bind mount resolve conf for networking +func bindMountResolvConf(ctx context.Context, nbd *NBDHandle, mntDir string) (func(), error) { + resolvDest := filepath.Join(mntDir, "etc/resolv.conf") + _ = os.Remove(resolvDest) // remove if it's a symlink or existing file + if err := os.WriteFile(resolvDest, nil, 0644); err != nil { + return nil, fmt.Errorf("creating resolv.conf mount point: %w", err) + } + if out, err := exec.CommandContext(ctx, "mount", "--bind", "/etc/resolv.conf", resolvDest).CombinedOutput(); err != nil { + return nil, fmt.Errorf("bind mount resolv.conf: %s: %w", string(out), err) + } + return func() { nbd.Unmount(resolvDest) }, nil +} diff --git a/spindle/engines/qemu/bakers/nbd.go b/spindle/engines/qemu/bakers/nbd.go index c39524d0..90d1d683 100644 --- a/spindle/engines/qemu/bakers/nbd.go +++ b/spindle/engines/qemu/bakers/nbd.go @@ -6,12 +6,14 @@ import ( "log/slog" "os" "os/exec" + "strings" "sync" "time" ) type NBDManager struct { - mu sync.Mutex + mu sync.Mutex + cleanupOnce sync.Once } type NBDHandle struct { @@ -21,9 +23,16 @@ type NBDHandle struct { } func (n *NBDManager) Connect(ctx context.Context, l *slog.Logger, diskPath string) (*NBDHandle, error) { + // cleanup in case we have failed state + n.cleanupOnce.Do(func() { + n.Cleanup(ctx, l) + }) + n.mu.Lock() - if _, err := os.Stat("/dev/nbd0"); os.IsNotExist(err) { + device := "/dev/nbd12" // use something other than nbd0 + + if _, err := os.Stat(device); os.IsNotExist(err) { l.Info("nbd module not loaded, trying to load it...") if out, err := exec.Command("modprobe", "nbd", "max_part=8").CombinedOutput(); err != nil { n.mu.Unlock() @@ -32,7 +41,7 @@ func (n *NBDManager) Connect(ctx context.Context, l *slog.Logger, diskPath strin } l.Info("connecting disk via nbd...", "disk", diskPath) - if out, err := exec.CommandContext(ctx, "qemu-nbd", "--connect=/dev/nbd0", diskPath).CombinedOutput(); err != nil { + if out, err := exec.CommandContext(ctx, "qemu-nbd", "--connect="+device, diskPath).CombinedOutput(); err != nil { n.mu.Unlock() return nil, fmt.Errorf("qemu-nbd connect: %s: %w", string(out), err) } @@ -40,18 +49,21 @@ func (n *NBDManager) Connect(ctx context.Context, l *slog.Logger, diskPath strin waitCtx, cancel := context.WithTimeout(ctx, 3*time.Second) defer cancel() for { - if _, err := os.Stat("/dev/nbd0"); err == nil { + if _, err := os.Stat(device); err == nil { break } select { case <-waitCtx.Done(): n.mu.Unlock() - return nil, fmt.Errorf("timeout waiting for /dev/nbd0 to appear") + return nil, fmt.Errorf("timeout waiting for %s to appear", device) case <-time.After(50 * time.Millisecond): } } - return &NBDHandle{mgr: n, device: "/dev/nbd0", l: l}, nil + // sleep 100ms just to make sure... + time.Sleep(100 * time.Millisecond) + + return &NBDHandle{mgr: n, device: device, l: l}, nil } // mounts the device node specified @@ -89,3 +101,33 @@ func (h *NBDHandle) Close() { h.l.Error("qemu-nbd disconnect failed", "error", err, "output", string(out)) } } + +func (n *NBDManager) Cleanup(ctx context.Context, l *slog.Logger) { + l.Info("cleaning up stale nbd state...") + + // 1. disconnect nbd0 + // do this first to make sure the device is not busy when we try to umount + _ = exec.CommandContext(ctx, "qemu-nbd", "--disconnect", "/dev/nbd0").Run() + + // 2. find and unmount stale spindle-mnt mounts + data, err := os.ReadFile("/proc/mounts") + if err != nil { + l.Error("failed to read /proc/mounts", "error", err) + return + } + + for _, line := range strings.Split(string(data), "\n") { + fields := strings.Fields(line) + if len(fields) < 2 { + continue + } + mnt := fields[1] + if strings.Contains(mnt, "spindle-mnt-") { + l.Info("unmounting stale mount", "path", mnt) + if out, err := exec.CommandContext(ctx, "umount", "-l", mnt).CombinedOutput(); err != nil { + l.Error("umount failed", "path", mnt, "error", err, "output", string(out)) + } + os.RemoveAll(mnt) + } + } +} diff --git a/spindle/engines/qemu/bakers/ubuntu.go b/spindle/engines/qemu/bakers/ubuntu.go index af14c671..228e827a 100644 --- a/spindle/engines/qemu/bakers/ubuntu.go +++ b/spindle/engines/qemu/bakers/ubuntu.go @@ -5,32 +5,108 @@ import ( "encoding/json" "fmt" "os" + "os/exec" "path/filepath" + "time" "tangled.org/core/log" ) -type UbuntuBaker struct{} +type UbuntuBaker struct { + nbd *NBDManager +} func (p *UbuntuBaker) Prepare(ctx context.Context, dir string) error { l := log.FromContext(ctx) - _, kerr := os.Stat(filepath.Join(dir, "kernel")) - _, ierr := os.Stat(filepath.Join(dir, "initrd")) - // ubuntu preparer assumes kernel and initrd are already downloaded - if os.IsNotExist(kerr) || os.IsNotExist(ierr) { - return fmt.Errorf("missing 'kernel' and/or 'initrd' in %s; please use ubuntu cloud images that provide separate kernel/initrd files (see 'unpacked/' on cloud-images.ubuntu.com)", dir) + // check that we have all the required files + for _, path := range []string{DiskName, KernelName, InitrdName} { + if _, err := os.Stat(filepath.Join(dir, path)); os.IsNotExist(err) { + return fmt.Errorf("missing '%s' in %s", path, dir) + } + } + + diskPath := DiskPath(dir) + + nbd, err := p.nbd.Connect(ctx, l, diskPath) + if err != nil { + return err + } + defer nbd.Close() + + // qemu-nbd might not have scanned partitions yet + if out, err := exec.CommandContext(ctx, "partprobe", nbd.Device()).CombinedOutput(); err != nil { + l.Warn("partprobe failed", "error", err, "output", string(out)) + } + _ = exec.CommandContext(ctx, "udevadm", "settle").Run() + + // wait for the partition to appear + partition := nbd.Device() + "p1" + waitCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + for { + if _, err := os.Stat(partition); err == nil { + break + } + select { + case <-waitCtx.Done(): + return fmt.Errorf("timeout waiting for partition %s to appear", partition) + case <-time.After(100 * time.Millisecond): + } } - configPath := filepath.Join(dir, "config.json") + mntDir, err := nbd.Mount(ctx, partition) + if err != nil { + return err + } + defer nbd.Unmount(mntDir) + + // run := func(command string, args ...string) ([]byte, error) { + // fullArgs := append([]string{mntDir, "/usr/bin/" + command}, args...) + // cmd := exec.CommandContext(ctx, "chroot", fullArgs...) + // cmd.Env = append(os.Environ(), "DEBIAN_FRONTEND=noninteractive") + // out, err := cmd.CombinedOutput() + // if err != nil { + // return nil, fmt.Errorf("chroot run %s: %s: %w", command, string(out), err) + // } + // return out, nil + // } + + // cleanup, err := bindMountResolvConf(ctx, nbd, mntDir) + // if err != nil { + // return err + // } + // defer cleanup() + + // todo(dawn): i tried doing apt stuff in chroot here but no budge :/ + // i didnt want to hardcode in a qemu-guest-agent either nor replicating apt repo setup... + // so instead we just write a later user-data file that does it for us at boot. + + configPath := ConfigPath(dir) metadata := ImageMetadata{ Cmdline: "root=LABEL=cloudimg-rootfs rw console=ttyS0", Shell: "/bin/bash", } - if b, err := json.MarshalIndent(metadata, "", " "); err == nil { - _ = os.WriteFile(configPath, b, 0644) + b, err := json.MarshalIndent(metadata, "", " ") + if err != nil { + return fmt.Errorf("marshaling image metadata: %w", err) + } + if err := os.WriteFile(configPath, b, 0644); err != nil { + return fmt.Errorf("writing image metadata: %w", err) + } + + // todo(dawn): this should be done while preparing the image... + userDataPath := UserDataPath(dir) + userData := ` +packages: + - qemu-guest-agent +runcmd: + - systemctl enable --now qemu-guest-agent +` + if err := os.WriteFile(userDataPath, []byte(userData), 0644); err != nil { + return fmt.Errorf("writing user-data: %w", err) } - l.Info("ubuntu configuration created") + l.Info("ubuntu preparation complete") return nil } diff --git a/spindle/engines/qemu/cloudinit.go b/spindle/engines/qemu/cloudinit.go index 846c3b6c..3af708d7 100644 --- a/spindle/engines/qemu/cloudinit.go +++ b/spindle/engines/qemu/cloudinit.go @@ -5,18 +5,21 @@ import ( "os" "os/exec" "path/filepath" + + "tangled.org/core/spindle/engines/qemu/bakers" ) -func generateSeedISO(dir string, enginePubKey string) error { +func generateSeedISO(dir string, extraUserData string) error { metaData := "instance-id: spindle-vm\nlocal-hostname: spindle\n" - userData := fmt.Sprintf(`#cloud-config + userData := `#cloud-config users: - name: build sudo: ALL=(ALL) NOPASSWD:ALL shell: /bin/sh - ssh_authorized_keys: - - %s -`, enginePubKey) +` + if extraUserData != "" { + userData += extraUserData + } metaDataPath := filepath.Join(dir, "meta-data") if err := os.WriteFile(metaDataPath, []byte(metaData), 0o644); err != nil { @@ -29,7 +32,7 @@ users: } // nocloud source expects volid "cidata" with joliet and rock ridge extensions - cmd := exec.Command("genisoimage", "-output", filepath.Join(dir, "cloud-init.iso"), "-volid", "cidata", "-joliet", "-rock", "meta-data", "user-data") + cmd := exec.Command("genisoimage", "-output", bakers.SeedISOPath(dir), "-volid", "cidata", "-joliet", "-rock", "meta-data", "user-data") cmd.Dir = dir cmd.Stdout = os.Stdout cmd.Stderr = os.Stderr diff --git a/spindle/engines/qemu/engine.go b/spindle/engines/qemu/engine.go index 23aeb474..be510ae1 100644 --- a/spindle/engines/qemu/engine.go +++ b/spindle/engines/qemu/engine.go @@ -1,9 +1,9 @@ package qemu import ( + "bufio" "context" - "crypto/ed25519" - "crypto/rand" + "encoding/base64" "encoding/json" "fmt" "log/slog" @@ -15,7 +15,6 @@ import ( "time" "github.com/digitalocean/go-qemu/qmp" - "golang.org/x/crypto/ssh" "gopkg.in/yaml.v3" "tangled.org/core/api/tangled" @@ -27,14 +26,6 @@ import ( "tangled.org/core/spindle/secrets" ) -type ResolvedImage struct { - kernel string - initrd string - disk string - cmdline string - shell string -} - type cleanupFunc func(context.Context) error type Engine struct { @@ -63,10 +54,6 @@ func (s Step) Name() string { return s.name } func (s Step) Command() string { return s.command } func (s Step) Kind() models.StepKind { return s.kind } -type setupSteps []models.Step - -func (ss *setupSteps) addStep(step models.Step) { *ss = append(*ss, step) } - func (e *Engine) InitWorkflow(twf tangled.Pipeline_Workflow, tpl tangled.Pipeline) (*models.Workflow, error) { swf := &models.Workflow{} var dwf manifestWorkflow @@ -86,28 +73,19 @@ func (e *Engine) InitWorkflow(twf tangled.Pipeline_Workflow, tpl tangled.Pipelin swf.Name = twf.Name swf.Environment = dwf.Environment - setup := &setupSteps{} - setup.addStep(models.BuildCloneStep(twf, *tpl.TriggerMetadata, e.cfg.Server.Dev)) - - swf.Steps = append(*setup, swf.Steps...) + setup := []models.Step{models.BuildCloneStep(twf, *tpl.TriggerMetadata, e.cfg.Server.Dev)} + swf.Steps = append(setup, swf.Steps...) img, err := e.resolveImage(dwf.Image) if err != nil { return nil, err } - swf.Data = vmState{ - kernel: img.kernel, - initrd: img.initrd, - disk: img.disk, - cmdline: img.cmdline, - shell: img.shell, - } - + swf.Data = vmState{img: img} return swf, nil } -// discover and resolve kernel, initrd, and disk from an image subfolder +// discover and resolve kernel, initrd, and disk and other config from an image subfolder func (e *Engine) resolveImage(name string) (ResolvedImage, error) { var img ResolvedImage if name == "" { @@ -118,10 +96,10 @@ func (e *Engine) resolveImage(name string) (ResolvedImage, error) { } imageDir := filepath.Join(e.cfg.QemuPipelines.ImageDir, name) - kernelPath := filepath.Join(imageDir, "kernel") - initrdPath := filepath.Join(imageDir, "initrd") - diskPath := filepath.Join(imageDir, "disk.qcow2") - configPath := filepath.Join(imageDir, "config.json") + kernelPath := bakers.KernelPath(imageDir) + initrdPath := bakers.InitrdPath(imageDir) + diskPath := bakers.DiskPath(imageDir) + configPath := bakers.ConfigPath(imageDir) if _, err := os.Stat(diskPath); err == nil { img.disk = diskPath @@ -143,9 +121,12 @@ func (e *Engine) resolveImage(name string) (ResolvedImage, error) { } } } + if b, err := os.ReadFile(filepath.Join(imageDir, bakers.UserDataName)); err == nil { + img.userData = string(b) + } if img.disk == "" { - return img, fmt.Errorf("missing 'disk.qcow2' in %s", imageDir) + return img, fmt.Errorf("missing '%s' in %s", bakers.DiskName, imageDir) } if img.kernel != "" && (img.initrd == "" || img.cmdline == "") { return img, fmt.Errorf("kernel requires initrd and cmdline, but 'initrd' and/or 'cmdline' is missing for %s", name) @@ -160,30 +141,13 @@ func (e *Engine) SetupWorkflow(ctx context.Context, wid models.WorkflowId, wf *m l := e.l.With("workflow", wid) l.Info("setting up qemu workflow") - setupStep := Step{name: "qemu vm setup", kind: models.StepKindSystem} - setupStepIdx := -1 + const setupStepIdx = -1 + setupStep := Step{name: "QEMU VM setup", kind: models.StepKindSystem} wfLogger.ControlWriter(setupStepIdx, setupStep, models.StepStatusStart).Write([]byte{0}) defer wfLogger.ControlWriter(setupStepIdx, setupStep, models.StepStatusEnd).Write([]byte{0}) - // generate ed25519 keypair for the engine to ssh into the guest - pubKey, privKey, err := ed25519.GenerateKey(rand.Reader) - if err != nil { - return fmt.Errorf("generating keypair: %w", err) - } - - sshSigner, err := ssh.NewSignerFromKey(privKey) - if err != nil { - return err - } - sshPubKey, err := ssh.NewPublicKey(pubKey) - if err != nil { - return err - } - enginePubKeyStr := string(ssh.MarshalAuthorizedKey(sshPubKey)) - - // the tempdir is configurable since some systems may have tmpfs as /tmp, - // which is not ideal if a workflow uses a lot of space. + // some systems have tmpfs at /tmp which is not ideal for large workloads targetTempDir := e.cfg.QemuPipelines.OverlayDir if targetTempDir == "" { targetTempDir = os.TempDir() @@ -198,23 +162,15 @@ func (e *Engine) SetupWorkflow(ctx context.Context, wid models.WorkflowId, wf *m }) state := wf.Data.(vmState) + img := state.img - // generate nocloud seed iso containing engine's authorized_keys - if err := generateSeedISO(tempDir, enginePubKeyStr); err != nil { + if err := generateSeedISO(tempDir, img.userData); err != nil { return fmt.Errorf("generating seed iso: %w", err) } - // note: there is an inherent TOCTOU race here between releasing the port - // and qemu binding it. nothing we can do about this in userspace without - // privilege, but in practice it's rarely an issue. - nl, err := net.Listen("tcp", "127.0.0.1:0") - if err != nil { - return fmt.Errorf("finding free port: %w", err) - } - sshPort := nl.Addr().(*net.TCPAddr).Port - nl.Close() - qmpSock := filepath.Join(tempDir, "qmp.sock") + qgaSock := filepath.Join(tempDir, "qga.sock") + // todo(dawn): ideally would be nice if we used qemu with the microvm enabled here... // but that is not compatible with cloud-init since it expects real hw enumeration... // and we would not be able to use standard cloud images, which is kind of annoying. @@ -227,14 +183,16 @@ func (e *Engine) SetupWorkflow(ctx context.Context, wid models.WorkflowId, wf *m // todo(dawn): ideally probably have "tiers" and let the spindle concile the tier using // what the user wants and what the user has exposed to them by the spindle operator? "-m", e.cfg.QemuPipelines.Memory, "-smp", fmt.Sprintf("%d", e.cfg.QemuPipelines.SMP), - "-display", "none", "-nodefaults", "-no-user-config", + "-display", "none", "-monitor", "none", "-nodefaults", "-no-user-config", // use snapshot=on to do copy-on-write without having us manage qcow overlays manually - "-drive", fmt.Sprintf("file=%s,media=disk,snapshot=on,if=virtio", state.disk), - "-drive", fmt.Sprintf("file=%s,media=cdrom", filepath.Join(tempDir, "cloud-init.iso")), - "-netdev", fmt.Sprintf("user,id=net0,hostfwd=tcp:127.0.0.1:%d-:22", sshPort), + "-drive", fmt.Sprintf("file=%s,media=disk,snapshot=on,if=virtio", img.disk), + "-drive", fmt.Sprintf("file=%s,media=cdrom", bakers.SeedISOPath(tempDir)), + "-netdev", "user,id=net0", "-device", "virtio-net-pci,netdev=net0", "-qmp", fmt.Sprintf("unix:%s,server,nowait", qmpSock), - "-monitor", "none", + "-chardev", fmt.Sprintf("socket,path=%s,server,nowait,id=qga0", qgaSock), + "-device", "virtio-serial", + "-device", "virtserialport,chardev=qga0,name=org.qemu.guest_agent.0", } if e.cfg.Server.Dev { @@ -245,16 +203,16 @@ func (e *Engine) SetupWorkflow(ctx context.Context, wid models.WorkflowId, wf *m // support booting using qemu bios still, but otherwise we make it faster! // incase someone wants to do this for whatever reason... - if state.kernel != "" { - argv = append(argv, "-kernel", state.kernel) - if state.initrd != "" { - argv = append(argv, "-initrd", state.initrd) + if img.kernel != "" { + argv = append(argv, "-kernel", img.kernel) + if img.initrd != "" { + argv = append(argv, "-initrd", img.initrd) } - if state.cmdline != "" { - argv = append(argv, "-append", state.cmdline) + if img.cmdline != "" { + argv = append(argv, "-append", img.cmdline) } } else { - // if we are booting with bios, we need to tell it to boot from the disk + // booting with bios: explicitly select disk argv = append(argv, "-boot", "order=c") } @@ -276,16 +234,28 @@ func (e *Engine) SetupWorkflow(ctx context.Context, wid models.WorkflowId, wf *m qemuCmd.Stdout = os.Stdout qemuCmd.Stderr = os.Stderr - startBoot := time.Now() + startedBootAt := time.Now() if err := qemuCmd.Start(); err != nil { return fmt.Errorf("starting qemu: %w", err) } var mon *qmp.SocketMonitor + + // cleanup qemu if we fail to setup at any point below + setupOk := false + defer func() { + if !setupOk { + _ = qemuCmd.Process.Kill() + if mon != nil { + _ = mon.Disconnect() + } + } + }() + qmpCtx, cancelQmp := context.WithTimeout(ctx, 10*time.Second) defer cancelQmp() - // retry qmp connect until the emulator is ready with its unix socket + // wait for qmp to be ready for { mon, err = qmp.NewSocketMonitor("unix", qmpSock, 2*time.Second) if err == nil { @@ -295,77 +265,60 @@ func (e *Engine) SetupWorkflow(ctx context.Context, wid models.WorkflowId, wf *m } select { case <-qmpCtx.Done(): - _ = qemuCmd.Process.Kill() return fmt.Errorf("qmp connect timeout: %w", err) case <-time.After(10 * time.Millisecond): } } - // check if the guest is running - raw, err := mon.Run([]byte(`{"execute":"query-status"}`)) + status, err := e.qmpQueryStatus(mon) if err != nil { - _ = qemuCmd.Process.Kill() - _ = mon.Disconnect() - return fmt.Errorf("qmp query-status failed: %w", err) - } - var resp map[string]any - if err := json.Unmarshal(raw, &resp); err != nil { - _ = qemuCmd.Process.Kill() - _ = mon.Disconnect() - return fmt.Errorf("qmp query-status parse failed: %w", err) + return err } - status, _ := resp["return"].(map[string]any)["status"].(string) + l.Info("qemu guest status", "status", status) if status != "running" { - _ = qemuCmd.Process.Kill() - _ = mon.Disconnect() return fmt.Errorf("qemu guest not running (status: %s)", status) } - sshConfig := &ssh.ClientConfig{ - User: "build", - Auth: []ssh.AuthMethod{ssh.PublicKeys(sshSigner)}, - HostKeyCallback: ssh.InsecureIgnoreHostKey(), - Timeout: 5 * time.Second, - } - - bootCtx, cancelBoot := context.WithTimeout(ctx, 90*time.Second) - defer cancelBoot() + qgaCtx, cancelQga := context.WithTimeout(ctx, time.Minute) + defer cancelQga() - // backoff until the guest finishes booting and starts sshd - var sshClient *ssh.Client + // wait for guest agent to be ready (aka boot) for { - sshClient, err = ssh.Dial("tcp", fmt.Sprintf("127.0.0.1:%d", sshPort), sshConfig) - if err == nil { - bootDuration := time.Since(startBoot).Round(time.Millisecond) - l.Debug("vm booted and ssh ready", "elapsed", bootDuration) - break + if _, err := os.Stat(qgaSock); err == nil { + pingCtx, cancelPing := context.WithTimeout(qgaCtx, 5*time.Second) + err = e.qgaGuestPing(pingCtx, qgaSock) + cancelPing() + if err == nil { + l.Info("vm booted and guest-agent ready", "elapsed", time.Since(startedBootAt).Round(time.Millisecond)) + break + } + l.Debug("qga guest-ping failed", "error", err) + } else { + l.Debug("qga socket not found yet", "path", qgaSock) } + select { - case <-bootCtx.Done(): - _ = qemuCmd.Process.Kill() - _ = mon.Disconnect() - return ErrBootTimeout + case <-qgaCtx.Done(): + return fmt.Errorf("qga connect timeout: %w", err) case <-time.After(100 * time.Millisecond): } } e.registerCleanup(wid, func(ctx context.Context) error { - _ = sshClient.Close() - // graceful powerdown so guest can sync filesystem - _, _ = mon.Run([]byte(`{"execute": "system_powerdown"}`)) + if err := e.qmpSystemPowerdown(mon); err != nil { + l.Error("failed to powerdown qemu guest", "workflow", wid, "error", err) + } done := make(chan error, 1) - go func() { - done <- qemuCmd.Wait() - }() + go func() { done <- qemuCmd.Wait() }() + select { case <-done: - case <-time.After(time.Second): + case <-time.After(5 * time.Second): _ = qemuCmd.Process.Kill() - // drain Wait after kill to avoid zombie - <-done + <-done // drain to avoid zombie } _ = mon.Disconnect() @@ -373,84 +326,196 @@ func (e *Engine) SetupWorkflow(ctx context.Context, wid models.WorkflowId, wf *m }) wf.Data = vmState{ - process: qemuCmd.Process, - qmpMon: mon, - sshClient: sshClient, - sshPort: sshPort, - tempDir: tempDir, - kernel: state.kernel, - initrd: state.initrd, - disk: state.disk, - shell: state.shell, + process: qemuCmd.Process, + qmpMon: mon, + qgaPath: qgaSock, + tempDir: tempDir, + img: img, } + setupOk = true return nil } -func (e *Engine) RunStep(ctx context.Context, wid models.WorkflowId, w *models.Workflow, idx int, secrets []secrets.UnlockedSecret, wfLogger models.WorkflowLogger) error { - state := w.Data.(vmState) - step := w.Steps[idx] +func (e *Engine) qmpRun(mon *qmp.SocketMonitor, command qmp.Command) ([]byte, error) { + b, err := json.Marshal(command) + if err != nil { + return nil, err + } + return mon.Run(b) +} - var envStr string - appendEnv := func(k, v string) { - // we pass the env vars to shell instead of through ssh - // (we would need to make sshd config have AcceptEnv in seed iso but that kinda sucks - // because that would reload sshd during boot, which is not ideal, so we can just do this.) - // todo(dawn): make images be prepared with sshd configured - envStr += fmt.Sprintf("%s=%s ", k, shellescape(v)) +// sends a command to the qemu guest agent and returns the response +func (e *Engine) qgaRun(ctx context.Context, sock string, command qmp.Command) ([]byte, error) { + b, err := json.Marshal(command) + if err != nil { + return nil, err } - for k, v := range w.Environment { - appendEnv(k, v) + conn, err := (&net.Dialer{}).DialContext(ctx, "unix", sock) + if err != nil { + return nil, err } - for _, s := range secrets { - appendEnv(s.Key, s.Value) + defer conn.Close() + + if dl, ok := ctx.Deadline(); ok { + _ = conn.SetDeadline(dl) } - if s, ok := step.(Step); ok { - for k, v := range s.environment { - appendEnv(k, v) - } + + if _, err := conn.Write(append(b, '\n')); err != nil { + return nil, err } - session, err := state.sshClient.NewSession() + return bufio.NewReader(conn).ReadBytes('\n') +} + +// query the qemu guest status +func (e *Engine) qmpQueryStatus(mon *qmp.SocketMonitor) (string, error) { + raw, err := e.qmpRun(mon, qmp.Command{Execute: "query-status"}) if err != nil { - return fmt.Errorf("ssh new session: %w", err) + return "", fmt.Errorf("qmp query-status failed: %w", err) + } + + var resp struct { + Return struct { + Status string `json:"status"` + } `json:"return"` } - defer session.Close() + if err := json.Unmarshal(raw, &resp); err != nil { + return "", fmt.Errorf("qmp query-status parse: %w", err) + } + return resp.Return.Status, nil +} + +// send a command to qemu to powerdown the guest gracefully +func (e *Engine) qmpSystemPowerdown(mon *qmp.SocketMonitor) error { + _, err := e.qmpRun(mon, qmp.Command{Execute: "system_powerdown"}) + return err +} + +// ping the guest agent to see if it's ready +func (e *Engine) qgaGuestPing(ctx context.Context, sock string) error { + _, err := e.qgaRun(ctx, sock, qmp.Command{Execute: "guest-ping"}) + return err +} - stdoutPipe, err := session.StdoutPipe() +// execute a command on the guest and return its pid +func (e *Engine) qgaGuestExec(ctx context.Context, sock string, shell string, command string, env []string) (int, error) { + cmdArgs := []string{shell, "-c", command} + raw, err := e.qgaRun(ctx, sock, qmp.Command{ + Execute: "guest-exec", + Args: map[string]any{ + "path": shell, + "arg": cmdArgs[1:], + "env": env, + "capture-output": true, + }, + }) if err != nil { - return fmt.Errorf("ssh stdout pipe: %w", err) + return 0, fmt.Errorf("qga guest-exec: %w", err) } - stderrPipe, err := session.StderrPipe() + + var resp struct { + Return struct { + Pid int `json:"pid"` + } `json:"return"` + } + if err := json.Unmarshal(raw, &resp); err != nil { + return 0, fmt.Errorf("qga guest-exec parse: %w", err) + } + return resp.Return.Pid, nil +} + +type guestExecStatus struct { + Exited bool + ExitCode int + // these are since the last time we asked for status + OutData []byte + ErrData []byte +} + +// get the status of a command executed with guest-exec +func (e *Engine) qgaGuestExecStatus(ctx context.Context, sock string, pid int) (guestExecStatus, error) { + var status guestExecStatus + raw, err := e.qgaRun(ctx, sock, qmp.Command{ + Execute: "guest-exec-status", + Args: map[string]any{"pid": pid}, + }) if err != nil { - return fmt.Errorf("ssh stderr pipe: %w", err) + return status, fmt.Errorf("qga guest-exec-status: %w", err) } - tailDone := make(chan error, 1) - go func() { - tailDone <- streamLogs(ctx, stdoutPipe, stderrPipe, idx, wfLogger) - }() + var resp struct { + Return struct { + Exited bool `json:"exited"` + ExitCode int `json:"exitcode"` + OutData string `json:"out-data"` + ErrData string `json:"err-data"` + } `json:"return"` + } + if err := json.Unmarshal(raw, &resp); err != nil { + return status, fmt.Errorf("qga guest-exec-status parse: %w", err) + } - // execute with the image's specified shell - cmd := fmt.Sprintf("%s%s -c %s", envStr, state.shell, shellescape(step.Command())) - if err := session.Start(cmd); err != nil { - return fmt.Errorf("session start: %w", err) + status.Exited = resp.Return.Exited + status.ExitCode = resp.Return.ExitCode + if resp.Return.OutData != "" { + status.OutData, _ = base64.StdEncoding.DecodeString(resp.Return.OutData) + } + if resp.Return.ErrData != "" { + status.ErrData, _ = base64.StdEncoding.DecodeString(resp.Return.ErrData) } - select { - case err := <-tailDone: - if err != nil { - e.l.Warn("log streaming error", "workflow", wid, "step", idx, "error", err) + return status, nil +} + +func (e *Engine) RunStep(ctx context.Context, wid models.WorkflowId, w *models.Workflow, idx int, secrets []secrets.UnlockedSecret, wfLogger models.WorkflowLogger) error { + state := w.Data.(vmState) + step := w.Steps[idx] + + env := make([]string, 0, len(w.Environment)+len(secrets)) + for k, v := range w.Environment { + env = append(env, k+"="+v) + } + for _, s := range secrets { + env = append(env, s.Key+"="+s.Value) + } + if s, ok := step.(Step); ok { + for k, v := range s.environment { + env = append(env, k+"="+v) } - case <-ctx.Done(): - _ = session.Signal(ssh.SIGKILL) - <-tailDone - return engine.ErrTimedOut } - if err := session.Wait(); err != nil { - return engine.ErrWorkflowFailed + pid, err := e.qgaGuestExec(ctx, state.qgaPath, state.img.shell, step.Command(), env) + if err != nil { + return err + } + + for { + status, err := e.qgaGuestExecStatus(ctx, state.qgaPath, pid) + if err != nil { + return err + } + + if len(status.OutData) > 0 { + _, _ = wfLogger.DataWriter(idx, "stdout").Write(status.OutData) + } + if len(status.ErrData) > 0 { + _, _ = wfLogger.DataWriter(idx, "stderr").Write(status.ErrData) + } + + if status.Exited { + if status.ExitCode != 0 { + return engine.ErrWorkflowFailed + } + break + } + + select { + case <-ctx.Done(): + return engine.ErrTimedOut + case <-time.After(100 * time.Millisecond): + } } return nil } diff --git a/spindle/engines/qemu/logs.go b/spindle/engines/qemu/logs.go deleted file mode 100644 index 100fdf36..00000000 --- a/spindle/engines/qemu/logs.go +++ /dev/null @@ -1,35 +0,0 @@ -package qemu - -import ( - "context" - "io" - "strings" - - "golang.org/x/sync/errgroup" - "tangled.org/core/spindle/models" -) - -// shellescape safely escapes a string for bash -c -func shellescape(str string) string { - return "'" + strings.ReplaceAll(str, "'", `'"'"'`) + "'" -} - -// streamLogs copies stdout and stderr from the SSH session to the workflow logger -func streamLogs(ctx context.Context, stdoutPipe io.Reader, stderrPipe io.Reader, idx int, wfLogger models.WorkflowLogger) error { - if wfLogger == nil { - return nil - } - - g, _ := errgroup.WithContext(ctx) - - g.Go(func() error { - _, err := io.Copy(wfLogger.DataWriter(idx, "stdout"), stdoutPipe) - return err - }) - g.Go(func() error { - _, err := io.Copy(wfLogger.DataWriter(idx, "stderr"), stderrPipe) - return err - }) - - return g.Wait() -} diff --git a/spindle/engines/qemu/models.go b/spindle/engines/qemu/models.go index 02c234f1..2052c9d2 100644 --- a/spindle/engines/qemu/models.go +++ b/spindle/engines/qemu/models.go @@ -4,21 +4,24 @@ import ( "os" "github.com/digitalocean/go-qemu/qmp" - "golang.org/x/crypto/ssh" ) type vmState struct { - process *os.Process - qmpMon *qmp.SocketMonitor - sshClient *ssh.Client - sshPort int - tempDir string + process *os.Process + qmpMon *qmp.SocketMonitor + qgaPath string + tempDir string - kernel string - initrd string - disk string - cmdline string - shell string + img ResolvedImage +} + +type ResolvedImage struct { + kernel string + initrd string + disk string + cmdline string // kernel command line + shell string // shell to use for workflow steps + userData string // extra cloud-init user-data } type manifestWorkflow struct { diff --git a/spindle/engines/qemu/setup.go b/spindle/engines/qemu/setup.go index af3c8fcd..fd2133a9 100644 --- a/spindle/engines/qemu/setup.go +++ b/spindle/engines/qemu/setup.go @@ -10,7 +10,6 @@ import ( "golang.org/x/sync/errgroup" "tangled.org/core/log" - "tangled.org/core/spindle/config" "tangled.org/core/spindle/engines/qemu/bakers" ) @@ -73,15 +72,9 @@ func downloadFile(ctx context.Context, url string, dest string) error { return os.Rename(tmp, dest) } -func SetupDefaultImages(ctx context.Context) error { - cfg, err := config.Load(ctx) - if err != nil { - return fmt.Errorf("loading config: %w", err) - } - - imageDir := cfg.QemuPipelines.ImageDir +func SetupDefaultImages(ctx context.Context, imageDir string) error { if imageDir == "" { - return fmt.Errorf("SPINDLE_QEMU_PIPELINES_IMAGE_DIR must be set") + return fmt.Errorf("imageDir must be set") } l := log.FromContext(ctx) @@ -101,11 +94,7 @@ func SetupDefaultImages(ctx context.Context) error { return err } - diskPath := filepath.Join(dir, "disk.qcow2") - if _, err := os.Stat(diskPath); err == nil { - l.Info("image already exists, skipping", "name", name) - return nil - } + diskPath := bakers.DiskPath(dir) l.Info("downloading image...", "name", name) if err := downloadFile(ctx, img.disk, diskPath); err != nil { @@ -113,18 +102,18 @@ func SetupDefaultImages(ctx context.Context) error { } if img.kernel != "" { - if err := downloadFile(ctx, img.kernel, filepath.Join(dir, "kernel")); err != nil { + if err := downloadFile(ctx, img.kernel, bakers.KernelPath(dir)); err != nil { return fmt.Errorf("downloading %s kernel: %w", name, err) } } if img.initrd != "" { - if err := downloadFile(ctx, img.initrd, filepath.Join(dir, "initrd")); err != nil { + if err := downloadFile(ctx, img.initrd, bakers.InitrdPath(dir)); err != nil { return fmt.Errorf("downloading %s initrd: %w", name, err) } } - if err := bakers.PrepareImage(ctx, cfg, name); err != nil { + if err := bakers.PrepareImage(ctx, imageDir, name); err != nil { return fmt.Errorf("preparing image %s: %w", name, err) } -- 2.51.2