driver/qemu: Basic lifecycle management of Qemu images

client/driver/README.md

@@ -1,5 +0,0 @@
-// +build ignore
-
-# Drivers
-
-Drivers run things

client/driver/qemu.go

@@ -0,0 +1,272 @@
+package driver
+
+import (
+	"bytes"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"io"
+	"log"
+	"net/http"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"regexp"
+	"strings"
+	"time"
+
+	"github.com/hashicorp/nomad/client/config"
+	"github.com/hashicorp/nomad/nomad/structs"
+)
+
+var (
+	reQemuVersion = regexp.MustCompile("QEMU emulator version ([\\d\\.]+),.+")
+)
+
+// QemuDriver is a driver for running images via Qemu
+// We attempt to chose sane defaults for now, with more configuration available
+// planned in the future
+type QemuDriver struct {
+	logger *log.Logger
+}
+
+// qemuHandle is returned from Start/Open as a handle to the PID
+type qemuHandle struct {
+	proc   *os.Process
+	vmID   string
+	waitCh chan error
+	doneCh chan struct{}
+}
+
+// qemuPID is a struct to map the pid running the process to the vm image on
+// disk
+type qemuPID struct {
+	Pid  int
+	VmID string
+}
+
+// NewQemuDriver is used to create a new exec driver
+func NewQemuDriver(logger *log.Logger) Driver {
+	d := &QemuDriver{
+		logger: logger,
+	}
+	return d
+}
+
+func (d *QemuDriver) Fingerprint(cfg *config.Config, node *structs.Node) (bool, error) {
+	outBytes, err := exec.Command("qemu-system-x86_64", "-version").Output()
+	if err != nil {
+		return false, nil
+	}
+	out := strings.TrimSpace(string(outBytes))
+
+	matches := reQemuVersion.FindStringSubmatch(out)
+	if len(matches) != 2 {
+		return false, fmt.Errorf("Unable to parse Qemu version string: %#v", matches)
+	}
+
+	node.Attributes["driver.qemu"] = "true"
+	node.Attributes["driver.qemu.version"] = matches[1]
+
+	return true, nil
+}
+
+// Run an existing Qemu image. Start() will pull down an existing, valid Qemu
+// image and save it to the Drivers Allocation Dir
+func (d *QemuDriver) Start(ctx *ExecContext, task *structs.Task) (DriverHandle, error) {
+	// Get the image source
+	source, ok := task.Config["image_source"]
+	if !ok || source == "" {
+		return nil, fmt.Errorf("Missing source image Qemu driver")
+	}
+
+	// Attempt to download the thing
+	// Should be extracted to some kind of Http Fetcher
+	// Right now, assume publicly accessible HTTP url
+	resp, err := http.Get(source)
+	if err != nil {
+		return nil, fmt.Errorf("Error downloading source for Qemu driver: %s", err)
+	}
+
+	// Create a location in the AllocDir to download and store the image.
+	// TODO: Caching
+	vmID := fmt.Sprintf("qemu-vm-%s-%s", structs.GenerateUUID(), filepath.Base(source))
+	fPath := filepath.Join(ctx.AllocDir, vmID)
+	vmPath, err := os.OpenFile(fPath, os.O_CREATE|os.O_WRONLY, 0666)
+	if err != nil {
+		return nil, fmt.Errorf("Error opening file to download too: %s", err)
+	}
+	defer vmPath.Close()
+
+	// Copy remote file to local AllocDir for execution
+	// TODO: a retry of sort if io.Copy fails, for large binaries
+	_, ioErr := io.Copy(vmPath, resp.Body)
+	if ioErr != nil {
+		return nil, fmt.Errorf("Error copying jar from source: %s", ioErr)
+	}
+
+	// compute and check checksum
+	if check, ok := task.Config["checksum"]; ok {
+		d.logger.Printf("[DEBUG] Running checksum on (%s)", vmID)
+		hasher := sha256.New()
+		file, err := os.Open(vmPath.Name())
+		if err != nil {
+			return nil, fmt.Errorf("Failed to open file for checksum")
+		}
+
+		io.Copy(hasher, file)
+
+		sum := hex.EncodeToString(hasher.Sum(nil))
+		if sum != check {
+			return nil, fmt.Errorf(
+				"Error in Qemu: checksums did not match.\nExpected (%s), got (%s)",
+				check,
+				sum)
+		}
+	}
+
+	// Parse configuration arguments
+	// Create the base arguments
+	accelerator := "tcg"
+	mem := "512M"
+	if acc, ok := task.Config["accelerator"]; ok {
+		accelerator = acc
+	}
+	if m, ok := task.Config["memory"]; ok {
+		mem = m
+	}
+
+	args := []string{
+		"qemu-system-x86_64",
+		"-machine", "type=pc,accel=" + accelerator,
+		"-name", vmID,
+		"-m", mem,
+		"-drive", "file=" + vmPath.Name(),
+		"-nodefconfig",
+		"-nodefaults",
+		"-nographic",
+	}
+
+	// TODO: Consolidate these into map of host/guest port when we have HCL
+	// Note: Host port must be open and available
+	if task.Config["guest_port"] != "" && task.Config["host_port"] != "" {
+		args = append(args,
+			"-netdev",
+			fmt.Sprintf("user,id=user.0,hostfwd=tcp::%s-:%s",
+				task.Config["host_port"],
+				task.Config["guest_port"]),
+			"-device", "virtio-net,netdev=user.0",
+		)
+	}
+
+	// If using KVM, add optimization args
+	if accelerator == "kvm" {
+		args = append(args,
+			"-enable-kvm",
+			"-cpu", "host",
+			// Do we have cores information available to the Driver?
+			// "-smp", fmt.Sprintf("%d", cores),
+		)
+	}
+
+	// Start Qemu
+	var outBuf, errBuf bytes.Buffer
+	cmd := exec.Command(args[0], args[1:]...)
+	cmd.Stdout = &outBuf
+	cmd.Stderr = &errBuf
+
+	d.logger.Printf("[DEBUG] Starting QemuVM command: %q", strings.Join(args, " "))
+	if err := cmd.Start(); err != nil {
+		return nil, fmt.Errorf(
+			"Error running QEMU: %s\n\nOutput: %s\n\nError: %s",
+			err, outBuf.String(), errBuf.String())
+	}
+
+	d.logger.Printf("[INFO] Started new QemuVM: %s", vmID)
+
+	// Create and Return Handle
+	h := &qemuHandle{
+		proc:   cmd.Process,
+		vmID:   vmPath.Name(),
+		doneCh: make(chan struct{}),
+		waitCh: make(chan error, 1),
+	}
+
+	go h.run()
+	return h, nil
+}
+
+func (d *QemuDriver) Open(ctx *ExecContext, handleID string) (DriverHandle, error) {
+	// Parse the handle
+	pidBytes := []byte(strings.TrimPrefix(handleID, "QEMU:"))
+	qpid := &qemuPID{}
+	if err := json.Unmarshal(pidBytes, qpid); err != nil {
+		return nil, fmt.Errorf("failed to parse Qemu handle '%s': %v", handleID, err)
+	}
+
+	// Find the process
+	proc, err := os.FindProcess(qpid.Pid)
+	if proc == nil || err != nil {
+		return nil, fmt.Errorf("failed to find Qemu PID %d: %v", qpid.Pid, err)
+	}
+
+	// Return a driver handle
+	h := &qemuHandle{
+		proc:   proc,
+		vmID:   qpid.VmID,
+		doneCh: make(chan struct{}),
+		waitCh: make(chan error, 1),
+	}
+
+	go h.run()
+	return h, nil
+}
+
+func (h *qemuHandle) ID() string {
+	// Return a handle to the PID
+	pid := &qemuPID{
+		Pid:  h.proc.Pid,
+		VmID: h.vmID,
+	}
+	data, err := json.Marshal(pid)
+	if err != nil {
+		log.Printf("[ERROR] failed to marshal Qemu PID to JSON: %s", err)
+	}
+	return fmt.Sprintf("QEMU:%s", string(data))
+}
+
+func (h *qemuHandle) WaitCh() chan error {
+	return h.waitCh
+}
+
+func (h *qemuHandle) Update(task *structs.Task) error {
+	// Update is not possible
+	return nil
+}
+
+// Kill is used to terminate the task. We send an Interrupt
+// and then provide a 5 second grace period before doing a Kill.
+//
+// TODO: allow a 'shutdown_command' that can be executed over a ssh connection
+// to the VM
+func (h *qemuHandle) Kill() error {
+	h.proc.Signal(os.Interrupt)
+	select {
+	case <-h.doneCh:
+		return nil
+	case <-time.After(5 * time.Second):
+		return h.proc.Kill()
+	}
+}
+
+func (h *qemuHandle) run() {
+	ps, err := h.proc.Wait()
+	close(h.doneCh)
+	if err != nil {
+		h.waitCh <- err
+	} else if !ps.Success() {
+		h.waitCh <- fmt.Errorf("task exited with error")
+	}
+	close(h.waitCh)
+}

