Merge pull request #3411 from cheeseprocedure/f-qemu-graceful-shutdown

Qemu driver: graceful shutdown feature
This commit is contained in:
Alex Dadgar 2017-11-03 16:41:34 -07:00 committed by GitHub
commit b3edc12dd9
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 385 additions and 38 deletions

View File

@ -3,8 +3,10 @@ package driver
import (
"context"
"encoding/json"
"errors"
"fmt"
"log"
"net"
"os"
"os/exec"
"path/filepath"
@ -13,7 +15,8 @@ import (
"strings"
"time"
"github.com/hashicorp/go-plugin"
"github.com/coreos/go-semver/semver"
plugin "github.com/hashicorp/go-plugin"
"github.com/hashicorp/nomad/client/config"
"github.com/hashicorp/nomad/client/driver/executor"
dstructs "github.com/hashicorp/nomad/client/driver/structs"
@ -26,12 +29,27 @@ import (
var (
reQemuVersion = regexp.MustCompile(`version (\d[\.\d+]+)`)
// Prior to qemu 2.10.1, monitor socket paths are truncated to 108 bytes.
// We should consider this if driver.qemu.version is < 2.10.1 and the
// generated monitor path is too long.
//
// Relevant fix is here:
// https://github.com/qemu/qemu/commit/ad9579aaa16d5b385922d49edac2c96c79bcfb6
qemuVersionLongSocketPathFix = semver.New("2.10.1")
)
const (
// The key populated in Node Attributes to indicate presence of the Qemu
// driver
// The key populated in Node Attributes to indicate presence of the Qemu driver
qemuDriverAttr = "driver.qemu"
qemuDriverVersionAttr = "driver.qemu.version"
// Represents an ACPI shutdown request to the VM (emulates pressing a physical power button)
// Reference: https://en.wikibooks.org/wiki/QEMU/Monitor
qemuGracefulShutdownMsg = "system_powerdown\n"
qemuMonitorSocketName = "qemu-monitor.sock"
// Maximum socket path length prior to qemu 2.10.1
qemuLegacyMaxMonitorPathLen = 108
)
// QemuDriver is a driver for running images via Qemu
@ -47,6 +65,7 @@ type QemuDriver struct {
type QemuDriverConfig struct {
ImagePath string `mapstructure:"image_path"`
Accelerator string `mapstructure:"accelerator"`
GracefulShutdown bool `mapstructure:"graceful_shutdown"`
PortMap []map[string]int `mapstructure:"port_map"` // A map of host port labels and to guest ports.
Args []string `mapstructure:"args"` // extra arguments to qemu executable
}
@ -56,6 +75,7 @@ type qemuHandle struct {
pluginClient *plugin.Client
userPid int
executor executor.Executor
monitorPath string
killTimeout time.Duration
maxKillTimeout time.Duration
logger *log.Logger
@ -64,6 +84,29 @@ type qemuHandle struct {
doneCh chan struct{}
}
// getMonitorPath is used to determine whether a qemu monitor socket can be
// safely created and accessed in the task directory by the version of qemu
// present on the host. If it is safe to use, the socket's full path is
// returned along with a nil error. Otherwise, an empty string is returned
// along with a descriptive error.
func (d *QemuDriver) getMonitorPath(dir string) (string, error) {
var longPathSupport bool
currentQemuVer := d.DriverContext.node.Attributes[qemuDriverVersionAttr]
currentQemuSemver := semver.New(currentQemuVer)
if currentQemuSemver.LessThan(*qemuVersionLongSocketPathFix) {
longPathSupport = false
d.logger.Printf("[DEBUG] driver.qemu: long socket paths are not available in this version of QEMU (%s)", currentQemuVer)
} else {
longPathSupport = true
d.logger.Printf("[DEBUG] driver.qemu: long socket paths available in this version of QEMU (%s)", currentQemuVer)
}
fullSocketPath := fmt.Sprintf("%s/%s", dir, qemuMonitorSocketName)
if len(fullSocketPath) > qemuLegacyMaxMonitorPathLen && longPathSupport == false {
return "", fmt.Errorf("monitor path is too long for this version of qemu")
}
return fullSocketPath, nil
}
// NewQemuDriver is used to create a new exec driver
func NewQemuDriver(ctx *DriverContext) Driver {
return &QemuDriver{DriverContext: *ctx}
@ -81,6 +124,10 @@ func (d *QemuDriver) Validate(config map[string]interface{}) error {
"accelerator": {
Type: fields.TypeString,
},
"graceful_shutdown": {
Type: fields.TypeBool,
Required: false,
},
"port_map": {
Type: fields.TypeArray,
},
@ -127,9 +174,11 @@ func (d *QemuDriver) Fingerprint(cfg *config.Config, node *structs.Node) (bool,
delete(node.Attributes, qemuDriverAttr)
return false, fmt.Errorf("Unable to parse Qemu version string: %#v", matches)
}
currentQemuVersion := matches[1]
node.Attributes[qemuDriverAttr] = "1"
node.Attributes["driver.qemu.version"] = matches[1]
node.Attributes[qemuDriverVersionAttr] = currentQemuVersion
return true, nil
}
@ -190,6 +239,22 @@ func (d *QemuDriver) Start(ctx *ExecContext, task *structs.Task) (*StartResponse
"-nographic",
}
var monitorPath string
if d.driverConfig.GracefulShutdown {
if runtime.GOOS == "windows" {
return nil, errors.New("QEMU graceful shutdown is unsupported on the Windows platform")
}
// This socket will be used to manage the virtual machine (for example,
// to perform graceful shutdowns)
monitorPath, err := d.getMonitorPath(ctx.TaskDir.Dir)
if err != nil {
d.logger.Printf("[ERR] driver.qemu: could not get qemu monitor path: %s", err)
return nil, err
}
d.logger.Printf("[DEBUG] driver.qemu: got monitor path OK: %s", monitorPath)
args = append(args, "-monitor", fmt.Sprintf("unix:%s,server,nowait", monitorPath))
}
// Add pass through arguments to qemu executable. A user can specify
// these arguments in driver task configuration. These arguments are
// passed directly to the qemu driver as command line options.
@ -231,6 +296,9 @@ func (d *QemuDriver) Start(ctx *ExecContext, task *structs.Task) (*StartResponse
// If using KVM, add optimization args
if accelerator == "kvm" {
if runtime.GOOS == "windows" {
return nil, errors.New("KVM accelerator is unsupported on the Windows platform")
}
args = append(args,
"-enable-kvm",
"-cpu", "host",
@ -239,7 +307,7 @@ func (d *QemuDriver) Start(ctx *ExecContext, task *structs.Task) (*StartResponse
)
}
d.logger.Printf("[DEBUG] Starting QemuVM command: %q", strings.Join(args, " "))
d.logger.Printf("[DEBUG] driver.qemu: starting QemuVM command: %q", strings.Join(args, " "))
pluginLogFile := filepath.Join(ctx.TaskDir.Dir, "executor.out")
executorConfig := &dstructs.ExecutorConfig{
LogFile: pluginLogFile,
@ -272,7 +340,7 @@ func (d *QemuDriver) Start(ctx *ExecContext, task *structs.Task) (*StartResponse
pluginClient.Kill()
return nil, err
}
d.logger.Printf("[INFO] Started new QemuVM: %s", vmID)
d.logger.Printf("[INFO] driver.qemu: started new QemuVM: %s", vmID)
// Create and Return Handle
maxKill := d.DriverContext.config.MaxKillTimeout
@ -282,6 +350,7 @@ func (d *QemuDriver) Start(ctx *ExecContext, task *structs.Task) (*StartResponse
userPid: ps.Pid,
killTimeout: GetKillTimeout(task.KillTimeout, maxKill),
maxKillTimeout: maxKill,
monitorPath: monitorPath,
version: d.config.Version.VersionNumber(),
logger: d.logger,
doneCh: make(chan struct{}),
@ -308,7 +377,7 @@ type qemuId struct {
func (d *QemuDriver) Open(ctx *ExecContext, handleID string) (DriverHandle, error) {
id := &qemuId{}
if err := json.Unmarshal([]byte(handleID), id); err != nil {
return nil, fmt.Errorf("Failed to parse handle '%s': %v", handleID, err)
return nil, fmt.Errorf("Failed to parse handle %q: %v", handleID, err)
}
pluginConfig := &plugin.ClientConfig{
@ -317,9 +386,9 @@ func (d *QemuDriver) Open(ctx *ExecContext, handleID string) (DriverHandle, erro
exec, pluginClient, err := createExecutorWithConfig(pluginConfig, d.config.LogOutput)
if err != nil {
d.logger.Println("[ERR] driver.qemu: error connecting to plugin so destroying plugin pid and user pid")
d.logger.Printf("[ERR] driver.qemu: error connecting to plugin so destroying plugin pid %d and user pid %d", id.PluginConfig.Pid, id.UserPid)
if e := destroyPlugin(id.PluginConfig.Pid, id.UserPid); e != nil {
d.logger.Printf("[ERR] driver.qemu: error destroying plugin and userpid: %v", e)
d.logger.Printf("[ERR] driver.qemu: error destroying plugin pid %d and userpid %d: %v", id.PluginConfig.Pid, id.UserPid, e)
}
return nil, fmt.Errorf("error connecting to plugin: %v", err)
}
@ -381,27 +450,43 @@ func (h *qemuHandle) Signal(s os.Signal) error {
return fmt.Errorf("Qemu driver can't send signals")
}
// TODO: allow a 'shutdown_command' that can be executed over a ssh connection
// to the VM
func (h *qemuHandle) Kill() error {
gracefulShutdownSent := false
// Attempt a graceful shutdown only if it was configured in the job
if h.monitorPath != "" {
if err := sendQemuShutdown(h.logger, h.monitorPath, h.userPid); err == nil {
gracefulShutdownSent = true
} else {
h.logger.Printf("[DEBUG] driver.qemu: error sending graceful shutdown for user process pid %d: %s", h.userPid, err)
}
}
// If Nomad did not send a graceful shutdown signal, issue an interrupt to
// the qemu process as a last resort
if gracefulShutdownSent == false {
if err := h.executor.ShutDown(); err != nil {
if h.pluginClient.Exited() {
return nil
}
return fmt.Errorf("executor Shutdown failed: %v", err)
}
}
// If the qemu process exits before the kill timeout is reached, doneChan
// will close and we'll exit without an error. If it takes too long, the
// timer will fire and we'll attempt to kill the process.
select {
case <-h.doneCh:
return nil
case <-time.After(h.killTimeout):
h.logger.Printf("[DEBUG] driver.qemu: kill timeout of %s exceeded for user process pid %d", h.killTimeout.String(), h.userPid)
if h.pluginClient.Exited() {
return nil
}
if err := h.executor.Exit(); err != nil {
return fmt.Errorf("executor Exit failed: %v", err)
}
return nil
}
}
@ -414,7 +499,7 @@ func (h *qemuHandle) run() {
ps, werr := h.executor.Wait()
if ps.ExitCode == 0 && werr != nil {
if e := killProcess(h.userPid); e != nil {
h.logger.Printf("[ERR] driver.qemu: error killing user process: %v", e)
h.logger.Printf("[ERR] driver.qemu: error killing user process pid %d: %v", h.userPid, e)
}
}
close(h.doneCh)
@ -427,3 +512,23 @@ func (h *qemuHandle) run() {
h.waitCh <- &dstructs.WaitResult{ExitCode: ps.ExitCode, Signal: ps.Signal, Err: werr}
close(h.waitCh)
}
// sendQemuShutdown attempts to issue an ACPI power-off command via the qemu
// monitor
func sendQemuShutdown(logger *log.Logger, monitorPath string, userPid int) error {
if monitorPath == "" {
return errors.New("monitorPath not set")
}
monitorSocket, err := net.Dial("unix", monitorPath)
if err != nil {
logger.Printf("[WARN] driver.qemu: could not connect to qemu monitor %q for user process pid %d: %s", monitorPath, userPid, err)
return err
}
defer monitorSocket.Close()
logger.Printf("[DEBUG] driver.qemu: sending graceful shutdown command to qemu monitor socket %q for user process pid %d", monitorPath, userPid)
_, err = monitorSocket.Write([]byte(qemuGracefulShutdownMsg))
if err != nil {
logger.Printf("[WARN] driver.qemu: failed to send shutdown message %q to monitor socket %q for user process pid %d: %s", qemuGracefulShutdownMsg, monitorPath, userPid, err)
}
return err
}

View File

@ -2,10 +2,12 @@ package driver
import (
"fmt"
"os"
"path/filepath"
"strings"
"syscall"
"testing"
"time"
"github.com/hashicorp/nomad/client/config"
"github.com/hashicorp/nomad/nomad/structs"
@ -39,15 +41,16 @@ func TestQemuDriver_Fingerprint(t *testing.T) {
if !apply {
t.Fatalf("should apply")
}
if node.Attributes["driver.qemu"] == "" {
if node.Attributes[qemuDriverAttr] == "" {
t.Fatalf("Missing Qemu driver")
}
if node.Attributes["driver.qemu.version"] == "" {
if node.Attributes[qemuDriverVersionAttr] == "" {
t.Fatalf("Missing Qemu driver version")
}
}
func TestQemuDriver_StartOpen_Wait(t *testing.T) {
logger := testLogger()
if !testutil.IsTravis() {
t.Parallel()
}
@ -58,6 +61,7 @@ func TestQemuDriver_StartOpen_Wait(t *testing.T) {
Config: map[string]interface{}{
"image_path": "linux-0.2.img",
"accelerator": "tcg",
"graceful_shutdown": false,
"port_map": []map[string]int{{
"main": 22,
"web": 8080,
@ -85,10 +89,11 @@ func TestQemuDriver_StartOpen_Wait(t *testing.T) {
// Copy the test image into the task's directory
dst := ctx.ExecCtx.TaskDir.Dir
copyFile("./test-resources/qemu/linux-0.2.img", filepath.Join(dst, "linux-0.2.img"), t)
if _, err := d.Prestart(ctx.ExecCtx, task); err != nil {
t.Fatalf("Prestart faild: %v", err)
t.Fatalf("Prestart failed: %v", err)
}
resp, err := d.Start(ctx.ExecCtx, task)
@ -112,7 +117,108 @@ func TestQemuDriver_StartOpen_Wait(t *testing.T) {
// Clean up
if err := resp.Handle.Kill(); err != nil {
fmt.Printf("\nError killing Qemu test: %s", err)
logger.Printf("Error killing Qemu test: %s", err)
}
}
func TestQemuDriver_GracefulShutdown(t *testing.T) {
logger := testLogger()
if !testutil.IsTravis() {
t.Parallel()
}
ctestutils.QemuCompatible(t)
task := &structs.Task{
Name: "linux",
Driver: "qemu",
Config: map[string]interface{}{
"image_path": "linux-0.2.img",
"accelerator": "tcg",
"graceful_shutdown": true,
"port_map": []map[string]int{{
"main": 22,
"web": 8080,
}},
"args": []string{"-nodefconfig", "-nodefaults"},
},
// With the use of tcg acceleration, it's very unlikely a qemu instance
// will boot (and gracefully halt) in a reasonable amount of time, so
// this timeout is kept low to reduce test execution time.
KillTimeout: time.Duration(1 * time.Second),
LogConfig: &structs.LogConfig{
MaxFiles: 10,
MaxFileSizeMB: 10,
},
Resources: &structs.Resources{
CPU: 500,
MemoryMB: 512,
Networks: []*structs.NetworkResource{
{
ReservedPorts: []structs.Port{{Label: "main", Value: 22000}, {Label: "web", Value: 80}},
},
},
},
}
ctx := testDriverContexts(t, task)
defer ctx.AllocDir.Destroy()
d := NewQemuDriver(ctx.DriverCtx)
apply, err := d.Fingerprint(&config.Config{}, ctx.DriverCtx.node)
if err != nil {
t.Fatalf("err: %v", err)
}
if !apply {
t.Fatalf("should apply")
}
dst := ctx.ExecCtx.TaskDir.Dir
copyFile("./test-resources/qemu/linux-0.2.img", filepath.Join(dst, "linux-0.2.img"), t)
if _, err := d.Prestart(ctx.ExecCtx, task); err != nil {
t.Fatalf("Prestart failed: %v", err)
}
resp, err := d.Start(ctx.ExecCtx, task)
if err != nil {
t.Fatalf("err: %v", err)
}
// Clean up
defer func() {
if err := resp.Handle.Kill(); err != nil {
logger.Printf("Error killing Qemu test: %s", err)
}
}()
// The monitor socket will not exist immediately, so we'll wait up to
// 5 seconds for it to become available.
monitorPath := fmt.Sprintf("%s/linux/%s", ctx.AllocDir.AllocDir, qemuMonitorSocketName)
monitorPathExists := false
for i := 0; i < 100; i++ {
if _, err := os.Stat(monitorPath); !os.IsNotExist(err) {
logger.Printf("monitor socket exists at %q\n", monitorPath)
monitorPathExists = true
break
}
time.Sleep(200 * time.Millisecond)
}
if monitorPathExists == false {
t.Fatalf("monitor socket did not exist after waiting 20 seconds")
}
// userPid supplied in sendQemuShutdown calls is bogus (it's used only
// for log output)
if err := sendQemuShutdown(ctx.DriverCtx.logger, "", 0); err == nil {
t.Fatalf("sendQemuShutdown should return an error if monitorPath parameter is empty")
}
if err := sendQemuShutdown(ctx.DriverCtx.logger, "/path/that/does/not/exist", 0); err == nil {
t.Fatalf("sendQemuShutdown should return an error if file does not exist at monitorPath")
}
if err := sendQemuShutdown(ctx.DriverCtx.logger, monitorPath, 0); err != nil {
t.Fatalf("unexpected error from sendQemuShutdown: %s", err)
}
}
@ -121,13 +227,15 @@ func TestQemuDriverUser(t *testing.T) {
t.Parallel()
}
ctestutils.QemuCompatible(t)
tasks := []*structs.Task{{
tasks := []*structs.Task{
{
Name: "linux",
Driver: "qemu",
User: "alice",
Config: map[string]interface{}{
"image_path": "linux-0.2.img",
"accelerator": "tcg",
"graceful_shutdown": false,
"port_map": []map[string]int{{
"main": 22,
"web": 8080,
@ -193,9 +301,127 @@ func TestQemuDriverUser(t *testing.T) {
resp.Handle.Kill()
t.Fatalf("Should've failed")
}
msg := task.Config["msg"].(string)
if !strings.Contains(err.Error(), msg) {
t.Fatalf("Expecting '%v' in '%v'", msg, err)
}
}
}
func TestQemuDriverGetMonitorPathOldQemu(t *testing.T) {
task := &structs.Task{
Name: "linux",
Driver: "qemu",
Config: map[string]interface{}{
"image_path": "linux-0.2.img",
"accelerator": "tcg",
"graceful_shutdown": true,
"port_map": []map[string]int{{
"main": 22,
"web": 8080,
}},
"args": []string{"-nodefconfig", "-nodefaults"},
},
KillTimeout: time.Duration(1 * time.Second),
LogConfig: &structs.LogConfig{
MaxFiles: 10,
MaxFileSizeMB: 10,
},
Resources: &structs.Resources{
CPU: 500,
MemoryMB: 512,
Networks: []*structs.NetworkResource{
{
ReservedPorts: []structs.Port{{Label: "main", Value: 22000}, {Label: "web", Value: 80}},
},
},
},
}
ctx := testDriverContexts(t, task)
defer ctx.AllocDir.Destroy()
// Simulate an older version of qemu which does not support long monitor socket paths
ctx.DriverCtx.node.Attributes[qemuDriverVersionAttr] = "2.0.0"
d := &QemuDriver{DriverContext: *ctx.DriverCtx}
shortPath := strings.Repeat("x", 10)
_, err := d.getMonitorPath(shortPath)
if err != nil {
t.Fatal("Should not have returned an error")
}
longPath := strings.Repeat("x", qemuLegacyMaxMonitorPathLen+100)
_, err = d.getMonitorPath(longPath)
if err == nil {
t.Fatal("Should have returned an error")
}
// Max length includes the '/' separator and socket name
maxLengthCount := qemuLegacyMaxMonitorPathLen - len(qemuMonitorSocketName) - 1
maxLengthLegacyPath := strings.Repeat("x", maxLengthCount)
_, err = d.getMonitorPath(maxLengthLegacyPath)
if err != nil {
t.Fatalf("Should not have returned an error: %s", err)
}
}
func TestQemuDriverGetMonitorPathNewQemu(t *testing.T) {
task := &structs.Task{
Name: "linux",
Driver: "qemu",
Config: map[string]interface{}{
"image_path": "linux-0.2.img",
"accelerator": "tcg",
"graceful_shutdown": true,
"port_map": []map[string]int{{
"main": 22,
"web": 8080,
}},
"args": []string{"-nodefconfig", "-nodefaults"},
},
KillTimeout: time.Duration(1 * time.Second),
LogConfig: &structs.LogConfig{
MaxFiles: 10,
MaxFileSizeMB: 10,
},
Resources: &structs.Resources{
CPU: 500,
MemoryMB: 512,
Networks: []*structs.NetworkResource{
{
ReservedPorts: []structs.Port{{Label: "main", Value: 22000}, {Label: "web", Value: 80}},
},
},
},
}
ctx := testDriverContexts(t, task)
defer ctx.AllocDir.Destroy()
// Simulate a version of qemu which supports long monitor socket paths
ctx.DriverCtx.node.Attributes[qemuDriverVersionAttr] = "2.99.99"
d := &QemuDriver{DriverContext: *ctx.DriverCtx}
shortPath := strings.Repeat("x", 10)
_, err := d.getMonitorPath(shortPath)
if err != nil {
t.Fatal("Should not have returned an error")
}
longPath := strings.Repeat("x", qemuLegacyMaxMonitorPathLen+100)
_, err = d.getMonitorPath(longPath)
if err != nil {
t.Fatal("Should not have returned an error")
}
maxLengthCount := qemuLegacyMaxMonitorPathLen - len(qemuMonitorSocketName) - 1
maxLengthLegacyPath := strings.Repeat("x", maxLengthCount)
_, err = d.getMonitorPath(maxLengthLegacyPath)
if err != nil {
t.Fatal("Should not have returned an error")
}
}

View File

@ -31,6 +31,7 @@ task "webservice" {
config {
image_path = "/path/to/my/linux.img"
accelerator = "kvm"
graceful_shutdown = true
args = ["-nodefaults", "-nodefconfig"]
}
}
@ -47,6 +48,21 @@ The `qemu` driver supports the following configuration in the job spec:
If the host machine has `qemu` installed with KVM support, users can specify
`kvm` for the `accelerator`. Default is `tcg`.
* `graceful_shutdown` `(bool: false)` - Using the [qemu
monitor](https://en.wikibooks.org/wiki/QEMU/Monitor), send an ACPI shutdown
signal to virtual machines rather than simply terminating them. This emulates
a physical power button press, and gives instances a chance to shut down
cleanly. If the VM is still running after ``kill_timeout``, it will be
forcefully terminated. (Note that
[prior to qemu 2.10.1](https://github.com/qemu/qemu/commit/ad9579aaa16d5b385922d49edac2c96c79bcfb6),
the monitor socket path is limited to 108 characters. Graceful shutdown will
be disabled if qemu is < 2.10.1 and the generated monitor path exceeds this
length. You may encounter this issue if you set long
[data_dir](https://www.nomadproject.io/docs/agent/configuration/index.html#data_dir)
or
[alloc_dir](https://www.nomadproject.io/docs/agent/configuration/client.html#alloc_dir)
paths.)
* `port_map` - (Optional) A key-value map of port labels.
```hcl