feat: add QEMU hard power cycle support

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)

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())
+		})
+	})
+})

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")
+}

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.

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)