diff --git a/cmd/spindle/main.go b/cmd/spindle/main.go index 083f29f6..2c58dc62 100644 --- a/cmd/spindle/main.go +++ b/cmd/spindle/main.go @@ -2,23 +2,76 @@ package main import ( "context" + "fmt" "log/slog" "os" + "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" ) func main() { + cmd := &cli.Command{ + Name: "spindle", + Usage: "spindle continuous integration runner", + Commands: []*cli.Command{ + Command(), + SetupQemuCommand(), + BakeImageCommand(), + }, + DefaultCommand: "run", + } + logger := tlog.New("spindle") slog.SetDefault(logger) ctx := context.Background() ctx = tlog.IntoContext(ctx, logger) - err := spindle.Run(ctx) - if err != nil { - logger.Error("error running spindle", "error", err) + if err := cmd.Run(ctx, os.Args); err != nil { + logger.Error(err.Error()) os.Exit(-1) } } + +func Command() *cli.Command { + return &cli.Command{ + Name: "run", + Usage: "run the spindle server", + Action: func(ctx context.Context, cmd *cli.Command) error { + return spindle.Run(ctx) + }, + } +} + +func SetupQemuCommand() *cli.Command { + return &cli.Command{ + Name: "setup-qemu", + Usage: "downloads and prepares default qemu engine images", + Action: func(ctx context.Context, cmd *cli.Command) error { + return qemu.SetupDefaultImages(ctx) + }, + } +} + +func BakeImageCommand() *cli.Command { + return &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) + } + name := cmd.Args().First() + if name == "" { + return fmt.Errorf("name argument (e.g. 'alpine') is required") + } + return bakers.PrepareImage(ctx, cfg, name) + }, + } +} diff --git a/flake.lock b/flake.lock index ea4a2361..13ff027f 100644 --- a/flake.lock +++ b/flake.lock @@ -117,14 +117,14 @@ "ibm-plex-mono-src": { "flake": false, "locked": { - "lastModified": 1731402384, + "lastModified": 1731402378, "narHash": "sha256-OwUmrPfEehLDz0fl2ChYLK8FQM2p0G1+EMrGsYEq+6g=", "type": "tarball", - "url": "https://github.com/IBM/plex/releases/download/@ibm%2Fplex-mono@1.1.0/ibm-plex-mono.zip" + "url": "https://github.com/IBM/plex/releases/download/@ibm/plex-mono@1.1.0/ibm-plex-mono.zip" }, "original": { "type": "tarball", - "url": "https://github.com/IBM/plex/releases/download/@ibm%2Fplex-mono@1.1.0/ibm-plex-mono.zip" + "url": "https://github.com/IBM/plex/releases/download/@ibm/plex-mono@1.1.0/ibm-plex-mono.zip" } }, "indigo": { diff --git a/flake.nix b/flake.nix index c227919e..8e3ad4ec 100644 --- a/flake.nix +++ b/flake.nix @@ -214,6 +214,9 @@ pkgs.redis pkgs.worker-build pkgs.cargo-generate + pkgs.qemu + pkgs.cloud-utils + pkgs.cdrkit (fenix.packages.${system}.combine [ fenix.packages.${system}.stable.cargo fenix.packages.${system}.stable.rustc diff --git a/go.mod b/go.mod index c650ff8a..b8080ccd 100644 --- a/go.mod +++ b/go.mod @@ -30,6 +30,7 @@ require ( github.com/cyphar/filepath-securejoin v0.4.1 github.com/dgraph-io/ristretto v0.2.0 github.com/did-method-plc/go-didplc v0.2.2 + github.com/digitalocean/go-qemu v0.0.0-20250212194115-ee9b0668d242 github.com/djherbis/buffer v1.2.0 github.com/djherbis/nio/v3 v3.0.1 github.com/docker/docker v28.2.2+incompatible @@ -142,6 +143,7 @@ require ( github.com/creack/pty v1.1.21 // indirect github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect + github.com/digitalocean/go-libvirt v0.0.0-20220804181439-8648fbde413e // indirect github.com/distribution/reference v0.6.0 // indirect github.com/dlclark/regexp2 v1.11.5 // indirect github.com/docker/go-connections v0.5.0 // indirect diff --git a/go.sum b/go.sum index cda5f648..706804ff 100644 --- a/go.sum +++ b/go.sum @@ -221,6 +221,10 @@ github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/r github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc= github.com/did-method-plc/go-didplc v0.2.2 h1:53HFhTT8NCAeFmZ6fdIZCf3PGDvj7A3cDjzOOEqn5XM= github.com/did-method-plc/go-didplc v0.2.2/go.mod h1:bKdJ21irnwNHgVLWWL32zUWqZueXYbJRUcxplZghByo= +github.com/digitalocean/go-libvirt v0.0.0-20220804181439-8648fbde413e h1:SCnqm8SjSa0QqRxXbo5YY//S+OryeJioe17nK+iDZpg= +github.com/digitalocean/go-libvirt v0.0.0-20220804181439-8648fbde413e/go.mod h1:o129ljs6alsIQTc8d6eweihqpmmrbxZ2g1jhgjhPykI= +github.com/digitalocean/go-qemu v0.0.0-20250212194115-ee9b0668d242 h1:rh6rt8pF5U4iyQ86h6lRDenJoX4ht2wFnZXB9ogIrIM= +github.com/digitalocean/go-qemu v0.0.0-20250212194115-ee9b0668d242/go.mod h1:LGHUtlhsY4vRGM6AHejEQKVI5e3eHbSylMHwTSpQtVw= github.com/distribution/reference v0.6.0 h1:0IXCQ5g4/QMHHkarYzh5l+u8T3t73zM5QvfrDyIgxBk= github.com/distribution/reference v0.6.0/go.mod h1:BbU0aIcezP1/5jX/8MP0YiH4SdvB5Y4f/wlDRiLyi3E= github.com/djherbis/buffer v1.1.0/go.mod h1:VwN8VdFkMY0DCALdY8o00d3IZ6Amz/UNVMWcSaJT44o= diff --git a/nix/gomod2nix.toml b/nix/gomod2nix.toml index d178c56d..8cdf6dd9 100644 --- a/nix/gomod2nix.toml +++ b/nix/gomod2nix.toml @@ -286,6 +286,12 @@ schema = 3 [mod."github.com/did-method-plc/go-didplc"] version = "v0.2.2" hash = "sha256-TF5vdW1U2q5F23ELmbqhdvpNgDQjBs9i/CWJlzqWNKs=" + [mod."github.com/digitalocean/go-libvirt"] + version = "v0.0.0-20220804181439-8648fbde413e" + hash = "sha256-xgRZCefeUNM76M9ht5FCgCf4dnwAdeP/r+FZlJ2gmeY=" + [mod."github.com/digitalocean/go-qemu"] + version = "v0.0.0-20250212194115-ee9b0668d242" + hash = "sha256-AzY84aq9CqGqYwmGkKoi4PX08E/hRQC46mrZwKSMwXE=" [mod."github.com/distribution/reference"] version = "v0.6.0" hash = "sha256-gr4tL+qz4jKyAtl8LINcxMSanztdt+pybj1T+2ulQv4=" diff --git a/nix/modules/spindle.nix b/nix/modules/spindle.nix index 502a681b..04afcb73 100644 --- a/nix/modules/spindle.nix +++ b/nix/modules/spindle.nix @@ -1,189 +1,248 @@ { config, lib, + pkgs, ... -}: let +}: +let cfg = config.services.tangled.spindle; in - with lib; { - options = { - services.tangled.spindle = { - enable = mkOption { +with lib; +{ + options = { + services.tangled.spindle = { + enable = mkOption { + type = types.bool; + default = false; + description = "Enable a tangled spindle"; + }; + package = mkOption { + type = types.package; + description = "Package to use for the spindle"; + }; + + server = { + listenAddr = mkOption { + type = types.str; + default = "0.0.0.0:6555"; + description = "Address to listen on"; + }; + + dbPath = mkOption { + type = types.path; + default = "/var/lib/spindle/spindle.db"; + description = "Path to the database file"; + }; + + hostname = mkOption { + type = types.str; + example = "my.spindle.com"; + description = "Hostname for the server (required)"; + }; + + plcUrl = mkOption { + type = types.str; + default = "https://plc.directory"; + description = "atproto PLC directory"; + }; + + jetstreamEndpoint = mkOption { + type = types.str; + default = "wss://jetstream1.us-west.bsky.network/subscribe"; + description = "Jetstream endpoint to subscribe to"; + }; + + dev = mkOption { type = types.bool; default = false; - description = "Enable a tangled spindle"; + description = "Enable development mode (disables signature verification)"; }; - package = mkOption { - type = types.package; - description = "Package to use for the spindle"; + + owner = mkOption { + type = types.str; + example = "did:plc:qfpnj4og54vl56wngdriaxug"; + description = "DID of owner (required)"; }; - server = { - listenAddr = mkOption { - type = types.str; - default = "0.0.0.0:6555"; - description = "Address to listen on"; - }; + maxJobCount = mkOption { + type = types.int; + default = 2; + example = 5; + description = "Maximum number of concurrent jobs to run"; + }; - dbPath = mkOption { - type = types.path; - default = "/var/lib/spindle/spindle.db"; - description = "Path to the database file"; - }; + queueSize = mkOption { + type = types.int; + default = 100; + example = 100; + description = "Maximum number of jobs queue up"; + }; - hostname = mkOption { - type = types.str; - example = "my.spindle.com"; - description = "Hostname for the server (required)"; - }; + maxConcurrentWorkflows = mkOption { + type = types.int; + default = 8; + description = "Maximum number of workflow containers running simultaneously (controls total memory usage)"; + }; - plcUrl = mkOption { + secrets = { + provider = mkOption { type = types.str; - default = "https://plc.directory"; - description = "atproto PLC directory"; + default = "sqlite"; + description = "Backend to use for secret management, valid options are 'sqlite', and 'openbao'."; }; - jetstreamEndpoint = mkOption { - type = types.str; - default = "wss://jetstream1.us-west.bsky.network/subscribe"; - description = "Jetstream endpoint to subscribe to"; + openbao = { + proxyAddr = mkOption { + type = types.str; + default = "http://127.0.0.1:8200"; + description = "Address of the OpenBAO proxy server"; + }; + mount = mkOption { + type = types.str; + default = "spindle"; + description = "Mount path in OpenBAO to read secrets from"; + }; }; + }; - dev = mkOption { + tap = { + embed = mkOption { type = types.bool; - default = false; - description = "Enable development mode (disables signature verification)"; + default = true; + description = "Run an embedded tap inside the spindle process"; }; - owner = mkOption { + url = mkOption { type = types.str; - example = "did:plc:qfpnj4og54vl56wngdriaxug"; - description = "DID of owner (required)"; + default = "http://[::1]:2480"; + description = "URL the spindle's tap client dials"; }; - maxJobCount = mkOption { - type = types.int; - default = 2; - example = 5; - description = "Maximum number of concurrent jobs to run"; - }; - - queueSize = mkOption { - type = types.int; - default = 100; - example = 100; - description = "Maximum number of jobs queue up"; + bind = mkOption { + type = types.str; + default = "[::1]:2480"; + description = "Loopback address the embedded tap server listens on"; }; - maxConcurrentWorkflows = mkOption { - type = types.int; - default = 8; - description = "Maximum number of workflow containers running simultaneously (controls total memory usage)"; + dbPath = mkOption { + type = types.path; + default = "/var/lib/spindle/tap.db"; + description = "Path to the embedded tap sqlite database"; }; - secrets = { - provider = mkOption { - type = types.str; - default = "sqlite"; - description = "Backend to use for secret management, valid options are 'sqlite', and 'openbao'."; - }; - - openbao = { - proxyAddr = mkOption { - type = types.str; - default = "http://127.0.0.1:8200"; - description = "Address of the OpenBAO proxy server"; - }; - mount = mkOption { - type = types.str; - default = "spindle"; - description = "Mount path in OpenBAO to read secrets from"; - }; - }; + relayUrl = mkOption { + type = types.str; + default = "https://bsky.network"; + description = "Relay used by the embedded tap firehose"; }; + }; + }; - tap = { - embed = mkOption { - type = types.bool; - default = true; - description = "Run an embedded tap inside the spindle process"; - }; - - url = mkOption { - type = types.str; - default = "http://[::1]:2480"; - description = "URL the spindle's tap client dials"; - }; + pipelines = { + # todo(dawn): move these into their own pipeline configs eventually + nixery = mkOption { + type = types.str; + default = "nixery.tangled.sh"; # note: this is *not* on tangled.org yet + description = "Nixery instance to use"; + }; - bind = mkOption { - type = types.str; - default = "[::1]:2480"; - description = "Loopback address the embedded tap server listens on"; - }; + workflowTimeout = mkOption { + type = types.str; + default = "5m"; + description = "Timeout for each step of a pipeline"; + }; - dbPath = mkOption { - type = types.path; - default = "/var/lib/spindle/tap.db"; - description = "Path to the embedded tap sqlite database"; - }; + maxJobMemoryMb = mkOption { + type = types.int; + default = 6144; + description = "Memory limit per workflow container in MiB (default 6 GiB)"; + }; - relayUrl = mkOption { - type = types.str; - default = "https://bsky.network"; - description = "Relay used by the embedded tap firehose"; - }; - }; + logBucket = mkOption { + type = types.str; + default = "tangled-logs"; + description = "S3 bucket for workflow logs"; }; - pipelines = { - nixery = mkOption { + qemu = { + imageDir = mkOption { type = types.str; - default = "nixery.tangled.sh"; # note: this is *not* on tangled.org yet - description = "Nixery instance to use"; + default = "/var/lib/spindle/images"; + description = "Directory containing QEMU images"; }; - - workflowTimeout = mkOption { + overlayDir = mkOption { type = types.str; - default = "5m"; - description = "Timeout for each step of a pipeline"; + default = "/tmp"; + description = "Directory to store QEMU temporary overlay files"; }; - - maxJobMemoryMb = mkOption { - type = types.int; - default = 6144; - description = "Memory limit per workflow container in MiB (default 6 GiB)"; + defaultImage = mkOption { + type = types.str; + default = "ubuntu-24.04"; + description = "Default image to use if none is specified in workflow"; }; - - logBucket = mkOption { + enableKVM = mkOption { + type = types.bool; + default = true; + description = "Enable KVM hardware acceleration"; + }; + memory = mkOption { type = types.str; - default = "tangled-logs"; - description = "S3 bucket for workflow logs"; + default = "2048"; + description = "Amount of RAM to assign per-VM (supports formats like 512 or 2G)"; + }; + smp = mkOption { + type = types.int; + default = 2; + description = "Number of CPU cores to assign per-VM"; }; }; + }; - environmentFile = mkOption { - type = with types; nullOr path; - default = null; - example = "/etc/spindle.env"; - description = '' - Additional environment file as defined in {manpage}`systemd.exec(5)`. + environmentFile = mkOption { + type = with types; nullOr path; + default = null; + example = "/etc/spindle.env"; + description = '' + Additional environment file as defined in {manpage}`systemd.exec(5)`. - Sensitive secrets such as {env}`AWS_SECRET_ACCESS_KEY`, - {env}`AWS_ACCESS_KEY_ID`, {env}`AWS_REGION` - may be passed to the service - without making them world readable in the nix store. - ''; - }; + Sensitive secrets such as {env}`AWS_SECRET_ACCESS_KEY`, + {env}`AWS_ACCESS_KEY_ID`, {env}`AWS_REGION` + may be passed to the service + without making them world readable in the nix store. + ''; }; }; + }; + + config = + let + deps = [ + pkgs.cdrkit + pkgs.qemu + ]; + in + mkIf cfg.enable { + environment.systemPackages = [ + (pkgs.writeShellScriptBin "spindle" '' + export PATH="${lib.makeBinPath deps}:$PATH" + ${lib.optionalString (cfg.environmentFile != null) "set -a; source ${cfg.environmentFile}; set +a"} + ${lib.concatMapStringsSep "\n" ( + e: "export ${e}" + ) config.systemd.services.spindle.serviceConfig.Environment} + exec ${cfg.package}/bin/spindle "$@" + '') + ]; - config = mkIf cfg.enable { virtualisation.docker.enable = true; systemd.services.spindle = { description = "spindle service"; - after = ["network.target" "docker.service"]; - wantedBy = ["multi-user.target"]; + after = [ + "network.target" + "docker.service" + ]; + wantedBy = [ "multi-user.target" ]; + path = deps; serviceConfig = { LogsDirectory = "spindle"; StateDirectory = "spindle"; @@ -211,6 +270,13 @@ in "SPINDLE_NIXERY_PIPELINES_NIXERY=${cfg.pipelines.nixery}" "SPINDLE_NIXERY_PIPELINES_WORKFLOW_TIMEOUT=${cfg.pipelines.workflowTimeout}" "SPINDLE_NIXERY_PIPELINES_MAX_JOB_MEMORY_MB=${toString cfg.pipelines.maxJobMemoryMb}" + "SPINDLE_QEMU_PIPELINES_IMAGE_DIR=${cfg.pipelines.qemu.imageDir}" + "SPINDLE_QEMU_PIPELINES_OVERLAY_DIR=${cfg.pipelines.qemu.overlayDir}" + "SPINDLE_QEMU_PIPELINES_DEFAULT_IMAGE=${cfg.pipelines.qemu.defaultImage}" + "SPINDLE_QEMU_PIPELINES_ENABLE_KVM=${lib.boolToString cfg.pipelines.qemu.enableKVM}" + "SPINDLE_QEMU_PIPELINES_MEMORY=${cfg.pipelines.qemu.memory}" + "SPINDLE_QEMU_PIPELINES_SMP=${toString cfg.pipelines.qemu.smp}" + "SPINDLE_QEMU_PIPELINES_WORKFLOW_TIMEOUT=${cfg.pipelines.workflowTimeout}" "SPINDLE_S3_LOG_BUCKET=${cfg.pipelines.logBucket}" ]; ExecStart = "${cfg.package}/bin/spindle"; @@ -218,4 +284,4 @@ in }; }; }; - } +} diff --git a/nix/vm.nix b/nix/vm.nix index f12debec..99c47681 100644 --- a/nix/vm.nix +++ b/nix/vm.nix @@ -4,6 +4,8 @@ hostSystem, self, }: let + lib = nixpkgs.lib; + envVar = name: let var = builtins.getEnv name; in @@ -19,8 +21,18 @@ plcUrl = envVarOr "TANGLED_VM_PLC_URL" "https://plc.directory"; jetstream = envVarOr "TANGLED_VM_JETSTREAM_ENDPOINT" "wss://jetstream1.us-west.bsky.network/subscribe"; + + checkFile = value: path: if builtins.pathExists path then lib.hasPrefix value (builtins.readFile path) else false; + _nestedVirt = + (checkFile "1" /sys/module/kvm_amd/parameters/nested) + || (checkFile "Y" /sys/module/kvm_intel/parameters/nested); + nestedVirtWarning = '' + KVM nested virtualisation is not enabled on this host. + You should enable it if you can for better performance when testing the QEMU spindle engine! + ''; + nestedVirt = lib.warnIf (!_nestedVirt) nestedVirtWarning _nestedVirt; in - nixpkgs.lib.nixosSystem { + lib.nixosSystem { inherit system; modules = [ self.nixosModules.knot @@ -36,9 +48,11 @@ in host.pkgs = import nixpkgs {system = hostSystem;}; graphics = false; - memorySize = 2048; - diskSize = 10 * 1024; + memorySize = 3072; + diskSize = 20 * 1024; cores = 2; + qemu.options = lib.optionals nestedVirt ["-enable-kvm" "-cpu host"]; + forwardPorts = [ # ssh { @@ -139,6 +153,10 @@ in pipelines = { logBucket = envVarOr "SPINDLE_S3_LOG_BUCKET" ""; + qemu = { + memory = "1024"; + smp = 2; + }; }; }; services.postgresql = { diff --git a/spindle/config/config.go b/spindle/config/config.go index 332355ba..f5669357 100644 --- a/spindle/config/config.go +++ b/spindle/config/config.go @@ -58,9 +58,20 @@ type S3 struct { LogBucket string `env:"LOG_BUCKET"` } +type QemuPipelines struct { + ImageDir string `env:"IMAGE_DIR, required"` + 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"` + Memory string `env:"MEMORY, default=2048"` + SMP int `env:"SMP, default=2"` + WorkflowTimeout string `env:"WORKFLOW_TIMEOUT, default=5m"` +} + type Config struct { Server Server `env:",prefix=SPINDLE_SERVER_"` NixeryPipelines NixeryPipelines `env:",prefix=SPINDLE_NIXERY_PIPELINES_"` + QemuPipelines QemuPipelines `env:",prefix=SPINDLE_QEMU_PIPELINES_"` S3 S3 `env:",prefix=SPINDLE_S3_"` } diff --git a/spindle/engines/qemu/bakers/alpine.go b/spindle/engines/qemu/bakers/alpine.go new file mode 100644 index 00000000..7e05a635 --- /dev/null +++ b/spindle/engines/qemu/bakers/alpine.go @@ -0,0 +1,102 @@ +package bakers + +import ( + "context" + "encoding/json" + "fmt" + "io" + "os" + "os/exec" + "path/filepath" + + "tangled.org/core/log" +) + +type AlpineBaker struct { + nbd *NBDManager +} + +func copyFile(src, dst string) error { + in, err := os.Open(src) + if err != nil { + return err + } + defer in.Close() + + out, err := os.Create(dst) + if err != nil { + return err + } + defer out.Close() + + if _, err := io.Copy(out, in); err != nil { + return err + } + return out.Sync() +} + +func (p *AlpineBaker) Prepare(ctx context.Context, dir string) error { + l := log.FromContext(ctx) + diskPath := filepath.Join(dir, "disk.qcow2") + + nbd, err := p.nbd.Connect(ctx, l, diskPath) + if err != nil { + return err + } + defer nbd.Close() + + // alpine images have no partition table, so we mount the base device directly + mntDir, err := nbd.Mount(ctx, nbd.Device()) + if err != nil { + return err + } + 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) + } + 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 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 { + return fmt.Errorf("installing packages: %s: %w", string(out), err) + } + + l.Info("extracting kernel and initrd...") + bootDir := filepath.Join(mntDir, "boot") + kernelSource := filepath.Join(bootDir, "vmlinuz-virt") + if _, err := os.Stat(kernelSource); os.IsNotExist(err) { + return fmt.Errorf("kernel 'vmlinuz-virt' not found in %s", bootDir) + } + initrdSource := filepath.Join(bootDir, "initramfs-virt") + if _, err := os.Stat(initrdSource); os.IsNotExist(err) { + return fmt.Errorf("initrd 'initramfs-virt' not found in %s", bootDir) + } + + if err := copyFile(kernelSource, filepath.Join(dir, "kernel")); err != nil { + return fmt.Errorf("extracting kernel: %w", err) + } + if err := copyFile(initrdSource, filepath.Join(dir, "initrd")); err != nil { + return fmt.Errorf("extracting initrd: %w", err) + } + + configPath := filepath.Join(dir, "config.json") + metadata := ImageMetadata{ + Cmdline: "root=/dev/vda rootfstype=ext4 rw console=ttyS0 modules=sd-mod,usb-storage,ext4,virtio_pci,virtio_blk", + Shell: "/bin/ash", + } + 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) + } + + l.Info("alpine preparation complete") + return nil +} diff --git a/spindle/engines/qemu/bakers/baker.go b/spindle/engines/qemu/bakers/baker.go new file mode 100644 index 00000000..3469159b --- /dev/null +++ b/spindle/engines/qemu/bakers/baker.go @@ -0,0 +1,59 @@ +package bakers + +import ( + "context" + "fmt" + "os" + "os/exec" + "path/filepath" + "strings" + + "tangled.org/core/log" + "tangled.org/core/spindle/config" +) + +type ImageMetadata struct { + Cmdline string `json:"cmdline"` + Shell string `json:"shell"` +} + +type ImageBaker interface { + Prepare(ctx context.Context, imageDir string) error +} + +var ( + nbd = &NBDManager{} + preparers = map[string]ImageBaker{ + "alpine": &AlpineBaker{nbd: nbd}, + "ubuntu": &UbuntuBaker{}, + } +) + +func PrepareImage(ctx context.Context, cfg *config.Config, name string) error { + name = strings.ToLower(name) + + imageDir := filepath.Join(cfg.QemuPipelines.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) + } + + l := log.FromContext(ctx) + l.Info("preparing image", "name", name) + + diskPath := filepath.Join(imageDir, "disk.qcow2") + // 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) + if out, err := exec.CommandContext(ctx, "qemu-img", "resize", diskPath, "8G").CombinedOutput(); err != nil { + return fmt.Errorf("qemu-img resize: %s: %w", string(out), err) + } + + // dispatch based on the prefix (e.g. 'alpine' from 'alpine-3.23') + parts := strings.Split(name, "-") + if preparer, ok := preparers[parts[0]]; ok { + l.Info("running preparer", "type", parts[0]) + return preparer.Prepare(ctx, imageDir) + } + + return nil +} diff --git a/spindle/engines/qemu/bakers/nbd.go b/spindle/engines/qemu/bakers/nbd.go new file mode 100644 index 00000000..c39524d0 --- /dev/null +++ b/spindle/engines/qemu/bakers/nbd.go @@ -0,0 +1,91 @@ +package bakers + +import ( + "context" + "fmt" + "log/slog" + "os" + "os/exec" + "sync" + "time" +) + +type NBDManager struct { + mu sync.Mutex +} + +type NBDHandle struct { + mgr *NBDManager + device string + l *slog.Logger +} + +func (n *NBDManager) Connect(ctx context.Context, l *slog.Logger, diskPath string) (*NBDHandle, error) { + n.mu.Lock() + + if _, err := os.Stat("/dev/nbd0"); 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() + return nil, fmt.Errorf("modprobe nbd: %s: %w", string(out), err) + } + } + + l.Info("connecting disk via nbd...", "disk", diskPath) + if out, err := exec.CommandContext(ctx, "qemu-nbd", "--connect=/dev/nbd0", diskPath).CombinedOutput(); err != nil { + n.mu.Unlock() + return nil, fmt.Errorf("qemu-nbd connect: %s: %w", string(out), err) + } + + waitCtx, cancel := context.WithTimeout(ctx, 3*time.Second) + defer cancel() + for { + if _, err := os.Stat("/dev/nbd0"); err == nil { + break + } + select { + case <-waitCtx.Done(): + n.mu.Unlock() + return nil, fmt.Errorf("timeout waiting for /dev/nbd0 to appear") + case <-time.After(50 * time.Millisecond): + } + } + + return &NBDHandle{mgr: n, device: "/dev/nbd0", l: l}, nil +} + +// mounts the device node specified +func (h *NBDHandle) Mount(ctx context.Context, device string) (string, error) { + mntDir, err := os.MkdirTemp("", "spindle-mnt-*") + if err != nil { + return "", err + } + + h.l.Info("mounting nbd device...", "device", device) + if out, err := exec.CommandContext(ctx, "mount", device, mntDir).CombinedOutput(); err != nil { + os.RemoveAll(mntDir) + return "", fmt.Errorf("mount %s: %s: %w", device, string(out), err) + } + + return mntDir, nil +} + +func (h *NBDHandle) Unmount(mntDir string) { + if out, err := exec.Command("umount", mntDir).CombinedOutput(); err != nil { + h.l.Error("umount failed", "error", err, "output", string(out)) + } + os.RemoveAll(mntDir) +} + +// returns the base node (/dev/nbd0) +func (h *NBDHandle) Device() string { + return h.device +} + +// disconnects the nbd and releases the lock +func (h *NBDHandle) Close() { + defer h.mgr.mu.Unlock() + if out, err := exec.Command("qemu-nbd", "--disconnect", h.device).CombinedOutput(); err != nil { + h.l.Error("qemu-nbd disconnect failed", "error", err, "output", string(out)) + } +} diff --git a/spindle/engines/qemu/bakers/ubuntu.go b/spindle/engines/qemu/bakers/ubuntu.go new file mode 100644 index 00000000..af14c671 --- /dev/null +++ b/spindle/engines/qemu/bakers/ubuntu.go @@ -0,0 +1,36 @@ +package bakers + +import ( + "context" + "encoding/json" + "fmt" + "os" + "path/filepath" + + "tangled.org/core/log" +) + +type UbuntuBaker struct{} + +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) + } + + configPath := filepath.Join(dir, "config.json") + 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) + } + + l.Info("ubuntu configuration created") + return nil +} diff --git a/spindle/engines/qemu/cloudinit.go b/spindle/engines/qemu/cloudinit.go new file mode 100644 index 00000000..846c3b6c --- /dev/null +++ b/spindle/engines/qemu/cloudinit.go @@ -0,0 +1,42 @@ +package qemu + +import ( + "fmt" + "os" + "os/exec" + "path/filepath" +) + +func generateSeedISO(dir string, enginePubKey string) error { + metaData := "instance-id: spindle-vm\nlocal-hostname: spindle\n" + userData := fmt.Sprintf(`#cloud-config +users: + - name: build + sudo: ALL=(ALL) NOPASSWD:ALL + shell: /bin/sh + ssh_authorized_keys: + - %s +`, enginePubKey) + + metaDataPath := filepath.Join(dir, "meta-data") + if err := os.WriteFile(metaDataPath, []byte(metaData), 0o644); err != nil { + return fmt.Errorf("writing meta-data: %w", err) + } + + userDataPath := filepath.Join(dir, "user-data") + if err := os.WriteFile(userDataPath, []byte(userData), 0o644); err != nil { + return fmt.Errorf("writing user-data: %w", err) + } + + // 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.Dir = dir + cmd.Stdout = os.Stdout + cmd.Stderr = os.Stderr + + if err := cmd.Run(); err != nil { + return fmt.Errorf("running genisoimage: %w", err) + } + + return nil +} diff --git a/spindle/engines/qemu/engine.go b/spindle/engines/qemu/engine.go new file mode 100644 index 00000000..23aeb474 --- /dev/null +++ b/spindle/engines/qemu/engine.go @@ -0,0 +1,493 @@ +package qemu + +import ( + "context" + "crypto/ed25519" + "crypto/rand" + "encoding/json" + "fmt" + "log/slog" + "net" + "os" + "os/exec" + "path/filepath" + "sync" + "time" + + "github.com/digitalocean/go-qemu/qmp" + "golang.org/x/crypto/ssh" + "gopkg.in/yaml.v3" + + "tangled.org/core/api/tangled" + "tangled.org/core/log" + "tangled.org/core/spindle/config" + "tangled.org/core/spindle/engine" + "tangled.org/core/spindle/engines/qemu/bakers" + "tangled.org/core/spindle/models" + "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 { + l *slog.Logger + cfg *config.Config + cleanupMu sync.Mutex + cleanup map[string][]cleanupFunc +} + +func New(ctx context.Context, cfg *config.Config) (*Engine, error) { + return &Engine{ + l: log.FromContext(ctx).With("component", "engine.qemu"), + cfg: cfg, + cleanup: make(map[string][]cleanupFunc), + }, nil +} + +type Step struct { + name string + kind models.StepKind + command string + environment map[string]string +} + +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 + + if err := yaml.Unmarshal([]byte(twf.Raw), &dwf); err != nil { + return nil, err + } + + for _, dstep := range dwf.Steps { + swf.Steps = append(swf.Steps, Step{ + name: dstep.Name, + kind: models.StepKindUser, + command: dstep.Command, + environment: dstep.Environment, + }) + } + 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...) + + 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, + } + + return swf, nil +} + +// discover and resolve kernel, initrd, and disk from an image subfolder +func (e *Engine) resolveImage(name string) (ResolvedImage, error) { + var img ResolvedImage + if name == "" { + name = e.cfg.QemuPipelines.DefaultImage + } + if name == "" { + return img, fmt.Errorf("no image specified in workflow and SPINDLE_QEMU_PIPELINES_DEFAULT_IMAGE is not set") + } + + 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") + + if _, err := os.Stat(diskPath); err == nil { + img.disk = diskPath + } + if _, err := os.Stat(kernelPath); err == nil { + img.kernel = kernelPath + } + if _, err := os.Stat(initrdPath); err == nil { + img.initrd = initrdPath + } + if b, err := os.ReadFile(configPath); err == nil { + var meta bakers.ImageMetadata + if err := json.Unmarshal(b, &meta); err == nil { + if meta.Cmdline != "" { + img.cmdline = meta.Cmdline + } + if meta.Shell != "" { + img.shell = meta.Shell + } + } + } + + if img.disk == "" { + return img, fmt.Errorf("missing 'disk.qcow2' in %s", 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) + } + if img.shell == "" { + return img, fmt.Errorf("shell is not configured for %s", name) + } + return img, nil +} + +func (e *Engine) SetupWorkflow(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, wfLogger models.WorkflowLogger) error { + l := e.l.With("workflow", wid) + l.Info("setting up qemu workflow") + + setupStep := Step{name: "qemu vm setup", kind: models.StepKindSystem} + setupStepIdx := -1 + + 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. + targetTempDir := e.cfg.QemuPipelines.OverlayDir + if targetTempDir == "" { + targetTempDir = os.TempDir() + } + + tempDir, err := os.MkdirTemp(targetTempDir, "qemu-wf-*") + if err != nil { + return err + } + e.registerCleanup(wid, func(ctx context.Context) error { + return os.RemoveAll(tempDir) + }) + + state := wf.Data.(vmState) + + // generate nocloud seed iso containing engine's authorized_keys + if err := generateSeedISO(tempDir, enginePubKeyStr); 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") + // 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. + // we also have to manage a virtiofsd process for the filesystem, instead of having to + // manage a qcow overlay (see https://ubuntu.com/server/docs/explanation/virtualisation/qemu-microvm/). + // also also need to be able to have some scripts for generating our own images + // (though we should already do this anyway since some like alpine don't provide + // "cloud ready" image files, at least with kernel and initrd) + argv := []string{ + // 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", + // 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), + "-device", "virtio-net-pci,netdev=net0", + "-qmp", fmt.Sprintf("unix:%s,server,nowait", qmpSock), + "-monitor", "none", + } + + if e.cfg.Server.Dev { + argv = append(argv, "-serial", "stdio") + } else { + argv = append(argv, "-serial", "none") + } + + // 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 state.cmdline != "" { + argv = append(argv, "-append", state.cmdline) + } + } else { + // if we are booting with bios, we need to tell it to boot from the disk + argv = append(argv, "-boot", "order=c") + } + + enableKVM := e.cfg.QemuPipelines.EnableKVM + if _, err := os.Stat("/dev/kvm"); err != nil { + if enableKVM { + l.Warn("kvm was requested but /dev/kvm is not accessible; falling back to software emulation", "error", err) + } + enableKVM = false + } + if enableKVM { + argv = append(argv, "-enable-kvm", "-cpu", "host") + } + + // todo(dawn): same with above, we assume x86_64 here, but should allow other archs, + // probably just auto detect as a default + qemuCmd := exec.Command("qemu-system-x86_64", argv...) + qemuCmd.Env = append(os.Environ(), "TMPDIR="+tempDir) + qemuCmd.Stdout = os.Stdout + qemuCmd.Stderr = os.Stderr + + startBoot := time.Now() + if err := qemuCmd.Start(); err != nil { + return fmt.Errorf("starting qemu: %w", err) + } + + var mon *qmp.SocketMonitor + qmpCtx, cancelQmp := context.WithTimeout(ctx, 10*time.Second) + defer cancelQmp() + + // retry qmp connect until the emulator is ready with its unix socket + for { + mon, err = qmp.NewSocketMonitor("unix", qmpSock, 2*time.Second) + if err == nil { + if err = mon.Connect(); err == nil { + break + } + } + 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"}`)) + 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) + } + 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() + + // backoff until the guest finishes booting and starts sshd + var sshClient *ssh.Client + 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 + } + select { + case <-bootCtx.Done(): + _ = qemuCmd.Process.Kill() + _ = mon.Disconnect() + return ErrBootTimeout + 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"}`)) + + done := make(chan error, 1) + go func() { + done <- qemuCmd.Wait() + }() + select { + case <-done: + case <-time.After(time.Second): + _ = qemuCmd.Process.Kill() + // drain Wait after kill to avoid zombie + <-done + } + + _ = mon.Disconnect() + return nil + }) + + 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, + } + + 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] + + 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)) + } + + for k, v := range w.Environment { + appendEnv(k, v) + } + for _, s := range secrets { + appendEnv(s.Key, s.Value) + } + if s, ok := step.(Step); ok { + for k, v := range s.environment { + appendEnv(k, v) + } + } + + session, err := state.sshClient.NewSession() + if err != nil { + return fmt.Errorf("ssh new session: %w", err) + } + defer session.Close() + + stdoutPipe, err := session.StdoutPipe() + if err != nil { + return fmt.Errorf("ssh stdout pipe: %w", err) + } + stderrPipe, err := session.StderrPipe() + if err != nil { + return fmt.Errorf("ssh stderr pipe: %w", err) + } + + tailDone := make(chan error, 1) + go func() { + tailDone <- streamLogs(ctx, stdoutPipe, stderrPipe, idx, wfLogger) + }() + + // 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) + } + + select { + case err := <-tailDone: + if err != nil { + e.l.Warn("log streaming error", "workflow", wid, "step", idx, "error", err) + } + case <-ctx.Done(): + _ = session.Signal(ssh.SIGKILL) + <-tailDone + return engine.ErrTimedOut + } + + if err := session.Wait(); err != nil { + return engine.ErrWorkflowFailed + } + return nil +} + +func (e *Engine) DestroyWorkflow(ctx context.Context, wid models.WorkflowId) error { + fns := e.drainCleanups(wid) + + // cleanups must be executed LIFO to respect resource dependencies + // (e.g. process shutdown before tempdir removal) + for i := len(fns) - 1; i >= 0; i-- { + if err := fns[i](ctx); err != nil { + e.l.Error("failed to cleanup workflow resource", "workflowId", wid, "error", err) + } + } + return nil +} + +func (e *Engine) WorkflowTimeout() time.Duration { + d, err := time.ParseDuration(e.cfg.QemuPipelines.WorkflowTimeout) + if err != nil { + return 5 * time.Minute + } + return d +} + +func (e *Engine) registerCleanup(wid models.WorkflowId, fn cleanupFunc) { + e.cleanupMu.Lock() + defer e.cleanupMu.Unlock() + key := wid.String() + e.cleanup[key] = append(e.cleanup[key], fn) +} + +func (e *Engine) drainCleanups(wid models.WorkflowId) []cleanupFunc { + e.cleanupMu.Lock() + defer e.cleanupMu.Unlock() + key := wid.String() + fns := e.cleanup[key] + delete(e.cleanup, key) + return fns +} diff --git a/spindle/engines/qemu/errors.go b/spindle/engines/qemu/errors.go new file mode 100644 index 00000000..fadc2645 --- /dev/null +++ b/spindle/engines/qemu/errors.go @@ -0,0 +1,8 @@ +package qemu + +import "errors" + +var ( + ErrOOMKilled = errors.New("container died due to OOM kill") + ErrBootTimeout = errors.New("timed out waiting for VM to boot") +) diff --git a/spindle/engines/qemu/logs.go b/spindle/engines/qemu/logs.go new file mode 100644 index 00000000..100fdf36 --- /dev/null +++ b/spindle/engines/qemu/logs.go @@ -0,0 +1,35 @@ +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 new file mode 100644 index 00000000..02c234f1 --- /dev/null +++ b/spindle/engines/qemu/models.go @@ -0,0 +1,33 @@ +package qemu + +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 + + kernel string + initrd string + disk string + cmdline string + shell string +} + +type manifestWorkflow struct { + Image string `yaml:"image"` + Steps []struct { + Name string `yaml:"name"` + Command string `yaml:"command"` + Environment map[string]string `yaml:"environment"` + } `yaml:"steps"` + Dependencies map[string][]string `yaml:"dependencies"` + Environment map[string]string `yaml:"environment"` +} diff --git a/spindle/engines/qemu/setup.go b/spindle/engines/qemu/setup.go new file mode 100644 index 00000000..af3c8fcd --- /dev/null +++ b/spindle/engines/qemu/setup.go @@ -0,0 +1,141 @@ +package qemu + +import ( + "context" + "fmt" + "io" + "net/http" + "os" + "path/filepath" + + "golang.org/x/sync/errgroup" + "tangled.org/core/log" + "tangled.org/core/spindle/config" + "tangled.org/core/spindle/engines/qemu/bakers" +) + +type DefaultImage struct { + disk string + kernel string + initrd string +} + +// todo(dawn): we should be able to just pull this from a json file or something +var defaultImages = map[string]DefaultImage{ + // alpine doesnt have files that will let us bypass bios :( + // we would ideally have our own image generation step... + "alpine-3.23": { + disk: "https://dl-cdn.alpinelinux.org/alpine/v3.23/releases/cloud/nocloud_alpine-3.23.3-x86_64-bios-tiny-r0.qcow2", + kernel: "", + initrd: "", + }, + "ubuntu-24.04": { + disk: "https://cloud-images.ubuntu.com/noble/current/noble-server-cloudimg-amd64.img", + kernel: "https://cloud-images.ubuntu.com/noble/current/unpacked/noble-server-cloudimg-amd64-vmlinuz-generic", + initrd: "https://cloud-images.ubuntu.com/noble/current/unpacked/noble-server-cloudimg-amd64-initrd-generic", + }, +} + +func downloadFile(ctx context.Context, url string, dest string) error { + req, err := http.NewRequestWithContext(ctx, "GET", url, nil) + if err != nil { + return err + } + + resp, err := http.DefaultClient.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("bad status: %s", resp.Status) + } + + tmp := dest + ".tmp" + out, err := os.Create(tmp) + if err != nil { + return err + } + defer os.Remove(tmp) // no-op if rename succeeds + + if _, err = io.Copy(out, resp.Body); err != nil { + out.Close() + return err + } + if err := out.Sync(); err != nil { + out.Close() + return err + } + if err := out.Close(); err != nil { + return err + } + 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 + if imageDir == "" { + return fmt.Errorf("SPINDLE_QEMU_PIPELINES_IMAGE_DIR must be set") + } + + l := log.FromContext(ctx) + l.Info("setting up default qemu images", "imageDir", imageDir) + + if err := os.MkdirAll(imageDir, 0755); err != nil { + return err + } + + g, ctx := errgroup.WithContext(ctx) + g.SetLimit(4) + + for name, img := range defaultImages { + g.Go(func() error { + dir := filepath.Join(imageDir, name) + if err := os.MkdirAll(dir, 0755); err != nil { + 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 + } + + l.Info("downloading image...", "name", name) + if err := downloadFile(ctx, img.disk, diskPath); err != nil { + return fmt.Errorf("downloading %s disk: %w", name, err) + } + + if img.kernel != "" { + if err := downloadFile(ctx, img.kernel, filepath.Join(dir, "kernel")); 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 { + return fmt.Errorf("downloading %s initrd: %w", name, err) + } + } + + if err := bakers.PrepareImage(ctx, cfg, name); err != nil { + return fmt.Errorf("preparing image %s: %w", name, err) + } + + return nil + }) + } + + if err := g.Wait(); err != nil { + return err + } + + l.Info("setup complete") + return nil +} diff --git a/spindle/models/clone.go b/spindle/models/clone.go index d3a17574..b765a74f 100644 --- a/spindle/models/clone.go +++ b/spindle/models/clone.go @@ -55,7 +55,11 @@ func BuildCloneStep(twf tangled.Pipeline_Workflow, tr tangled.Pipeline_TriggerMe } } - repoURL := BuildRepoURL(tr.Repo, dev) + repoURL := BuildRepoURL(tr.Repo) + + if dev { + repoURL = RewriteLocalhost(repoURL, twf.Engine) + } var cloneOpts tangled.Pipeline_CloneOpts if twf.Clone != nil { @@ -102,23 +106,30 @@ func extractCommitSHA(tr tangled.Pipeline_TriggerMetadata) (string, error) { } // BuildRepoURL constructs the repository URL from repo metadata. -func BuildRepoURL(repo *tangled.Pipeline_TriggerRepo, devMode bool) string { - scheme := "https://" - if devMode { - scheme = "http://" +func BuildRepoURL(repo *tangled.Pipeline_TriggerRepo) string { + if repo == nil { + return "" } + scheme := "https://" // Get host from knot host := repo.Knot - // In dev mode, replace localhost with host.docker.internal for Docker networking - if devMode && strings.Contains(host, "localhost") { - host = strings.ReplaceAll(host, "localhost", "host.docker.internal") - } - return fmt.Sprintf("%s%s/%s", scheme, host, *repo.RepoDid) } +// RewriteLocalhost adapts a localhost repository URL for container/VM environments +func RewriteLocalhost(repoURL string, engine string) string { + if !strings.Contains(repoURL, "localhost") { + return repoURL + } + repoURL = strings.ReplaceAll(repoURL, "https://", "http://") + if engine == "qemu" { + return strings.ReplaceAll(repoURL, "localhost", "10.0.2.2") + } + return strings.ReplaceAll(repoURL, "localhost", "host.docker.internal") +} + // buildFetchArgs constructs the arguments for git fetch based on clone options func buildFetchArgs(clone tangled.Pipeline_CloneOpts, sha string) []string { args := []string{} diff --git a/spindle/models/pipeline_env.go b/spindle/models/pipeline_env.go index cf9bf64c..5e9703c8 100644 --- a/spindle/models/pipeline_env.go +++ b/spindle/models/pipeline_env.go @@ -10,7 +10,7 @@ import ( // PipelineEnvVars extracts environment variables from pipeline trigger metadata. // These are framework-provided variables that are injected into workflow steps. -func PipelineEnvVars(tr *tangled.Pipeline_TriggerMetadata, pipelineId PipelineId, devMode bool) map[string]string { +func PipelineEnvVars(tr *tangled.Pipeline_TriggerMetadata, pipelineId PipelineId) map[string]string { if tr == nil { return nil } @@ -34,7 +34,7 @@ func PipelineEnvVars(tr *tangled.Pipeline_TriggerMetadata, pipelineId PipelineId env["TANGLED_REPO_REPO_DID"] = *tr.Repo.RepoDid } env["TANGLED_REPO_DEFAULT_BRANCH"] = tr.Repo.DefaultBranch - env["TANGLED_REPO_URL"] = BuildRepoURL(tr.Repo, devMode) + env["TANGLED_REPO_URL"] = BuildRepoURL(tr.Repo) } switch workflow.TriggerKind(tr.Kind) { diff --git a/spindle/models/pipeline_env_test.go b/spindle/models/pipeline_env_test.go index 574f207a..80907739 100644 --- a/spindle/models/pipeline_env_test.go +++ b/spindle/models/pipeline_env_test.go @@ -27,7 +27,7 @@ func TestPipelineEnvVars_PushBranch(t *testing.T) { Knot: "example.com", Rkey: "123123", } - env := PipelineEnvVars(tr, id, false) + env := PipelineEnvVars(tr, id) // Check standard CI variable if env["CI"] != "true" { @@ -90,7 +90,7 @@ func TestPipelineEnvVars_PushTag(t *testing.T) { Knot: "example.com", Rkey: "123123", } - env := PipelineEnvVars(tr, id, false) + env := PipelineEnvVars(tr, id) if env["TANGLED_REF"] != "refs/tags/v1.2.3" { t.Errorf("Expected TANGLED_REF='refs/tags/v1.2.3', got '%s'", env["TANGLED_REF"]) @@ -123,7 +123,7 @@ func TestPipelineEnvVars_PullRequest(t *testing.T) { Knot: "example.com", Rkey: "123123", } - env := PipelineEnvVars(tr, id, false) + env := PipelineEnvVars(tr, id) // Check ref variables for PR if env["TANGLED_REF"] != "refs/heads/feature-branch" { @@ -179,7 +179,7 @@ func TestPipelineEnvVars_ManualWithInputs(t *testing.T) { Knot: "example.com", Rkey: "123123", } - env := PipelineEnvVars(tr, id, false) + env := PipelineEnvVars(tr, id) // Check manual input variables if env["TANGLED_INPUT_VERSION"] != "1.0.0" { @@ -216,7 +216,7 @@ func TestPipelineEnvVars_DevMode(t *testing.T) { Knot: "example.com", Rkey: "123123", } - env := PipelineEnvVars(tr, id, true) + env := PipelineEnvVars(tr, id) // Dev mode should use http:// and replace localhost with host.docker.internal expectedURL := "http://host.docker.internal:3000/did:plc:boltless" @@ -230,7 +230,7 @@ func TestPipelineEnvVars_NilTrigger(t *testing.T) { Knot: "example.com", Rkey: "123123", } - env := PipelineEnvVars(nil, id, false) + env := PipelineEnvVars(nil, id) if env != nil { t.Error("Expected nil env for nil trigger") @@ -252,7 +252,7 @@ func TestPipelineEnvVars_NilPushData(t *testing.T) { Knot: "example.com", Rkey: "123123", } - env := PipelineEnvVars(tr, id, false) + env := PipelineEnvVars(tr, id) // Should still have repo variables if env["TANGLED_REPO_KNOT"] != "example.com" { diff --git a/spindle/server.go b/spindle/server.go index 67dc63d6..1aee0ad5 100644 --- a/spindle/server.go +++ b/spindle/server.go @@ -25,6 +25,7 @@ import ( "tangled.org/core/spindle/db" "tangled.org/core/spindle/engine" "tangled.org/core/spindle/engines/nixery" + "tangled.org/core/spindle/engines/qemu" "tangled.org/core/spindle/models" "tangled.org/core/spindle/queue" "tangled.org/core/spindle/secrets" @@ -332,8 +333,14 @@ func Run(ctx context.Context) error { return err } + qemuEng, err := qemu.New(ctx, cfg) + if err != nil { + return err + } + s, err := New(ctx, cfg, map[string]models.Engine{ "nixery": nixeryEng, + "qemu": qemuEng, }) if err != nil { return err @@ -409,7 +416,7 @@ func (s *Spindle) processPipeline(ctx context.Context, src eventconsumer.Source, workflows := make(map[models.Engine][]models.Workflow) // Build pipeline environment variables once for all workflows - pipelineEnv := models.PipelineEnvVars(tpl.TriggerMetadata, pipelineId, s.cfg.Server.Dev) + pipelineEnv := models.PipelineEnvVars(tpl.TriggerMetadata, pipelineId) for _, w := range tpl.Workflows { if w != nil { @@ -451,6 +458,13 @@ func (s *Spindle) processPipeline(ctx context.Context, src eventconsumer.Source, } maps.Copy(ewf.Environment, pipelineEnv) + // if in dev mode we have to replace localhost with the correct host alias + if s.cfg.Server.Dev { + if repoUrl, ok := ewf.Environment["TANGLED_REPO_URL"]; ok { + ewf.Environment["TANGLED_REPO_URL"] = models.RewriteLocalhost(repoUrl, w.Engine) + } + } + workflows[eng] = append(workflows[eng], *ewf) err = s.db.StatusPending(models.WorkflowId{