client/driver/qemu_test.go

@@ -0,0 +1,85 @@
+package driver
+
+import (
+	"fmt"
+	"os"
+	"testing"
+
+	"github.com/hashicorp/nomad/client/config"
+	"github.com/hashicorp/nomad/nomad/structs"
+)
+
+func TestQemuDriver_Handle(t *testing.T) {
+	h := &qemuHandle{
+		proc:   &os.Process{Pid: 123},
+		vmID:   "vmid",
+		doneCh: make(chan struct{}),
+		waitCh: make(chan error, 1),
+	}
+
+	actual := h.ID()
+	expected := `QEMU:{"Pid":123,"VmID":"vmid"}`
+	if actual != expected {
+		t.Errorf("Expected `%s`, found `%s`", expected, actual)
+	}
+}
+
+func TestQemuDriver_Fingerprint(t *testing.T) {
+	d := NewQemuDriver(testLogger())
+	node := &structs.Node{
+		Attributes: make(map[string]string),
+	}
+	apply, err := d.Fingerprint(&config.Config{}, node)
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	if !apply {
+		t.Fatalf("should apply")
+	}
+	if node.Attributes["driver.qemu"] == "" {
+		t.Fatalf("Missing Qemu driver")
+	}
+	if node.Attributes["driver.qemu.version"] == "" {
+		t.Fatalf("Missing Qemu driver version")
+	}
+}
+
+func TestQemuDriver_Start(t *testing.T) {
+	ctx := NewExecContext()
+	ctx.AllocDir = os.TempDir()
+	d := NewQemuDriver(testLogger())
+
+	// TODO: use test server to load from a fixture
+	task := &structs.Task{
+		Config: map[string]string{
+			"image_source": "https://dl.dropboxusercontent.com/u/47675/jar_thing/linux-0.2.img",
+			"memory":       "512M",
+			"accelerator":  "tcg",
+			"host_port":    "8080",
+			"guest_port":   "8081",
+			"checksum":     "a5e836985934c3392cbbd9b26db55a7d35a8d7ae1deb7ca559dd9c0159572544",
+			// ssh u/p would be here
+		},
+	}
+	handle, err := d.Start(ctx, task)
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	if handle == nil {
+		t.Fatalf("missing handle")
+	}
+
+	// Attempt to open
+	handle2, err := d.Open(ctx, handle.ID())
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	if handle2 == nil {
+		t.Fatalf("missing handle")
+	}
+
+	// Clean up
+	if err := handle.Kill(); err != nil {
+		fmt.Printf("\nError killing Qemu test: %s", err)
+	}
+}