forked from
tangled.org/core
Monorepo for Tangled
spindle/qemu: use vms for building the images
Signed-off-by: dawn <90008@gaze.systems>
+336
-315
+91
-64
spindle/engines/qemu/bakers/alpine.go
+91
-64
spindle/engines/qemu/bakers/alpine.go
···
1
1
package bakers
2
2
3
3
import (
4
+
"bytes"
4
5
"context"
5
6
"encoding/json"
6
7
"fmt"
7
-
"io"
8
8
"os"
9
-
"os/exec"
10
-
"path/filepath"
9
+
"time"
11
10
12
11
"tangled.org/core/log"
12
+
"tangled.org/core/spindle/engines/qemu/virt"
13
13
)
14
14
15
-
type AlpineBaker struct {
16
-
nbd *NBDManager
17
-
}
18
-
19
-
func copyFile(src, dst string) error {
20
-
in, err := os.Open(src)
21
-
if err != nil {
22
-
return err
23
-
}
24
-
defer in.Close()
25
-
26
-
out, err := os.Create(dst)
27
-
if err != nil {
28
-
return err
29
-
}
30
-
defer out.Close()
31
-
32
-
if _, err := io.Copy(out, in); err != nil {
33
-
return err
34
-
}
35
-
return out.Sync()
36
-
}
15
+
type AlpineBaker struct{}
37
16
38
17
func (p *AlpineBaker) Prepare(ctx context.Context, dir string) error {
39
18
l := log.FromContext(ctx)
···
44
23
return fmt.Errorf("missing '%s' in %s", DiskName, dir)
45
24
}
46
25
47
-
nbd, err := p.nbd.Connect(ctx, l, diskPath)
48
-
if err != nil {
49
-
return err
50
-
}
51
-
defer nbd.Close()
26
+
// use a temporary user-data to install the guest agent
27
+
userData := `
28
+
packages:
29
+
- qemu-guest-agent
30
+
runcmd:
31
+
- rc-update add qemu-guest-agent default
32
+
- rc-service qemu-guest-agent start
33
+
`
52
34
53
-
// alpine images have no partition table, so we mount the base device directly
54
-
mntDir, err := nbd.Mount(ctx, nbd.Device())
35
+
l.Info("starting vm for alpine preparation", "disk", diskPath)
36
+
img := virt.ResolvedImage{
37
+
Disk: diskPath,
38
+
Shell: "/bin/ash",
39
+
UserData: userData,
40
+
}
41
+
conf := virt.VMConfig{
42
+
Memory: "1G",
43
+
SMP: 1,
44
+
EnableKVM: true,
45
+
EnableDiskCow: false, // writable disk!
46
+
BootTimeout: 2 * time.Minute,
47
+
}
48
+
state, err := virt.StartVM(ctx, conf, img, l)
55
49
if err != nil {
56
-
return err
50
+
return fmt.Errorf("starting preparation vm: %w", err)
57
51
}
58
-
defer nbd.Unmount(mntDir)
52
+
defer state.Close()
53
+
54
+
l.Info("vm is up, updating packages and ensuring guest agent is enabled")
55
+
56
+
alpineInterfaces := `auto lo
57
+
iface lo inet loopback
58
+
59
+
auto eth0
60
+
iface eth0 inet static
61
+
address 10.0.2.15
62
+
netmask 255.255.255.0
63
+
gateway 10.0.2.2
64
+
hostname alpine
65
+
`
59
66
60
-
cleanup, err := bindMountResolvConf(ctx, nbd, mntDir)
61
-
if err != nil {
62
-
return err
67
+
cmds := []string{
68
+
"apk update",
69
+
"apk add --no-cache git curl qemu-guest-agent alpine-conf",
70
+
fmt.Sprintf("cat > /etc/network/interfaces <<EOF\n%sEOF", alpineInterfaces),
71
+
"setup-dns -d local -n 8.8.8.8 -n 8.8.4.4",
72
+
"sed -i 's/need net/after net/' /etc/init.d/chronyd",
73
+
"rc-update del chronyd boot || true",
74
+
"rc-update add chronyd default || true",
75
+
"rc-update add qemu-guest-agent default",
63
76
}
64
-
defer cleanup()
77
+
env := []string{}
65
78
66
-
run := func(command string, args ...string) ([]byte, error) {
67
-
fullArgs := append([]string{mntDir, command}, args...)
68
-
cmd := exec.CommandContext(ctx, "chroot", fullArgs...)
69
-
out, err := cmd.CombinedOutput()
79
+
for _, cmd := range cmds {
80
+
l.Info("running preparation command", "command", cmd)
81
+
var out bytes.Buffer
82
+
exitCode, err := state.QGAExec(ctx, cmd, env, &out, &out)
70
83
if err != nil {
71
-
return nil, fmt.Errorf("%s: %s: %w", command, string(out), err)
84
+
return fmt.Errorf("qga exec '%s': %w", cmd, err)
85
+
}
86
+
if exitCode != 0 {
87
+
return fmt.Errorf("command '%s' failed with exit code %d: %s", cmd, exitCode, out.String())
72
88
}
73
-
return out, nil
74
89
}
75
90
76
-
l.Info("installing packages...")
77
-
if out, err := run("/sbin/apk", "add", "--no-cache", "git", "curl", "qemu-guest-agent"); err != nil {
78
-
return fmt.Errorf("installing packages: %s: %w", string(out), err)
91
+
l.Info("extracting kernel and initrd via qga...")
92
+
kernelBytes, err := state.QGAGuestFileReadFull(ctx, "/boot/vmlinuz-virt")
93
+
if err != nil {
94
+
return fmt.Errorf("extracting kernel: %w", err)
79
95
}
80
-
81
-
l.Info("enabling qemu-guest-agent...")
82
-
if out, err := run("/sbin/rc-update", "add", "qemu-guest-agent", "default"); err != nil {
83
-
return fmt.Errorf("enabling qemu-guest-agent: %s: %w", string(out), err)
96
+
if err := os.WriteFile(KernelPath(dir), kernelBytes, 0644); err != nil {
97
+
return fmt.Errorf("writing kernel: %w", err)
84
98
}
85
99
86
-
l.Info("extracting kernel and initrd...")
87
-
bootDir := filepath.Join(mntDir, "boot")
88
-
kernelSource := filepath.Join(bootDir, "vmlinuz-virt")
89
-
if _, err := os.Stat(kernelSource); os.IsNotExist(err) {
90
-
return fmt.Errorf("kernel 'vmlinuz-virt' not found in %s", bootDir)
100
+
initrdBytes, err := state.QGAGuestFileReadFull(ctx, "/boot/initramfs-virt")
101
+
if err != nil {
102
+
return fmt.Errorf("extracting initrd: %w", err)
91
103
}
92
-
initrdSource := filepath.Join(bootDir, "initramfs-virt")
93
-
if _, err := os.Stat(initrdSource); os.IsNotExist(err) {
94
-
return fmt.Errorf("initrd 'initramfs-virt' not found in %s", bootDir)
104
+
if err := os.WriteFile(InitrdPath(dir), initrdBytes, 0644); err != nil {
105
+
return fmt.Errorf("writing initrd: %w", err)
95
106
}
96
107
97
-
if err := copyFile(kernelSource, KernelPath(dir)); err != nil {
98
-
return fmt.Errorf("extracting kernel: %w", err)
108
+
l.Info("shutting down preparation vm")
109
+
if err := state.QMPSystemPowerdown(); err != nil {
110
+
l.Warn("failed to send powerdown signal", "error", err)
99
111
}
100
-
if err := copyFile(initrdSource, InitrdPath(dir)); err != nil {
101
-
return fmt.Errorf("extracting initrd: %w", err)
112
+
113
+
// wait for the process to exit
114
+
done := make(chan error, 1)
115
+
go func() {
116
+
_, err := state.Process.Wait()
117
+
done <- err
118
+
}()
119
+
120
+
select {
121
+
case err := <-done:
122
+
if err != nil {
123
+
l.Warn("vm exited with error", "error", err)
124
+
}
125
+
case <-time.After(10 * time.Second):
126
+
l.Warn("vm did not shut down gracefully; killing")
127
+
_ = state.Process.Kill()
128
+
case <-ctx.Done():
129
+
return ctx.Err()
102
130
}
103
131
104
132
configPath := ConfigPath(dir)
···
114
142
return fmt.Errorf("writing image metadata: %w", err)
115
143
}
116
144
117
-
l.Info("alpine preparation complete")
118
145
return nil
119
146
}
+13
-19
spindle/engines/qemu/bakers/baker.go
+13
-19
spindle/engines/qemu/bakers/baker.go
···
37
37
}
38
38
39
39
var (
40
-
nbd = &NBDManager{}
41
40
preparers = map[string]ImageBaker{
42
-
"alpine": &AlpineBaker{nbd: nbd},
43
-
"ubuntu": &UbuntuBaker{nbd: nbd},
41
+
"alpine": &AlpineBaker{},
42
+
"ubuntu": &UbuntuBaker{},
44
43
}
45
44
)
46
45
···
52
51
return fmt.Errorf("image directory %s does not exist; did you run 'spindle setup'?", imageDir)
53
52
}
54
53
55
-
l := log.FromContext(ctx)
56
-
l.Info("preparing image", "name", name)
54
+
l := log.FromContext(ctx).WithGroup(name)
55
+
l.Info("preparing image")
57
56
58
57
diskPath := DiskPath(imageDir)
59
58
// todo(dawn): this should be configurable
···
67
66
parts := strings.Split(name, "-")
68
67
if preparer, ok := preparers[parts[0]]; ok {
69
68
l.Info("running preparer", "type", parts[0])
70
-
return preparer.Prepare(ctx, imageDir)
69
+
err := preparer.Prepare(log.IntoContext(ctx, l), imageDir)
70
+
if err != nil {
71
+
l.Error("preparer failed", "error", err)
72
+
return err
73
+
}
74
+
l.Info("preparation complete")
75
+
return nil
71
76
}
72
77
73
-
return nil
74
-
}
78
+
l.Error("unknown image type", "type", parts[0])
75
79
76
-
// bind mount resolve conf for networking
77
-
func bindMountResolvConf(ctx context.Context, nbd *NBDHandle, mntDir string) (func(), error) {
78
-
resolvDest := filepath.Join(mntDir, "etc/resolv.conf")
79
-
_ = os.Remove(resolvDest) // remove if it's a symlink or existing file
80
-
if err := os.WriteFile(resolvDest, nil, 0644); err != nil {
81
-
return nil, fmt.Errorf("creating resolv.conf mount point: %w", err)
82
-
}
83
-
if out, err := exec.CommandContext(ctx, "mount", "--bind", "/etc/resolv.conf", resolvDest).CombinedOutput(); err != nil {
84
-
return nil, fmt.Errorf("bind mount resolv.conf: %s: %w", string(out), err)
85
-
}
86
-
return func() { nbd.Unmount(resolvDest) }, nil
80
+
return nil
87
81
}
-133
spindle/engines/qemu/bakers/nbd.go
-133
spindle/engines/qemu/bakers/nbd.go
···
1
-
package bakers
2
-
3
-
import (
4
-
"context"
5
-
"fmt"
6
-
"log/slog"
7
-
"os"
8
-
"os/exec"
9
-
"strings"
10
-
"sync"
11
-
"time"
12
-
)
13
-
14
-
type NBDManager struct {
15
-
mu sync.Mutex
16
-
cleanupOnce sync.Once
17
-
}
18
-
19
-
type NBDHandle struct {
20
-
mgr *NBDManager
21
-
device string
22
-
l *slog.Logger
23
-
}
24
-
25
-
func (n *NBDManager) Connect(ctx context.Context, l *slog.Logger, diskPath string) (*NBDHandle, error) {
26
-
// cleanup in case we have failed state
27
-
n.cleanupOnce.Do(func() {
28
-
n.Cleanup(ctx, l)
29
-
})
30
-
31
-
n.mu.Lock()
32
-
33
-
device := "/dev/nbd12" // use something other than nbd0
34
-
35
-
if _, err := os.Stat(device); os.IsNotExist(err) {
36
-
l.Info("nbd module not loaded, trying to load it...")
37
-
if out, err := exec.Command("modprobe", "nbd", "max_part=8").CombinedOutput(); err != nil {
38
-
n.mu.Unlock()
39
-
return nil, fmt.Errorf("modprobe nbd: %s: %w", string(out), err)
40
-
}
41
-
}
42
-
43
-
l.Info("connecting disk via nbd...", "disk", diskPath)
44
-
if out, err := exec.CommandContext(ctx, "qemu-nbd", "--connect="+device, diskPath).CombinedOutput(); err != nil {
45
-
n.mu.Unlock()
46
-
return nil, fmt.Errorf("qemu-nbd connect: %s: %w", string(out), err)
47
-
}
48
-
49
-
waitCtx, cancel := context.WithTimeout(ctx, 3*time.Second)
50
-
defer cancel()
51
-
for {
52
-
if _, err := os.Stat(device); err == nil {
53
-
break
54
-
}
55
-
select {
56
-
case <-waitCtx.Done():
57
-
n.mu.Unlock()
58
-
return nil, fmt.Errorf("timeout waiting for %s to appear", device)
59
-
case <-time.After(50 * time.Millisecond):
60
-
}
61
-
}
62
-
63
-
// sleep 100ms just to make sure...
64
-
time.Sleep(100 * time.Millisecond)
65
-
66
-
return &NBDHandle{mgr: n, device: device, l: l}, nil
67
-
}
68
-
69
-
// mounts the device node specified
70
-
func (h *NBDHandle) Mount(ctx context.Context, device string) (string, error) {
71
-
mntDir, err := os.MkdirTemp("", "spindle-mnt-*")
72
-
if err != nil {
73
-
return "", err
74
-
}
75
-
76
-
h.l.Info("mounting nbd device...", "device", device)
77
-
if out, err := exec.CommandContext(ctx, "mount", device, mntDir).CombinedOutput(); err != nil {
78
-
os.RemoveAll(mntDir)
79
-
return "", fmt.Errorf("mount %s: %s: %w", device, string(out), err)
80
-
}
81
-
82
-
return mntDir, nil
83
-
}
84
-
85
-
func (h *NBDHandle) Unmount(mntDir string) {
86
-
if out, err := exec.Command("umount", mntDir).CombinedOutput(); err != nil {
87
-
h.l.Error("umount failed", "error", err, "output", string(out))
88
-
}
89
-
os.RemoveAll(mntDir)
90
-
}
91
-
92
-
// returns the base node (/dev/nbd0)
93
-
func (h *NBDHandle) Device() string {
94
-
return h.device
95
-
}
96
-
97
-
// disconnects the nbd and releases the lock
98
-
func (h *NBDHandle) Close() {
99
-
defer h.mgr.mu.Unlock()
100
-
if out, err := exec.Command("qemu-nbd", "--disconnect", h.device).CombinedOutput(); err != nil {
101
-
h.l.Error("qemu-nbd disconnect failed", "error", err, "output", string(out))
102
-
}
103
-
}
104
-
105
-
func (n *NBDManager) Cleanup(ctx context.Context, l *slog.Logger) {
106
-
l.Info("cleaning up stale nbd state...")
107
-
108
-
// 1. disconnect nbd0
109
-
// do this first to make sure the device is not busy when we try to umount
110
-
_ = exec.CommandContext(ctx, "qemu-nbd", "--disconnect", "/dev/nbd0").Run()
111
-
112
-
// 2. find and unmount stale spindle-mnt mounts
113
-
data, err := os.ReadFile("/proc/mounts")
114
-
if err != nil {
115
-
l.Error("failed to read /proc/mounts", "error", err)
116
-
return
117
-
}
118
-
119
-
for _, line := range strings.Split(string(data), "\n") {
120
-
fields := strings.Fields(line)
121
-
if len(fields) < 2 {
122
-
continue
123
-
}
124
-
mnt := fields[1]
125
-
if strings.Contains(mnt, "spindle-mnt-") {
126
-
l.Info("unmounting stale mount", "path", mnt)
127
-
if out, err := exec.CommandContext(ctx, "umount", "-l", mnt).CombinedOutput(); err != nil {
128
-
l.Error("umount failed", "path", mnt, "error", err, "output", string(out))
129
-
}
130
-
os.RemoveAll(mnt)
131
-
}
132
-
}
133
-
}
+82
-58
spindle/engines/qemu/bakers/ubuntu.go
+82
-58
spindle/engines/qemu/bakers/ubuntu.go
···
1
1
package bakers
2
2
3
3
import (
4
+
"bytes"
4
5
"context"
5
6
"encoding/json"
6
7
"fmt"
7
8
"os"
8
-
"os/exec"
9
9
"path/filepath"
10
10
"time"
11
11
12
12
"tangled.org/core/log"
13
+
"tangled.org/core/spindle/engines/qemu/virt"
13
14
)
14
15
15
-
type UbuntuBaker struct {
16
-
nbd *NBDManager
17
-
}
16
+
type UbuntuBaker struct{}
18
17
19
18
func (p *UbuntuBaker) Prepare(ctx context.Context, dir string) error {
20
19
l := log.FromContext(ctx)
···
28
27
29
28
diskPath := DiskPath(dir)
30
29
31
-
nbd, err := p.nbd.Connect(ctx, l, diskPath)
30
+
// use a temporary user-data to install the guest agent
31
+
// so we can use vm management for the rest of the preparation
32
+
userData := `
33
+
packages:
34
+
- qemu-guest-agent
35
+
runcmd:
36
+
- systemctl enable --now qemu-guest-agent
37
+
`
38
+
39
+
l.Info("starting vm for ubuntu preparation", "disk", diskPath)
40
+
img := virt.ResolvedImage{
41
+
Disk: diskPath,
42
+
Kernel: KernelPath(dir),
43
+
Initrd: InitrdPath(dir),
44
+
Cmdline: "root=LABEL=cloudimg-rootfs rw console=ttyS0",
45
+
Shell: "/bin/bash",
46
+
UserData: userData,
47
+
}
48
+
conf := virt.VMConfig{
49
+
Memory: "1G",
50
+
SMP: 1,
51
+
EnableKVM: true,
52
+
EnableDiskCow: false, // make sure we have a writable disk!
53
+
BootTimeout: 2 * time.Minute,
54
+
}
55
+
state, err := virt.StartVM(ctx, conf, img, l)
32
56
if err != nil {
33
-
return err
57
+
return fmt.Errorf("starting preparation vm: %w", err)
34
58
}
35
-
defer nbd.Close()
59
+
defer state.Close()
60
+
61
+
l.Info("vm is up, updating apt repos and ensuring guest agent is enabled")
62
+
63
+
ubuntuNetwork := `[Match]
64
+
Name=ens3
36
65
37
-
// qemu-nbd might not have scanned partitions yet
38
-
if out, err := exec.CommandContext(ctx, "partprobe", nbd.Device()).CombinedOutput(); err != nil {
39
-
l.Warn("partprobe failed", "error", err, "output", string(out))
66
+
[Network]
67
+
Address=10.0.2.15/24
68
+
Gateway=10.0.2.2
69
+
`
70
+
resolvConf := `nameserver 8.8.8.8
71
+
nameserver 8.8.4.4
72
+
`
73
+
74
+
cmds := []string{
75
+
"cloud-init status --wait",
76
+
"apt-get update",
77
+
"apt-get install -y qemu-guest-agent",
78
+
"mkdir -p /etc/systemd/network",
79
+
fmt.Sprintf("cat > /etc/systemd/network/25-ens3.network <<EOF\n%sEOF", ubuntuNetwork),
80
+
fmt.Sprintf("cat > /etc/resolv.conf <<EOF\n%sEOF", resolvConf),
81
+
"systemctl enable systemd-networkd",
82
+
"systemctl enable --now qemu-guest-agent",
83
+
"apt-get clean",
40
84
}
41
-
_ = exec.CommandContext(ctx, "udevadm", "settle").Run()
85
+
env := []string{"DEBIAN_FRONTEND=noninteractive"}
42
86
43
-
// wait for the partition to appear
44
-
partition := nbd.Device() + "p1"
45
-
waitCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
46
-
defer cancel()
47
-
for {
48
-
if _, err := os.Stat(partition); err == nil {
49
-
break
87
+
for _, cmd := range cmds {
88
+
l.Info("running preparation command", "command", cmd)
89
+
var out bytes.Buffer
90
+
exitCode, err := state.QGAExec(ctx, cmd, env, &out, &out)
91
+
if err != nil {
92
+
return fmt.Errorf("qga exec '%s': %w", cmd, err)
50
93
}
51
-
select {
52
-
case <-waitCtx.Done():
53
-
return fmt.Errorf("timeout waiting for partition %s to appear", partition)
54
-
case <-time.After(100 * time.Millisecond):
94
+
if exitCode != 0 {
95
+
return fmt.Errorf("command '%s' failed with exit code %d: %s", cmd, exitCode, out.String())
55
96
}
56
97
}
57
98
58
-
mntDir, err := nbd.Mount(ctx, partition)
59
-
if err != nil {
60
-
return err
99
+
l.Info("shutting down preparation vm")
100
+
if err := state.QMPSystemPowerdown(); err != nil {
101
+
l.Warn("failed to send powerdown signal", "error", err)
61
102
}
62
-
defer nbd.Unmount(mntDir)
63
103
64
-
// run := func(command string, args ...string) ([]byte, error) {
65
-
// fullArgs := append([]string{mntDir, "/usr/bin/" + command}, args...)
66
-
// cmd := exec.CommandContext(ctx, "chroot", fullArgs...)
67
-
// cmd.Env = append(os.Environ(), "DEBIAN_FRONTEND=noninteractive")
68
-
// out, err := cmd.CombinedOutput()
69
-
// if err != nil {
70
-
// return nil, fmt.Errorf("chroot run %s: %s: %w", command, string(out), err)
71
-
// }
72
-
// return out, nil
73
-
// }
74
-
75
-
// cleanup, err := bindMountResolvConf(ctx, nbd, mntDir)
76
-
// if err != nil {
77
-
// return err
78
-
// }
79
-
// defer cleanup()
104
+
done := make(chan error, 1)
105
+
go func() {
106
+
_, err := state.Process.Wait()
107
+
done <- err
108
+
}()
80
109
81
-
// todo(dawn): i tried doing apt stuff in chroot here but no budge :/
82
-
// i didnt want to hardcode in a qemu-guest-agent either nor replicating apt repo setup...
83
-
// so instead we just write a later user-data file that does it for us at boot.
110
+
select {
111
+
case err := <-done:
112
+
if err != nil {
113
+
l.Warn("vm exited with error", "error", err)
114
+
}
115
+
case <-time.After(10 * time.Second):
116
+
l.Warn("vm did not shut down gracefully; killing")
117
+
_ = state.Process.Kill()
118
+
case <-ctx.Done():
119
+
return ctx.Err()
120
+
}
84
121
85
122
configPath := ConfigPath(dir)
86
123
metadata := ImageMetadata{
···
95
132
return fmt.Errorf("writing image metadata: %w", err)
96
133
}
97
134
98
-
// todo(dawn): this should be done while preparing the image...
99
-
userDataPath := UserDataPath(dir)
100
-
userData := `
101
-
packages:
102
-
- qemu-guest-agent
103
-
runcmd:
104
-
- systemctl enable --now qemu-guest-agent
105
-
`
106
-
if err := os.WriteFile(userDataPath, []byte(userData), 0644); err != nil {
107
-
return fmt.Errorf("writing user-data: %w", err)
108
-
}
109
-
110
-
l.Info("ubuntu preparation complete")
111
135
return nil
112
136
}
+15
-31
spindle/engines/qemu/engine.go
+15
-31
spindle/engines/qemu/engine.go
···
147
147
img := state.Img
148
148
149
149
actualState, err := virt.StartVM(ctx, virt.VMConfig{
150
-
Memory: e.cfg.QemuPipelines.Memory,
151
-
SMP: e.cfg.QemuPipelines.SMP,
152
-
EnableKVM: e.cfg.QemuPipelines.EnableKVM,
153
-
Dev: e.cfg.Server.Dev,
154
-
OverlayDir: e.cfg.QemuPipelines.OverlayDir,
150
+
Memory: e.cfg.QemuPipelines.Memory,
151
+
SMP: e.cfg.QemuPipelines.SMP,
152
+
EnableKVM: e.cfg.QemuPipelines.EnableKVM,
153
+
EnableDiskCow: true,
154
+
Dev: e.cfg.Server.Dev,
155
+
OverlayDir: e.cfg.QemuPipelines.OverlayDir,
156
+
BootTimeout: time.Minute,
155
157
}, img, l)
156
158
if err != nil {
157
159
return err
···
202
204
}
203
205
}
204
206
205
-
pid, err := state.QGAGuestExec(ctx, step.Command(), env)
207
+
stdout := wfLogger.DataWriter(idx, "stdout")
208
+
stderr := wfLogger.DataWriter(idx, "stderr")
209
+
exitCode, err := state.QGAExec(ctx, step.Command(), env, stdout, stderr)
206
210
if err != nil {
211
+
if ctx.Err() != nil {
212
+
return engine.ErrTimedOut
213
+
}
207
214
return err
208
215
}
209
216
210
-
for {
211
-
status, err := state.QGAGuestExecStatus(ctx, pid)
212
-
if err != nil {
213
-
return err
214
-
}
215
-
216
-
if len(status.OutData) > 0 {
217
-
_, _ = wfLogger.DataWriter(idx, "stdout").Write(status.OutData)
218
-
}
219
-
if len(status.ErrData) > 0 {
220
-
_, _ = wfLogger.DataWriter(idx, "stderr").Write(status.ErrData)
221
-
}
222
-
223
-
if status.Exited {
224
-
if status.ExitCode != 0 {
225
-
return engine.ErrWorkflowFailed
226
-
}
227
-
break
228
-
}
229
-
230
-
select {
231
-
case <-ctx.Done():
232
-
return engine.ErrTimedOut
233
-
case <-time.After(100 * time.Millisecond):
234
-
}
217
+
if exitCode != 0 {
218
+
return engine.ErrWorkflowFailed
235
219
}
236
220
return nil
237
221
}
+8
-5
spindle/engines/qemu/virt/models.go
+8
-5
spindle/engines/qemu/virt/models.go
···
2
2
3
3
import (
4
4
"os"
5
+
"time"
5
6
6
7
"github.com/digitalocean/go-qemu/qmp"
7
8
)
8
9
9
10
type VMConfig struct {
10
-
Memory string
11
-
SMP int
12
-
EnableKVM bool
13
-
Dev bool
14
-
OverlayDir string
11
+
Memory string
12
+
SMP int
13
+
EnableKVM bool
14
+
EnableDiskCow bool
15
+
Dev bool
16
+
OverlayDir string
17
+
BootTimeout time.Duration
15
18
}
16
19
17
20
type ResolvedImage struct {
+127
-5
spindle/engines/qemu/virt/vm.go
+127
-5
spindle/engines/qemu/virt/vm.go
···
6
6
"encoding/base64"
7
7
"encoding/json"
8
8
"fmt"
9
+
"io"
9
10
"log/slog"
10
11
"net"
11
12
"os"
···
17
18
)
18
19
19
20
func StartVM(ctx context.Context, cfg VMConfig, img ResolvedImage, l *slog.Logger) (*VMState, error) {
21
+
if cfg.BootTimeout == 0 {
22
+
cfg.BootTimeout = time.Minute
23
+
}
24
+
20
25
// some systems have tmpfs at /tmp which is not ideal for large workloads
21
26
targetTempDir := cfg.OverlayDir
22
27
if targetTempDir == "" {
···
62
67
"-m", cfg.Memory, "-smp", fmt.Sprintf("%d", cfg.SMP),
63
68
"-display", "none", "-monitor", "none", "-nodefaults", "-no-user-config",
64
69
// use snapshot=on to do copy-on-write without having us manage qcow overlays manually
65
-
"-drive", fmt.Sprintf("file=%s,media=disk,snapshot=on,if=virtio", img.Disk),
70
+
"-drive", fmt.Sprintf("file=%s,media=disk,snapshot=%s,if=virtio", img.Disk, func() string {
71
+
if cfg.EnableDiskCow {
72
+
return "on"
73
+
}
74
+
return "off"
75
+
}()),
66
76
"-drive", fmt.Sprintf("file=%s,media=cdrom", SeedISOPath(tempDir)),
67
77
"-netdev", "user,id=net0",
68
78
"-device", "virtio-net-pci,netdev=net0",
···
75
85
if cfg.Dev {
76
86
argv = append(argv, "-serial", "stdio")
77
87
} else {
78
-
argv = append(argv, "-serial", "none")
88
+
argv = append(argv, "-serial", "null")
79
89
}
80
90
81
91
// support booting using qemu bios still, but otherwise we make it faster!
···
150
160
return nil, fmt.Errorf("qemu guest not running (status: %s)", status)
151
161
}
152
162
153
-
qgaCtx, cancelQga := context.WithTimeout(ctx, time.Minute)
163
+
qgaCtx, cancelQga := context.WithTimeout(ctx, cfg.BootTimeout)
154
164
defer cancelQga()
155
165
156
166
// wait for guest agent to be ready (aka boot)
157
167
for {
158
168
if _, err := os.Stat(qgaSock); err == nil {
159
-
pingCtx, cancelPing := context.WithTimeout(qgaCtx, 5*time.Second)
169
+
pingCtx, cancelPing := context.WithTimeout(qgaCtx, 250*time.Millisecond)
160
170
err = state.QGAGuestPing(pingCtx)
161
171
cancelPing()
162
172
if err == nil {
163
173
l.Info("vm booted and guest-agent ready", "elapsed", time.Since(startedBootAt).Round(time.Millisecond))
164
174
break
165
175
}
166
-
l.Debug("qga guest-ping failed", "error", err)
176
+
// l.Debug("qga guest-ping failed", "error", err)
167
177
}
168
178
169
179
select {
···
318
328
319
329
return status, nil
320
330
}
331
+
332
+
// qgaexec executes a command on the guest and waits for it to finish, writing its output to stdout/stderr
333
+
func (s *VMState) QGAExec(ctx context.Context, command string, env []string, stdout, stderr io.Writer) (int, error) {
334
+
pid, err := s.QGAGuestExec(ctx, command, env)
335
+
if err != nil {
336
+
return 0, err
337
+
}
338
+
339
+
for {
340
+
status, err := s.QGAGuestExecStatus(ctx, pid)
341
+
if err != nil {
342
+
return 0, err
343
+
}
344
+
345
+
if len(status.OutData) > 0 && stdout != nil {
346
+
_, _ = stdout.Write(status.OutData)
347
+
}
348
+
if len(status.ErrData) > 0 && stderr != nil {
349
+
_, _ = stderr.Write(status.ErrData)
350
+
}
351
+
352
+
if status.Exited {
353
+
return status.ExitCode, nil
354
+
}
355
+
356
+
select {
357
+
case <-ctx.Done():
358
+
return 0, ctx.Err()
359
+
case <-time.After(100 * time.Millisecond):
360
+
}
361
+
}
362
+
}
363
+
364
+
// qgaguestfileopen opens a file on the guest and returns a handle
365
+
func (s *VMState) QGAGuestFileOpen(ctx context.Context, path string, mode string) (int, error) {
366
+
raw, err := s.QGARun(ctx, qmp.Command{
367
+
Execute: "guest-file-open",
368
+
Args: map[string]any{"path": path, "mode": mode},
369
+
})
370
+
if err != nil {
371
+
return 0, fmt.Errorf("qga guest-file-open: %w", err)
372
+
}
373
+
374
+
var resp struct {
375
+
Return int `json:"return"`
376
+
}
377
+
if err := json.Unmarshal(raw, &resp); err != nil {
378
+
return 0, fmt.Errorf("qga guest-file-open parse: %w", err)
379
+
}
380
+
return resp.Return, nil
381
+
}
382
+
383
+
// qgaguestfileread reads data from an open file handle
384
+
func (s *VMState) QGAGuestFileRead(ctx context.Context, handle int, count int) ([]byte, bool, error) {
385
+
raw, err := s.QGARun(ctx, qmp.Command{
386
+
Execute: "guest-file-read",
387
+
Args: map[string]any{"handle": handle, "count": count},
388
+
})
389
+
if err != nil {
390
+
return nil, false, fmt.Errorf("qga guest-file-read: %w", err)
391
+
}
392
+
393
+
var resp struct {
394
+
Return struct {
395
+
BufB64 string `json:"buf-b64"`
396
+
EOF bool `json:"eof"`
397
+
} `json:"return"`
398
+
}
399
+
if err := json.Unmarshal(raw, &resp); err != nil {
400
+
return nil, false, fmt.Errorf("qga guest-file-read parse: %w", err)
401
+
}
402
+
403
+
data, err := base64.StdEncoding.DecodeString(resp.Return.BufB64)
404
+
if err != nil {
405
+
return nil, false, fmt.Errorf("qga guest-file-read decode: %w", err)
406
+
}
407
+
408
+
return data, resp.Return.EOF, nil
409
+
}
410
+
411
+
// qgaguestfileclose closes an open file handle
412
+
func (s *VMState) QGAGuestFileClose(ctx context.Context, handle int) error {
413
+
_, err := s.QGARun(ctx, qmp.Command{
414
+
Execute: "guest-file-close",
415
+
Args: map[string]any{"handle": handle},
416
+
})
417
+
return err
418
+
}
419
+
420
+
// qgaguestfilereadfull reads an entire file from the guest and returns its content
421
+
func (s *VMState) QGAGuestFileReadFull(ctx context.Context, path string) ([]byte, error) {
422
+
handle, err := s.QGAGuestFileOpen(ctx, path, "r")
423
+
if err != nil {
424
+
return nil, err
425
+
}
426
+
defer func() { _ = s.QGAGuestFileClose(ctx, handle) }()
427
+
428
+
var fullData []byte
429
+
const chunkSize = 1024 * 1024 // 1MB chunks
430
+
for {
431
+
data, eof, err := s.QGAGuestFileRead(ctx, handle, chunkSize)
432
+
if err != nil {
433
+
return nil, err
434
+
}
435
+
fullData = append(fullData, data...)
436
+
if eof {
437
+
break
438
+
}
439
+
}
440
+
441
+
return fullData, nil
442
+
}