diff --git a/matcher/helpers.go b/matcher/helpers.go index bc1bffd..81fe27b 100644 --- a/matcher/helpers.go +++ b/matcher/helpers.go @@ -12,6 +12,7 @@ import ( "github.com/pkg/errors" "github.com/spectrocloud/peg/pkg/controller" + "github.com/spectrocloud/peg/pkg/machine" "github.com/spectrocloud/peg/pkg/machine/types" "go.uber.org/zap/buffer" @@ -55,6 +56,23 @@ func (vm VM) Reboot(t ...int) { machineReboot(vm.machine, t...) } +// HardPowerCycle SIGKILLs the VM and relaunches it against the same on-disk +// state, then waits up to sshTimeoutSec for SSH. QEMU backend only. +func (vm VM) HardPowerCycle(ctx context.Context, sshTimeoutSec int) (context.Context, error) { + q, ok := vm.machine.(*machine.QEMU) + if !ok { + return ctx, errors.Errorf("HardPowerCycle requires QEMU backend, got %T", vm.machine) + } + newCtx, err := q.HardPowerCycle(ctx) + if err != nil { + return ctx, err + } + if err := waitForMachineConnection(newCtx, vm.machine, sshTimeoutSec); err != nil { + return newCtx, err + } + return newCtx, nil +} + func (vm VM) DetachCD() error { return vm.machine.DetachCD() } @@ -272,6 +290,35 @@ func machineEventuallyConnects(m types.Machine, t ...int) { }, time.Duration(dur)*time.Second, 5*time.Second).Should(Equal("ping\n"), "Machine did not become reachable in time") } +func waitForMachineConnection(ctx context.Context, m types.Machine, timeoutSec int) error { + if timeoutSec <= 0 { + return fmt.Errorf("SSH timeout must be greater than zero") + } + + deadline := time.NewTimer(time.Duration(timeoutSec) * time.Second) + defer deadline.Stop() + + var lastErr error + for { + out, err := m.Command("echo ping") + if err == nil && out == "ping\n" { + return nil + } + lastErr = err + + select { + case <-ctx.Done(): + return fmt.Errorf("QEMU stopped before SSH reconnected: %w", ctx.Err()) + case <-deadline.C: + if lastErr != nil { + return fmt.Errorf("SSH did not reconnect within %ds: %w", timeoutSec, lastErr) + } + return fmt.Errorf("SSH did not reconnect within %ds", timeoutSec) + case <-time.After(5 * time.Second): + } + } +} + func machineReboot(m types.Machine, t ...int) { machineSudo(m, "reboot") //nolint:errcheck time.Sleep(1 * time.Minute) diff --git a/matcher/helpers_test.go b/matcher/helpers_test.go new file mode 100644 index 0000000..131c5c2 --- /dev/null +++ b/matcher/helpers_test.go @@ -0,0 +1,22 @@ +package matcher + +import ( + "context" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + "github.com/spectrocloud/peg/pkg/machine" +) + +var _ = Describe("VM.HardPowerCycle", func() { + Context("when the backend is not QEMU", func() { + It("returns an informative error and the original context", func() { + vm := VM{machine: &machine.Docker{}} + ctx, err := vm.HardPowerCycle(context.Background(), 60) + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("HardPowerCycle requires QEMU backend")) + Expect(ctx).NotTo(BeNil()) + }) + }) +}) diff --git a/matcher/matcher_suite_test.go b/matcher/matcher_suite_test.go new file mode 100644 index 0000000..dd22d54 --- /dev/null +++ b/matcher/matcher_suite_test.go @@ -0,0 +1,13 @@ +package matcher_test + +import ( + "testing" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +func TestMatcher(t *testing.T) { + RegisterFailHandler(Fail) + RunSpecs(t, "Matcher Suite") +} diff --git a/pkg/machine/machine.go b/pkg/machine/machine.go index 784038e..f5ed02c 100644 --- a/pkg/machine/machine.go +++ b/pkg/machine/machine.go @@ -7,6 +7,7 @@ import ( "os" "path/filepath" "strings" + "sync" "time" "github.com/codingsince1985/checksum" @@ -117,31 +118,43 @@ func prepare(mc *types.MachineConfig) error { return nil } -func monitor(ctx context.Context, p *process.Process, f func(p *process.Process)) context.Context { - // A new context that will be "Done" when the process exits - // The caller can use it to monitor the process. - newCtx, cancelFunc := context.WithCancel(ctx) +func monitor(ctx context.Context, p *process.Process, f func(p *process.Process)) (context.Context, func()) { + // resultCtx is done when the process exits or monitoring is intentionally + // stopped. watchCtx lets callers suppress failure handling before an expected + // process termination, such as a hard power cycle. + resultCtx, cancelResult := context.WithCancel(ctx) + watchCtx, cancelWatch := context.WithCancel(ctx) + done := make(chan struct{}) go func() { ticker := time.NewTicker(3 * time.Second) + defer ticker.Stop() + defer cancelResult() + defer close(done) for { select { - case <-ctx.Done(): - cancelFunc() + case <-watchCtx.Done(): return case <-ticker.C: if !p.IsAlive() { code, err := p.ExitCode() - if err != nil || code != "0" { + if (err != nil || code != "0") && f != nil { f(p) } - cancelFunc() return } } } }() - return newCtx + var stopOnce sync.Once + stop := func() { + stopOnce.Do(func() { + cancelWatch() + <-done + }) + } + + return resultCtx, stop } // New returns a new machine. diff --git a/pkg/machine/qemu.go b/pkg/machine/qemu.go index 1169c7e..d485e8a 100644 --- a/pkg/machine/qemu.go +++ b/pkg/machine/qemu.go @@ -1,13 +1,16 @@ package machine import ( + "errors" "fmt" "net" "os" "os/exec" "path" "path/filepath" + "strconv" "strings" + "syscall" "time" "context" @@ -21,6 +24,8 @@ import ( type QEMU struct { machineConfig types.MachineConfig process *process.Process + lifecycleCtx context.Context + stopMonitor func() } // findQEMUBinary searches for qemu-system-x86_64 in common installation paths @@ -53,20 +58,32 @@ func findQEMUBinary(arch string) (string, error) { func (q *QEMU) Create(ctx context.Context) (context.Context, error) { log.Info("Create qemu machine") + q.lifecycleCtx = ctx - driveSizes := q.driveSizes() + userDrives, err := q.ensureDisks() + if err != nil { + return ctx, err + } + return q.launch(ctx, userDrives) +} + +// ensureDisks creates auto-managed disks if needed and returns the full drive list. +func (q *QEMU) ensureDisks() ([]string, error) { userDrives := q.machineConfig.Drives if q.machineConfig.AutoDriveSetup && len(userDrives) == 0 { - for i, s := range driveSizes { + for i, s := range q.driveSizes() { filename := fmt.Sprintf("%s-%d.img", q.machineConfig.ID, i) - err := q.CreateDisk(filename, s) - if err != nil { - return ctx, fmt.Errorf("creating disk with size %s: %w", s, err) + if err := q.CreateDisk(filename, s); err != nil { + return nil, fmt.Errorf("creating disk with size %s: %w", s, err) } userDrives = append(userDrives, filepath.Join(q.machineConfig.StateDir, filename)) } } + return userDrives, nil +} +// launch starts the QEMU process against the given drive paths without touching disk images. +func (q *QEMU) launch(ctx context.Context, userDrives []string) (context.Context, error) { genDrives := func(m types.MachineConfig) []string { var allDrives []string scsiAdded := false @@ -199,8 +216,24 @@ func (q *QEMU) Create(ctx context.Context) (context.Context, error) { q.process = qemu - newCtx := monitor(ctx, qemu, q.machineConfig.OnFailure) - return newCtx, qemu.Run() + if err := qemu.Run(); err != nil { + return ctx, err + } + + return q.startProcessMonitor(ctx), nil +} + +func (q *QEMU) startProcessMonitor(ctx context.Context) context.Context { + newCtx, stopMonitor := monitor(ctx, q.process, q.machineConfig.OnFailure) + q.stopMonitor = stopMonitor + return newCtx +} + +func (q *QEMU) stopProcessMonitor() { + if q.stopMonitor != nil { + q.stopMonitor() + q.stopMonitor = nil + } } func (q *QEMU) Config() types.MachineConfig { @@ -254,9 +287,110 @@ func (q *QEMU) Screenshot() (string, error) { } func (q *QEMU) Stop() error { + q.stopProcessMonitor() return process.New(process.WithStateDir(q.machineConfig.StateDir)).Stop() } +// HardPowerCycle simulates power loss: SIGKILLs qemu (no SIGTERM grace) and relaunches against existing disk state. +func (q *QEMU) HardPowerCycle(ctx context.Context) (context.Context, error) { + log.Info("Hard power-cycling QEMU VM (SIGKILL, no grace period)") + + launchCtx := q.lifecycleCtx + if launchCtx == nil { + launchCtx = ctx + } + if err := launchCtx.Err(); err != nil { + return ctx, fmt.Errorf("QEMU lifecycle context is done: %w", err) + } + if q.process == nil || !q.Alive() { + return ctx, fmt.Errorf("QEMU process is not running") + } + + // Synchronize with the old monitor before killing QEMU so the expected + // non-zero exit cannot be reported through OnFailure. + q.stopProcessMonitor() + if err := q.kill9(); err != nil { + if q.process != nil && q.Alive() { + ctx = q.startProcessMonitor(launchCtx) + } + return ctx, fmt.Errorf("failed to SIGKILL qemu process: %w", err) + } + if err := q.waitForProcessExit(5 * time.Second); err != nil { + return ctx, fmt.Errorf("qemu process did not exit: %w", err) + } + + // Stale state from the killed process must be gone before relaunch. + for _, stalePath := range []string{filepath.Join(q.machineConfig.StateDir, "pid"), q.monitorSockFile()} { + if err := removeIfExists(stalePath); err != nil { + return ctx, fmt.Errorf("removing stale QEMU state %q: %w", stalePath, err) + } + } + + // reuse existing disks; do not re-run AutoDriveSetup + userDrives := q.machineConfig.Drives + if q.machineConfig.AutoDriveSetup && len(userDrives) == 0 { + for i := range q.driveSizes() { + filename := fmt.Sprintf("%s-%d.img", q.machineConfig.ID, i) + userDrives = append(userDrives, filepath.Join(q.machineConfig.StateDir, filename)) + } + } + for _, drive := range userDrives { + if _, err := os.Stat(drive); err != nil { + return ctx, fmt.Errorf("QEMU drive %q is unavailable after power loss: %w", drive, err) + } + } + + return q.launch(launchCtx, userDrives) +} + +func removeIfExists(name string) error { + err := os.Remove(name) + if err != nil && !errors.Is(err, os.ErrNotExist) { + return err + } + return nil +} + +// kill9 reads {StateDir}/pid and sends SIGKILL directly, skipping the SIGTERM grace period. +func (q *QEMU) kill9() error { + pidfile := filepath.Join(q.machineConfig.StateDir, "pid") + pidBytes, err := os.ReadFile(pidfile) + if err != nil { + return fmt.Errorf("reading pidfile %s: %w", pidfile, err) + } + pid, err := strconv.Atoi(strings.TrimSpace(string(pidBytes))) + if err != nil { + return fmt.Errorf("parsing pid %q: %w", string(pidBytes), err) + } + if q.process != nil && q.process.PID != "" && strconv.Itoa(pid) != strings.TrimSpace(q.process.PID) { + return fmt.Errorf("pidfile identifies process %d, expected QEMU process %s", pid, q.process.PID) + } + proc, err := os.FindProcess(pid) + if err != nil { + return fmt.Errorf("finding pid %d: %w", pid, err) + } + if err := proc.Signal(syscall.SIGKILL); err != nil { + // already-dead is fine; post-condition is "process is dead" + if errors.Is(err, os.ErrProcessDone) { + return nil + } + return fmt.Errorf("sending SIGKILL to pid %d: %w", pid, err) + } + return nil +} + +// waitForProcessExit polls IsAlive until the process exits or timeout elapses. +func (q *QEMU) waitForProcessExit(timeout time.Duration) error { + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if !process.New(process.WithStateDir(q.machineConfig.StateDir)).IsAlive() { + return nil + } + time.Sleep(50 * time.Millisecond) + } + return fmt.Errorf("process still alive after %s", timeout) +} + func (q *QEMU) Clean() error { if q.machineConfig.StateDir != "" { return os.RemoveAll(q.machineConfig.StateDir) diff --git a/pkg/machine/qemu_internal_test.go b/pkg/machine/qemu_internal_test.go new file mode 100644 index 0000000..7b88d78 --- /dev/null +++ b/pkg/machine/qemu_internal_test.go @@ -0,0 +1,175 @@ +package machine + +import ( + "context" + "os" + "os/exec" + "path/filepath" + "strconv" + "sync/atomic" + "syscall" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + process "github.com/mudler/go-processmanager" + "github.com/spectrocloud/peg/pkg/machine/types" +) + +var _ = Describe("QEMU internal helpers", func() { + var ( + stateDir string + q *QEMU + ) + + spawnSleep := func() *exec.Cmd { + cmd := exec.Command("sleep", "30") + Expect(cmd.Start()).To(Succeed()) + pidfile := filepath.Join(stateDir, "pid") + Expect(os.WriteFile(pidfile, []byte(strconv.Itoa(cmd.Process.Pid)), 0o644)).To(Succeed()) + return cmd + } + + BeforeEach(func() { + stateDir = GinkgoT().TempDir() + q = &QEMU{machineConfig: types.MachineConfig{StateDir: stateDir}} + }) + + Describe("kill9", func() { + It("sends SIGKILL to the process named in the pidfile", func() { + cmd := spawnSleep() + Expect(q.kill9()).To(Succeed()) + + err := cmd.Wait() + Expect(err).To(HaveOccurred()) + exitErr, ok := err.(*exec.ExitError) + Expect(ok).To(BeTrue()) + ws, ok := exitErr.Sys().(syscall.WaitStatus) + Expect(ok).To(BeTrue()) + Expect(ws.Signaled()).To(BeTrue()) + Expect(ws.Signal()).To(Equal(syscall.SIGKILL)) + }) + + It("errors when the pidfile is missing", func() { + err := q.kill9() + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("reading pidfile")) + }) + + It("errors when the pid is unparseable", func() { + Expect(os.WriteFile(filepath.Join(stateDir, "pid"), []byte("not-a-pid"), 0o644)).To(Succeed()) + err := q.kill9() + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("parsing pid")) + }) + + It("is idempotent against an already-dead process", func() { + cmd := spawnSleep() + Expect(q.kill9()).To(Succeed()) + _ = cmd.Wait() + Expect(q.kill9()).To(Succeed()) + }) + }) + + Describe("waitForProcessExit", func() { + It("returns promptly when the process dies", func() { + cmd := spawnSleep() + go func() { + time.Sleep(100 * time.Millisecond) + _ = cmd.Process.Signal(syscall.SIGKILL) + _, _ = cmd.Process.Wait() + }() + + start := time.Now() + Expect(q.waitForProcessExit(2 * time.Second)).To(Succeed()) + Expect(time.Since(start)).To(BeNumerically("<", time.Second)) + }) + + It("times out when the process stays alive", func() { + cmd := spawnSleep() + DeferCleanup(func() { + _ = cmd.Process.Signal(syscall.SIGKILL) + _, _ = cmd.Process.Wait() + }) + + err := q.waitForProcessExit(200 * time.Millisecond) + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("still alive")) + }) + }) + + Describe("HardPowerCycle kill + cleanup phase", func() { + It("preserves disk files in StateDir and removes the pidfile", func() { + q.machineConfig.ID = "testvm" + cmd := spawnSleep() + + diskPath := filepath.Join(stateDir, "testvm-0.img") + diskContent := []byte("pretend qcow2 header and data") + Expect(os.WriteFile(diskPath, diskContent, 0o644)).To(Succeed()) + + Expect(q.kill9()).To(Succeed()) + _ = cmd.Wait() + Expect(q.waitForProcessExit(2 * time.Second)).To(Succeed()) + _ = os.Remove(filepath.Join(stateDir, "pid")) + + got, err := os.ReadFile(diskPath) + Expect(err).NotTo(HaveOccurred()) + Expect(got).To(Equal(diskContent)) + + _, err = os.Stat(filepath.Join(stateDir, "pid")) + Expect(os.IsNotExist(err)).To(BeTrue()) + }) + }) + + Describe("HardPowerCycle", func() { + It("relaunches QEMU with preserved disks without reporting an expected failure", func() { + stateDir = GinkgoT().TempDir() + fakeQEMU := filepath.Join(stateDir, "fake-qemu") + Expect(os.WriteFile(fakeQEMU, []byte("#!/bin/sh\nwhile :; do sleep 1; done\n"), 0o755)).To(Succeed()) + + diskPath := filepath.Join(stateDir, "disk.img") + diskContent := []byte("durable disk state") + Expect(os.WriteFile(diskPath, diskContent, 0o600)).To(Succeed()) + + var failureCalls atomic.Int32 + q = &QEMU{machineConfig: types.MachineConfig{ + StateDir: stateDir, + Process: fakeQEMU, + Memory: "128", + CPU: "1", + Arch: "x86_64", + Drives: []string{diskPath}, + OnFailure: func(_ *process.Process) { failureCalls.Add(1) }, + SSH: &types.SSH{}, + AutoDriveSetup: false, + }} + + ctx, cancel := context.WithCancel(context.Background()) + DeferCleanup(cancel) + processCtx, err := q.Create(ctx) + Expect(err).NotTo(HaveOccurred()) + DeferCleanup(func() { + if q.Alive() { + _ = q.Stop() + } + }) + + Eventually(q.Alive, 2*time.Second, 20*time.Millisecond).Should(BeTrue()) + firstPID, err := os.ReadFile(filepath.Join(stateDir, "pid")) + Expect(err).NotTo(HaveOccurred()) + + newProcessCtx, err := q.HardPowerCycle(processCtx) + Expect(err).NotTo(HaveOccurred()) + Expect(processCtx.Err()).To(Equal(context.Canceled)) + Expect(newProcessCtx.Err()).NotTo(HaveOccurred()) + Eventually(q.Alive, 2*time.Second, 20*time.Millisecond).Should(BeTrue()) + + secondPID, err := os.ReadFile(filepath.Join(stateDir, "pid")) + Expect(err).NotTo(HaveOccurred()) + Expect(secondPID).NotTo(Equal(firstPID)) + Expect(os.ReadFile(diskPath)).To(Equal(diskContent)) + Consistently(failureCalls.Load, 3500*time.Millisecond, 100*time.Millisecond).Should(BeZero()) + }) + }) +})