forked from
tangled.org/core
Monorepo for Tangled
spindle/qemu: reorganize and split the vm management
Signed-off-by: dawn <90008@gaze.systems>
+400
-359
+5
-5
spindle/engines/qemu/bakers/baker.go
+5
-5
spindle/engines/qemu/bakers/baker.go
···
11
11
"tangled.org/core/log"
12
12
)
13
13
14
+
type ImageMetadata struct {
15
+
Cmdline string `json:"cmdline"`
16
+
Shell string `json:"shell"`
17
+
}
18
+
14
19
const (
15
20
DiskName = "disk"
16
21
KernelName = "kernel"
···
26
31
func ConfigPath(dir string) string { return filepath.Join(dir, ConfigName) }
27
32
func UserDataPath(dir string) string { return filepath.Join(dir, UserDataName) }
28
33
func SeedISOPath(dir string) string { return filepath.Join(dir, SeedISOName) }
29
-
30
-
type ImageMetadata struct {
31
-
Cmdline string `json:"cmdline"`
32
-
Shell string `json:"shell"`
33
-
}
34
34
35
35
type ImageBaker interface {
36
36
Prepare(ctx context.Context, imageDir string) error
+8
-4
spindle/engines/qemu/cloudinit.go
spindle/engines/qemu/virt/cloudinit.go
+8
-4
spindle/engines/qemu/cloudinit.go
spindle/engines/qemu/virt/cloudinit.go
···
1
-
package qemu
1
+
package virt
2
2
3
3
import (
4
4
"fmt"
5
5
"os"
6
6
"os/exec"
7
7
"path/filepath"
8
+
)
8
9
9
-
"tangled.org/core/spindle/engines/qemu/bakers"
10
-
)
10
+
const SeedISOName = "seed.iso"
11
+
12
+
func SeedISOPath(dir string) string {
13
+
return filepath.Join(dir, SeedISOName)
14
+
}
11
15
12
16
func generateSeedISO(dir string, extraUserData string) error {
13
17
metaData := "instance-id: spindle-vm\nlocal-hostname: spindle\n"
···
32
36
}
33
37
34
38
// nocloud source expects volid "cidata" with joliet and rock ridge extensions
35
-
cmd := exec.Command("genisoimage", "-output", bakers.SeedISOPath(dir), "-volid", "cidata", "-joliet", "-rock", "meta-data", "user-data")
39
+
cmd := exec.Command("genisoimage", "-output", SeedISOPath(dir), "-volid", "cidata", "-joliet", "-rock", "meta-data", "user-data")
36
40
cmd.Dir = dir
37
41
cmd.Stdout = os.Stdout
38
42
cmd.Stderr = os.Stderr
+34
-318
spindle/engines/qemu/engine.go
+34
-318
spindle/engines/qemu/engine.go
···
1
1
package qemu
2
2
3
3
import (
4
-
"bufio"
5
4
"context"
6
-
"encoding/base64"
7
5
"encoding/json"
8
6
"fmt"
9
7
"log/slog"
10
-
"net"
11
8
"os"
12
-
"os/exec"
13
9
"path/filepath"
14
10
"sync"
15
11
"time"
16
12
17
-
"github.com/digitalocean/go-qemu/qmp"
18
13
"gopkg.in/yaml.v3"
19
14
20
15
"tangled.org/core/api/tangled"
···
22
17
"tangled.org/core/spindle/config"
23
18
"tangled.org/core/spindle/engine"
24
19
"tangled.org/core/spindle/engines/qemu/bakers"
20
+
"tangled.org/core/spindle/engines/qemu/virt"
25
21
"tangled.org/core/spindle/models"
26
22
"tangled.org/core/spindle/secrets"
27
23
)
···
81
77
return nil, err
82
78
}
83
79
84
-
swf.Data = vmState{img: img}
80
+
swf.Data = &virt.VMState{Img: img}
85
81
return swf, nil
86
82
}
87
83
88
84
// discover and resolve kernel, initrd, and disk and other config from an image subfolder
89
-
func (e *Engine) resolveImage(name string) (ResolvedImage, error) {
90
-
var img ResolvedImage
85
+
func (e *Engine) resolveImage(name string) (virt.ResolvedImage, error) {
86
+
var img virt.ResolvedImage
91
87
if name == "" {
92
88
name = e.cfg.QemuPipelines.DefaultImage
93
89
}
···
102
98
configPath := bakers.ConfigPath(imageDir)
103
99
104
100
if _, err := os.Stat(diskPath); err == nil {
105
-
img.disk = diskPath
101
+
img.Disk = diskPath
106
102
}
107
103
if _, err := os.Stat(kernelPath); err == nil {
108
-
img.kernel = kernelPath
104
+
img.Kernel = kernelPath
109
105
}
110
106
if _, err := os.Stat(initrdPath); err == nil {
111
-
img.initrd = initrdPath
107
+
img.Initrd = initrdPath
112
108
}
113
109
if b, err := os.ReadFile(configPath); err == nil {
114
110
var meta bakers.ImageMetadata
115
111
if err := json.Unmarshal(b, &meta); err == nil {
116
112
if meta.Cmdline != "" {
117
-
img.cmdline = meta.Cmdline
113
+
img.Cmdline = meta.Cmdline
118
114
}
119
115
if meta.Shell != "" {
120
-
img.shell = meta.Shell
116
+
img.Shell = meta.Shell
121
117
}
122
118
}
123
119
}
124
120
if b, err := os.ReadFile(filepath.Join(imageDir, bakers.UserDataName)); err == nil {
125
-
img.userData = string(b)
121
+
img.UserData = string(b)
126
122
}
127
123
128
-
if img.disk == "" {
124
+
if img.Disk == "" {
129
125
return img, fmt.Errorf("missing '%s' in %s", bakers.DiskName, imageDir)
130
126
}
131
-
if img.kernel != "" && (img.initrd == "" || img.cmdline == "") {
127
+
if img.Kernel != "" && (img.Initrd == "" || img.Cmdline == "") {
132
128
return img, fmt.Errorf("kernel requires initrd and cmdline, but 'initrd' and/or 'cmdline' is missing for %s", name)
133
129
}
134
-
if img.shell == "" {
130
+
if img.Shell == "" {
135
131
return img, fmt.Errorf("shell is not configured for %s", name)
136
132
}
137
133
return img, nil
···
147
143
wfLogger.ControlWriter(setupStepIdx, setupStep, models.StepStatusStart).Write([]byte{0})
148
144
defer wfLogger.ControlWriter(setupStepIdx, setupStep, models.StepStatusEnd).Write([]byte{0})
149
145
150
-
// some systems have tmpfs at /tmp which is not ideal for large workloads
151
-
targetTempDir := e.cfg.QemuPipelines.OverlayDir
152
-
if targetTempDir == "" {
153
-
targetTempDir = os.TempDir()
154
-
}
155
-
156
-
tempDir, err := os.MkdirTemp(targetTempDir, "qemu-wf-*")
157
-
if err != nil {
158
-
return err
159
-
}
160
-
e.registerCleanup(wid, func(ctx context.Context) error {
161
-
return os.RemoveAll(tempDir)
162
-
})
163
-
164
-
state := wf.Data.(vmState)
165
-
img := state.img
166
-
167
-
if err := generateSeedISO(tempDir, img.userData); err != nil {
168
-
return fmt.Errorf("generating seed iso: %w", err)
169
-
}
170
-
171
-
qmpSock := filepath.Join(tempDir, "qmp.sock")
172
-
qgaSock := filepath.Join(tempDir, "qga.sock")
173
-
174
-
// todo(dawn): ideally would be nice if we used qemu with the microvm enabled here...
175
-
// but that is not compatible with cloud-init since it expects real hw enumeration...
176
-
// and we would not be able to use standard cloud images, which is kind of annoying.
177
-
// we also have to manage a virtiofsd process for the filesystem, instead of having to
178
-
// manage a qcow overlay (see https://ubuntu.com/server/docs/explanation/virtualisation/qemu-microvm/).
179
-
// also also need to be able to have some scripts for generating our own images
180
-
// (though we should already do this anyway since some like alpine don't provide
181
-
// "cloud ready" image files, at least with kernel and initrd)
182
-
argv := []string{
183
-
// todo(dawn): ideally probably have "tiers" and let the spindle concile the tier using
184
-
// what the user wants and what the user has exposed to them by the spindle operator?
185
-
"-m", e.cfg.QemuPipelines.Memory, "-smp", fmt.Sprintf("%d", e.cfg.QemuPipelines.SMP),
186
-
"-display", "none", "-monitor", "none", "-nodefaults", "-no-user-config",
187
-
// use snapshot=on to do copy-on-write without having us manage qcow overlays manually
188
-
"-drive", fmt.Sprintf("file=%s,media=disk,snapshot=on,if=virtio", img.disk),
189
-
"-drive", fmt.Sprintf("file=%s,media=cdrom", bakers.SeedISOPath(tempDir)),
190
-
"-netdev", "user,id=net0",
191
-
"-device", "virtio-net-pci,netdev=net0",
192
-
"-qmp", fmt.Sprintf("unix:%s,server,nowait", qmpSock),
193
-
"-chardev", fmt.Sprintf("socket,path=%s,server,nowait,id=qga0", qgaSock),
194
-
"-device", "virtio-serial",
195
-
"-device", "virtserialport,chardev=qga0,name=org.qemu.guest_agent.0",
196
-
}
197
-
198
-
if e.cfg.Server.Dev {
199
-
argv = append(argv, "-serial", "stdio")
200
-
} else {
201
-
argv = append(argv, "-serial", "none")
202
-
}
203
-
204
-
// support booting using qemu bios still, but otherwise we make it faster!
205
-
// incase someone wants to do this for whatever reason...
206
-
if img.kernel != "" {
207
-
argv = append(argv, "-kernel", img.kernel)
208
-
if img.initrd != "" {
209
-
argv = append(argv, "-initrd", img.initrd)
210
-
}
211
-
if img.cmdline != "" {
212
-
argv = append(argv, "-append", img.cmdline)
213
-
}
214
-
} else {
215
-
// booting with bios: explicitly select disk
216
-
argv = append(argv, "-boot", "order=c")
217
-
}
218
-
219
-
enableKVM := e.cfg.QemuPipelines.EnableKVM
220
-
if _, err := os.Stat("/dev/kvm"); err != nil {
221
-
if enableKVM {
222
-
l.Warn("kvm was requested but /dev/kvm is not accessible; falling back to software emulation", "error", err)
223
-
}
224
-
enableKVM = false
225
-
}
226
-
if enableKVM {
227
-
argv = append(argv, "-enable-kvm", "-cpu", "host")
228
-
}
229
-
230
-
// todo(dawn): same with above, we assume x86_64 here, but should allow other archs,
231
-
// probably just auto detect as a default
232
-
qemuCmd := exec.Command("qemu-system-x86_64", argv...)
233
-
qemuCmd.Env = append(os.Environ(), "TMPDIR="+tempDir)
234
-
qemuCmd.Stdout = os.Stdout
235
-
qemuCmd.Stderr = os.Stderr
236
-
237
-
startedBootAt := time.Now()
238
-
if err := qemuCmd.Start(); err != nil {
239
-
return fmt.Errorf("starting qemu: %w", err)
240
-
}
241
-
242
-
var mon *qmp.SocketMonitor
243
-
244
-
// cleanup qemu if we fail to setup at any point below
245
-
setupOk := false
246
-
defer func() {
247
-
if !setupOk {
248
-
_ = qemuCmd.Process.Kill()
249
-
if mon != nil {
250
-
_ = mon.Disconnect()
251
-
}
252
-
}
253
-
}()
254
-
255
-
qmpCtx, cancelQmp := context.WithTimeout(ctx, 10*time.Second)
256
-
defer cancelQmp()
257
-
258
-
// wait for qmp to be ready
259
-
for {
260
-
mon, err = qmp.NewSocketMonitor("unix", qmpSock, 2*time.Second)
261
-
if err == nil {
262
-
if err = mon.Connect(); err == nil {
263
-
break
264
-
}
265
-
}
266
-
select {
267
-
case <-qmpCtx.Done():
268
-
return fmt.Errorf("qmp connect timeout: %w", err)
269
-
case <-time.After(10 * time.Millisecond):
270
-
}
271
-
}
146
+
state := wf.Data.(*virt.VMState)
147
+
img := state.Img
272
148
273
-
status, err := e.qmpQueryStatus(mon)
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,
155
+
}, img, l)
274
156
if err != nil {
275
157
return err
276
158
}
277
159
278
-
l.Info("qemu guest status", "status", status)
279
-
if status != "running" {
280
-
return fmt.Errorf("qemu guest not running (status: %s)", status)
281
-
}
282
-
283
-
qgaCtx, cancelQga := context.WithTimeout(ctx, time.Minute)
284
-
defer cancelQga()
285
-
286
-
// wait for guest agent to be ready (aka boot)
287
-
for {
288
-
if _, err := os.Stat(qgaSock); err == nil {
289
-
pingCtx, cancelPing := context.WithTimeout(qgaCtx, 5*time.Second)
290
-
err = e.qgaGuestPing(pingCtx, qgaSock)
291
-
cancelPing()
292
-
if err == nil {
293
-
l.Info("vm booted and guest-agent ready", "elapsed", time.Since(startedBootAt).Round(time.Millisecond))
294
-
break
295
-
}
296
-
l.Debug("qga guest-ping failed", "error", err)
297
-
} else {
298
-
l.Debug("qga socket not found yet", "path", qgaSock)
299
-
}
300
-
301
-
select {
302
-
case <-qgaCtx.Done():
303
-
return fmt.Errorf("qga connect timeout: %w", err)
304
-
case <-time.After(100 * time.Millisecond):
305
-
}
306
-
}
307
-
308
160
e.registerCleanup(wid, func(ctx context.Context) error {
309
161
// graceful powerdown so guest can sync filesystem
310
-
if err := e.qmpSystemPowerdown(mon); err != nil {
162
+
if err := actualState.QMPSystemPowerdown(); err != nil {
311
163
l.Error("failed to powerdown qemu guest", "workflow", wid, "error", err)
312
164
}
313
165
314
166
done := make(chan error, 1)
315
-
go func() { done <- qemuCmd.Wait() }()
167
+
go func() {
168
+
_, err := actualState.Process.Wait()
169
+
done <- err
170
+
}()
316
171
317
172
select {
318
173
case <-done:
319
174
case <-time.After(5 * time.Second):
320
-
_ = qemuCmd.Process.Kill()
175
+
_ = actualState.Process.Kill()
321
176
<-done // drain to avoid zombie
322
177
}
323
178
324
-
_ = mon.Disconnect()
179
+
_ = actualState.QMPMon.Disconnect()
180
+
_ = os.RemoveAll(actualState.TempDir)
325
181
return nil
326
182
})
327
183
328
-
wf.Data = vmState{
329
-
process: qemuCmd.Process,
330
-
qmpMon: mon,
331
-
qgaPath: qgaSock,
332
-
tempDir: tempDir,
333
-
img: img,
334
-
}
335
-
336
-
setupOk = true
184
+
wf.Data = actualState
337
185
return nil
338
186
}
339
187
340
-
func (e *Engine) qmpRun(mon *qmp.SocketMonitor, command qmp.Command) ([]byte, error) {
341
-
b, err := json.Marshal(command)
342
-
if err != nil {
343
-
return nil, err
344
-
}
345
-
return mon.Run(b)
346
-
}
347
-
348
-
// sends a command to the qemu guest agent and returns the response
349
-
func (e *Engine) qgaRun(ctx context.Context, sock string, command qmp.Command) ([]byte, error) {
350
-
b, err := json.Marshal(command)
351
-
if err != nil {
352
-
return nil, err
353
-
}
354
-
355
-
conn, err := (&net.Dialer{}).DialContext(ctx, "unix", sock)
356
-
if err != nil {
357
-
return nil, err
358
-
}
359
-
defer conn.Close()
360
-
361
-
if dl, ok := ctx.Deadline(); ok {
362
-
_ = conn.SetDeadline(dl)
363
-
}
364
-
365
-
if _, err := conn.Write(append(b, '\n')); err != nil {
366
-
return nil, err
367
-
}
368
-
369
-
return bufio.NewReader(conn).ReadBytes('\n')
370
-
}
371
-
372
-
// query the qemu guest status
373
-
func (e *Engine) qmpQueryStatus(mon *qmp.SocketMonitor) (string, error) {
374
-
raw, err := e.qmpRun(mon, qmp.Command{Execute: "query-status"})
375
-
if err != nil {
376
-
return "", fmt.Errorf("qmp query-status failed: %w", err)
377
-
}
378
-
379
-
var resp struct {
380
-
Return struct {
381
-
Status string `json:"status"`
382
-
} `json:"return"`
383
-
}
384
-
if err := json.Unmarshal(raw, &resp); err != nil {
385
-
return "", fmt.Errorf("qmp query-status parse: %w", err)
386
-
}
387
-
return resp.Return.Status, nil
388
-
}
389
-
390
-
// send a command to qemu to powerdown the guest gracefully
391
-
func (e *Engine) qmpSystemPowerdown(mon *qmp.SocketMonitor) error {
392
-
_, err := e.qmpRun(mon, qmp.Command{Execute: "system_powerdown"})
393
-
return err
394
-
}
395
-
396
-
// ping the guest agent to see if it's ready
397
-
func (e *Engine) qgaGuestPing(ctx context.Context, sock string) error {
398
-
_, err := e.qgaRun(ctx, sock, qmp.Command{Execute: "guest-ping"})
399
-
return err
400
-
}
401
-
402
-
// execute a command on the guest and return its pid
403
-
func (e *Engine) qgaGuestExec(ctx context.Context, sock string, shell string, command string, env []string) (int, error) {
404
-
cmdArgs := []string{shell, "-c", command}
405
-
raw, err := e.qgaRun(ctx, sock, qmp.Command{
406
-
Execute: "guest-exec",
407
-
Args: map[string]any{
408
-
"path": shell,
409
-
"arg": cmdArgs[1:],
410
-
"env": env,
411
-
"capture-output": true,
412
-
},
413
-
})
414
-
if err != nil {
415
-
return 0, fmt.Errorf("qga guest-exec: %w", err)
416
-
}
417
-
418
-
var resp struct {
419
-
Return struct {
420
-
Pid int `json:"pid"`
421
-
} `json:"return"`
422
-
}
423
-
if err := json.Unmarshal(raw, &resp); err != nil {
424
-
return 0, fmt.Errorf("qga guest-exec parse: %w", err)
425
-
}
426
-
return resp.Return.Pid, nil
427
-
}
428
-
429
-
type guestExecStatus struct {
430
-
Exited bool
431
-
ExitCode int
432
-
// these are since the last time we asked for status
433
-
OutData []byte
434
-
ErrData []byte
435
-
}
436
-
437
-
// get the status of a command executed with guest-exec
438
-
func (e *Engine) qgaGuestExecStatus(ctx context.Context, sock string, pid int) (guestExecStatus, error) {
439
-
var status guestExecStatus
440
-
raw, err := e.qgaRun(ctx, sock, qmp.Command{
441
-
Execute: "guest-exec-status",
442
-
Args: map[string]any{"pid": pid},
443
-
})
444
-
if err != nil {
445
-
return status, fmt.Errorf("qga guest-exec-status: %w", err)
446
-
}
447
-
448
-
var resp struct {
449
-
Return struct {
450
-
Exited bool `json:"exited"`
451
-
ExitCode int `json:"exitcode"`
452
-
OutData string `json:"out-data"`
453
-
ErrData string `json:"err-data"`
454
-
} `json:"return"`
455
-
}
456
-
if err := json.Unmarshal(raw, &resp); err != nil {
457
-
return status, fmt.Errorf("qga guest-exec-status parse: %w", err)
458
-
}
459
-
460
-
status.Exited = resp.Return.Exited
461
-
status.ExitCode = resp.Return.ExitCode
462
-
if resp.Return.OutData != "" {
463
-
status.OutData, _ = base64.StdEncoding.DecodeString(resp.Return.OutData)
464
-
}
465
-
if resp.Return.ErrData != "" {
466
-
status.ErrData, _ = base64.StdEncoding.DecodeString(resp.Return.ErrData)
467
-
}
468
-
469
-
return status, nil
470
-
}
471
-
472
188
func (e *Engine) RunStep(ctx context.Context, wid models.WorkflowId, w *models.Workflow, idx int, secrets []secrets.UnlockedSecret, wfLogger models.WorkflowLogger) error {
473
-
state := w.Data.(vmState)
189
+
state := w.Data.(*virt.VMState)
474
190
step := w.Steps[idx]
475
191
476
192
env := make([]string, 0, len(w.Environment)+len(secrets))
···
486
202
}
487
203
}
488
204
489
-
pid, err := e.qgaGuestExec(ctx, state.qgaPath, state.img.shell, step.Command(), env)
205
+
pid, err := state.QGAGuestExec(ctx, step.Command(), env)
490
206
if err != nil {
491
207
return err
492
208
}
493
209
494
210
for {
495
-
status, err := e.qgaGuestExecStatus(ctx, state.qgaPath, pid)
211
+
status, err := state.QGAGuestExecStatus(ctx, pid)
496
212
if err != nil {
497
213
return err
498
214
}
-8
spindle/engines/qemu/errors.go
-8
spindle/engines/qemu/errors.go
-24
spindle/engines/qemu/models.go
-24
spindle/engines/qemu/models.go
···
1
1
package qemu
2
2
3
-
import (
4
-
"os"
5
-
6
-
"github.com/digitalocean/go-qemu/qmp"
7
-
)
8
-
9
-
type vmState struct {
10
-
process *os.Process
11
-
qmpMon *qmp.SocketMonitor
12
-
qgaPath string
13
-
tempDir string
14
-
15
-
img ResolvedImage
16
-
}
17
-
18
-
type ResolvedImage struct {
19
-
kernel string
20
-
initrd string
21
-
disk string
22
-
cmdline string // kernel command line
23
-
shell string // shell to use for workflow steps
24
-
userData string // extra cloud-init user-data
25
-
}
26
-
27
3
type manifestWorkflow struct {
28
4
Image string `yaml:"image"`
29
5
Steps []struct {
+33
spindle/engines/qemu/virt/models.go
+33
spindle/engines/qemu/virt/models.go
···
1
+
package virt
2
+
3
+
import (
4
+
"os"
5
+
6
+
"github.com/digitalocean/go-qemu/qmp"
7
+
)
8
+
9
+
type VMConfig struct {
10
+
Memory string
11
+
SMP int
12
+
EnableKVM bool
13
+
Dev bool
14
+
OverlayDir string
15
+
}
16
+
17
+
type ResolvedImage struct {
18
+
Kernel string
19
+
Initrd string
20
+
Disk string
21
+
Cmdline string // kernel command line
22
+
Shell string // shell to use for workflow steps
23
+
UserData string // extra cloud-init user-data
24
+
}
25
+
26
+
type VMState struct {
27
+
Process *os.Process
28
+
QMPMon *qmp.SocketMonitor
29
+
QGAPath string
30
+
TempDir string
31
+
32
+
Img ResolvedImage
33
+
}
+320
spindle/engines/qemu/virt/vm.go
+320
spindle/engines/qemu/virt/vm.go
···
1
+
package virt
2
+
3
+
import (
4
+
"bufio"
5
+
"context"
6
+
"encoding/base64"
7
+
"encoding/json"
8
+
"fmt"
9
+
"log/slog"
10
+
"net"
11
+
"os"
12
+
"os/exec"
13
+
"path/filepath"
14
+
"time"
15
+
16
+
"github.com/digitalocean/go-qemu/qmp"
17
+
)
18
+
19
+
func StartVM(ctx context.Context, cfg VMConfig, img ResolvedImage, l *slog.Logger) (*VMState, error) {
20
+
// some systems have tmpfs at /tmp which is not ideal for large workloads
21
+
targetTempDir := cfg.OverlayDir
22
+
if targetTempDir == "" {
23
+
targetTempDir = os.TempDir()
24
+
}
25
+
26
+
tempDir, err := os.MkdirTemp(targetTempDir, "qemu-vm-*")
27
+
if err != nil {
28
+
return nil, err
29
+
}
30
+
31
+
state := &VMState{
32
+
TempDir: tempDir,
33
+
Img: img,
34
+
}
35
+
36
+
cleanup := func() {
37
+
_ = state.Close()
38
+
_ = os.RemoveAll(tempDir)
39
+
}
40
+
41
+
if err := generateSeedISO(tempDir, img.UserData); err != nil {
42
+
cleanup()
43
+
return nil, fmt.Errorf("generating seed iso: %w", err)
44
+
}
45
+
46
+
qmpSock := filepath.Join(tempDir, "qmp.sock")
47
+
qgaSock := filepath.Join(tempDir, "qga.sock")
48
+
49
+
state.QGAPath = qgaSock
50
+
51
+
// todo(dawn): ideally would be nice if we used qemu with the microvm enabled here...
52
+
// but that is not compatible with cloud-init since it expects real hw enumeration...
53
+
// and we would not be able to use standard cloud images, which is kind of annoying.
54
+
// we also have to manage a virtiofsd process for the filesystem, instead of having to
55
+
// manage a qcow overlay (see https://ubuntu.com/server/docs/explanation/virtualisation/qemu-microvm/).
56
+
// also also need to be able to have some scripts for generating our own images
57
+
// (though we should already do this anyway since some like alpine don't provide
58
+
// "cloud ready" image files, at least with kernel and initrd)
59
+
argv := []string{
60
+
// todo(dawn): ideally probably have "tiers" and let the spindle concile the tier using
61
+
// what the user wants and what the user has exposed to them by the spindle operator?
62
+
"-m", cfg.Memory, "-smp", fmt.Sprintf("%d", cfg.SMP),
63
+
"-display", "none", "-monitor", "none", "-nodefaults", "-no-user-config",
64
+
// 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),
66
+
"-drive", fmt.Sprintf("file=%s,media=cdrom", SeedISOPath(tempDir)),
67
+
"-netdev", "user,id=net0",
68
+
"-device", "virtio-net-pci,netdev=net0",
69
+
"-qmp", fmt.Sprintf("unix:%s,server,nowait", qmpSock),
70
+
"-chardev", fmt.Sprintf("socket,path=%s,server,nowait,id=qga0", qgaSock),
71
+
"-device", "virtio-serial",
72
+
"-device", "virtserialport,chardev=qga0,name=org.qemu.guest_agent.0",
73
+
}
74
+
75
+
if cfg.Dev {
76
+
argv = append(argv, "-serial", "stdio")
77
+
} else {
78
+
argv = append(argv, "-serial", "none")
79
+
}
80
+
81
+
// support booting using qemu bios still, but otherwise we make it faster!
82
+
// incase someone wants to do this for whatever reason...
83
+
if img.Kernel != "" {
84
+
argv = append(argv, "-kernel", img.Kernel)
85
+
if img.Initrd != "" {
86
+
argv = append(argv, "-initrd", img.Initrd)
87
+
}
88
+
if img.Cmdline != "" {
89
+
argv = append(argv, "-append", img.Cmdline)
90
+
}
91
+
} else {
92
+
argv = append(argv, "-boot", "order=c")
93
+
}
94
+
95
+
enableKVM := cfg.EnableKVM
96
+
if _, err := os.Stat("/dev/kvm"); err != nil {
97
+
if enableKVM {
98
+
l.Warn("kvm was requested but /dev/kvm is not accessible; falling back to software emulation", "error", err)
99
+
}
100
+
enableKVM = false
101
+
}
102
+
if enableKVM {
103
+
argv = append(argv, "-enable-kvm", "-cpu", "host")
104
+
}
105
+
106
+
// todo(dawn): same with above, we assume x86_64 here, but should allow other archs,
107
+
// probably just auto detect as a default
108
+
qemuCmd := exec.Command("qemu-system-x86_64", argv...)
109
+
qemuCmd.Env = append(os.Environ(), "TMPDIR="+tempDir)
110
+
qemuCmd.Stdout = os.Stdout
111
+
qemuCmd.Stderr = os.Stderr
112
+
113
+
startedBootAt := time.Now()
114
+
if err := qemuCmd.Start(); err != nil {
115
+
cleanup()
116
+
return nil, fmt.Errorf("starting qemu: %w", err)
117
+
}
118
+
state.Process = qemuCmd.Process
119
+
120
+
qmpCtx, cancelQmp := context.WithTimeout(ctx, 10*time.Second)
121
+
defer cancelQmp()
122
+
123
+
var mon *qmp.SocketMonitor
124
+
// wait for qmp to be ready
125
+
for {
126
+
mon, err = qmp.NewSocketMonitor("unix", qmpSock, 2*time.Second)
127
+
if err == nil {
128
+
if err = mon.Connect(); err == nil {
129
+
state.QMPMon = mon
130
+
break
131
+
}
132
+
}
133
+
select {
134
+
case <-qmpCtx.Done():
135
+
cleanup()
136
+
return nil, fmt.Errorf("qmp connect timeout: %w", err)
137
+
case <-time.After(10 * time.Millisecond):
138
+
}
139
+
}
140
+
141
+
status, err := state.QMPQueryStatus()
142
+
if err != nil {
143
+
cleanup()
144
+
return nil, err
145
+
}
146
+
147
+
l.Info("qemu guest status", "status", status)
148
+
if status != "running" {
149
+
cleanup()
150
+
return nil, fmt.Errorf("qemu guest not running (status: %s)", status)
151
+
}
152
+
153
+
qgaCtx, cancelQga := context.WithTimeout(ctx, time.Minute)
154
+
defer cancelQga()
155
+
156
+
// wait for guest agent to be ready (aka boot)
157
+
for {
158
+
if _, err := os.Stat(qgaSock); err == nil {
159
+
pingCtx, cancelPing := context.WithTimeout(qgaCtx, 5*time.Second)
160
+
err = state.QGAGuestPing(pingCtx)
161
+
cancelPing()
162
+
if err == nil {
163
+
l.Info("vm booted and guest-agent ready", "elapsed", time.Since(startedBootAt).Round(time.Millisecond))
164
+
break
165
+
}
166
+
l.Debug("qga guest-ping failed", "error", err)
167
+
}
168
+
169
+
select {
170
+
case <-qgaCtx.Done():
171
+
cleanup()
172
+
return nil, fmt.Errorf("qga connect timeout: %w", err)
173
+
case <-time.After(100 * time.Millisecond):
174
+
}
175
+
}
176
+
177
+
return state, nil
178
+
}
179
+
180
+
func (s *VMState) Close() error {
181
+
if s.QMPMon != nil {
182
+
_ = s.QMPMon.Disconnect()
183
+
}
184
+
if s.Process != nil {
185
+
_ = s.Process.Kill()
186
+
}
187
+
return nil
188
+
}
189
+
190
+
func (s *VMState) QMPRun(command qmp.Command) ([]byte, error) {
191
+
b, err := json.Marshal(command)
192
+
if err != nil {
193
+
return nil, err
194
+
}
195
+
return s.QMPMon.Run(b)
196
+
}
197
+
198
+
// sends a command to the qemu guest agent and returns the response
199
+
func (s *VMState) QGARun(ctx context.Context, command qmp.Command) ([]byte, error) {
200
+
b, err := json.Marshal(command)
201
+
if err != nil {
202
+
return nil, err
203
+
}
204
+
205
+
conn, err := (&net.Dialer{}).DialContext(ctx, "unix", s.QGAPath)
206
+
if err != nil {
207
+
return nil, err
208
+
}
209
+
defer conn.Close()
210
+
211
+
if dl, ok := ctx.Deadline(); ok {
212
+
_ = conn.SetDeadline(dl)
213
+
}
214
+
215
+
if _, err := conn.Write(append(b, '\n')); err != nil {
216
+
return nil, err
217
+
}
218
+
219
+
return bufio.NewReader(conn).ReadBytes('\n')
220
+
}
221
+
222
+
// query the qemu guest status
223
+
func (s *VMState) QMPQueryStatus() (string, error) {
224
+
raw, err := s.QMPRun(qmp.Command{Execute: "query-status"})
225
+
if err != nil {
226
+
return "", fmt.Errorf("qmp query-status failed: %w", err)
227
+
}
228
+
229
+
var resp struct {
230
+
Return struct {
231
+
Status string `json:"status"`
232
+
} `json:"return"`
233
+
}
234
+
if err := json.Unmarshal(raw, &resp); err != nil {
235
+
return "", fmt.Errorf("qmp query-status parse: %w", err)
236
+
}
237
+
return resp.Return.Status, nil
238
+
}
239
+
240
+
// send a command to qemu to powerdown the guest gracefully
241
+
func (s *VMState) QMPSystemPowerdown() error {
242
+
_, err := s.QMPRun(qmp.Command{Execute: "system_powerdown"})
243
+
return err
244
+
}
245
+
246
+
// ping the guest agent to see if it's ready
247
+
func (s *VMState) QGAGuestPing(ctx context.Context) error {
248
+
_, err := s.QGARun(ctx, qmp.Command{Execute: "guest-ping"})
249
+
return err
250
+
}
251
+
252
+
// execute a command on the guest and return its pid
253
+
func (s *VMState) QGAGuestExec(ctx context.Context, command string, env []string) (int, error) {
254
+
cmdArgs := []string{s.Img.Shell, "-c", command}
255
+
raw, err := s.QGARun(ctx, qmp.Command{
256
+
Execute: "guest-exec",
257
+
Args: map[string]any{
258
+
"path": s.Img.Shell,
259
+
"arg": cmdArgs[1:],
260
+
"env": env,
261
+
"capture-output": true,
262
+
},
263
+
})
264
+
if err != nil {
265
+
return 0, fmt.Errorf("qga guest-exec: %w", err)
266
+
}
267
+
268
+
var resp struct {
269
+
Return struct {
270
+
Pid int `json:"pid"`
271
+
} `json:"return"`
272
+
}
273
+
if err := json.Unmarshal(raw, &resp); err != nil {
274
+
return 0, fmt.Errorf("qga guest-exec parse: %w", err)
275
+
}
276
+
return resp.Return.Pid, nil
277
+
}
278
+
279
+
type GuestExecStatus struct {
280
+
Exited bool
281
+
ExitCode int
282
+
// these are since the last time we asked for status
283
+
OutData []byte
284
+
ErrData []byte
285
+
}
286
+
287
+
// get the status of a command executed with guest-exec
288
+
func (s *VMState) QGAGuestExecStatus(ctx context.Context, pid int) (GuestExecStatus, error) {
289
+
var status GuestExecStatus
290
+
raw, err := s.QGARun(ctx, qmp.Command{
291
+
Execute: "guest-exec-status",
292
+
Args: map[string]any{"pid": pid},
293
+
})
294
+
if err != nil {
295
+
return status, fmt.Errorf("qga guest-exec-status: %w", err)
296
+
}
297
+
298
+
var resp struct {
299
+
Return struct {
300
+
Exited bool `json:"exited"`
301
+
ExitCode int `json:"exitcode"`
302
+
OutData string `json:"out-data"`
303
+
ErrData string `json:"err-data"`
304
+
} `json:"return"`
305
+
}
306
+
if err := json.Unmarshal(raw, &resp); err != nil {
307
+
return status, fmt.Errorf("qga guest-exec-status parse: %w", err)
308
+
}
309
+
310
+
status.Exited = resp.Return.Exited
311
+
status.ExitCode = resp.Return.ExitCode
312
+
if resp.Return.OutData != "" {
313
+
status.OutData, _ = base64.StdEncoding.DecodeString(resp.Return.OutData)
314
+
}
315
+
if resp.Return.ErrData != "" {
316
+
status.ErrData, _ = base64.StdEncoding.DecodeString(resp.Return.ErrData)
317
+
}
318
+
319
+
return status, nil
320
+
}