NOMAD_tuned.patch

From 8a79b3c6ee2e854f20b1469ebea99c4bc55ea33d Mon Sep 17 00:00:00 2001
From: Pier Luigi Ventre <pierventre@hotmail.com>
Date: Tue, 17 May 2016 17:11:39 +0200
Subject: [PATCH 1/2] Changes: - Adds XenTunedDriver to list of driver -
 Implements XenTunedDriver

---
 client/driver/driver.go    |  15 +-
 client/driver/xen_tuned.go | 528 +++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 536 insertions(+), 7 deletions(-)
 create mode 100644 client/driver/xen_tuned.go

diff --git a/client/driver/driver.go b/client/driver/driver.go
index 88e54da..d316cd7 100644
--- a/client/driver/driver.go
+++ b/client/driver/driver.go
@@ -19,13 +19,14 @@ import (
 // BuiltinDrivers contains the built in registered drivers
 // which are available for allocation handling
 var BuiltinDrivers = map[string]Factory{
-	"docker":   NewDockerDriver,
-	"exec":     NewExecDriver,
-	"raw_exec": NewRawExecDriver,
-	"java":     NewJavaDriver,
-	"qemu":     NewQemuDriver,
-	"rkt":      NewRktDriver,
-	"xen":      NewXenDriver,
+	"docker":   	NewDockerDriver,
+	"exec":     	NewExecDriver,
+	"raw_exec": 	NewRawExecDriver,
+	"java":     	NewJavaDriver,
+	"qemu":     	NewQemuDriver,
+	"rkt":      	NewRktDriver,
+	"xen":      	NewXenDriver,
+	"xen_tuned":	NewXenTunedDriver,
 }
 
 // NewDriver is used to instantiate and return a new driver
diff --git a/client/driver/xen_tuned.go b/client/driver/xen_tuned.go
new file mode 100644
index 0000000..6642cc5
--- /dev/null
+++ b/client/driver/xen_tuned.go
@@ -0,0 +1,528 @@
+package driver
+
+import (
+	"encoding/json"
+	"fmt"
+	"log"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+	"time"
+	"io"
+
+	"github.com/hashicorp/go-plugin"
+	"github.com/hashicorp/nomad/client/allocdir"
+	"github.com/hashicorp/nomad/client/config"
+	"github.com/hashicorp/nomad/client/driver/executor"
+	cstructs "github.com/hashicorp/nomad/client/driver/structs"
+	"github.com/hashicorp/nomad/client/fingerprint"
+	"github.com/hashicorp/nomad/nomad/structs"
+	"github.com/mitchellh/mapstructure"
+)
+
+// XenDriver is a driver for running Xen.
+type XenTunedDriver struct {
+	DriverContext
+	fingerprint.StaticFingerprinter
+}
+
+// Configuration for XenDriver
+type XenTunedDriverConfig struct {
+	Accelerator string           `mapstructure:"accelerator"`
+	PortMap     []map[string]int `mapstructure:"port_map"` // A map of host port labels and to guest ports.
+}
+
+// old xenHandle is returned from Start/Open as a handle to the PID (identical to qemu and java).
+
+/*
+type xenHandle struct {
+	cmd    executor.Executor
+	waitCh chan *cstructs.WaitResult
+	doneCh chan struct{}
+}
+*/
+
+// xenHandle is returned from Start/Open as a handle to the PID (identical to qemu)
+// TODO verify if it is ok
+type xenTunedHandle struct {
+	pluginClient  *plugin.Client
+	userPid       int
+	executor      executor.Executor
+	allocDir      *allocdir.AllocDir
+	killTimeout   time.Duration
+	logger        *log.Logger
+	waitCh        chan *cstructs.WaitResult
+	doneCh        chan struct{}
+	get_config    time.Duration
+	alloc_dir     time.Duration
+	down_artifact time.Duration
+	init_env      time.Duration
+}
+
+// NewXenDriver is used to create a new exec driver (identical to qemu and java).
+func NewXenTunedDriver(ctx *DriverContext) Driver {
+	return &XenTunedDriver{DriverContext: *ctx}
+}
+
+// Return the driver to be used
+func (d *XenTunedDriver) Fingerprint(cfg *config.Config, node *structs.Node) (bool, error) {
+
+	bin := "xl"
+	outBytes, err := exec.Command(bin, "info").Output()
+	if err != nil {
+		return false, nil
+	}
+	out := strings.TrimSpace(string(outBytes))
+	matches1 := reMajVersion.FindStringSubmatch(out)
+	if len(matches1) != 2 {
+		return false, fmt.Errorf("Unable to parse Xen major version string: %#v", matches1)
+	}
+	matches2 := reMinVersion.FindStringSubmatch(out)
+	if len(matches2) != 2 {
+		return false, fmt.Errorf("Unable to parse Xen minor version string: %#v", matches2)
+	}
+	matches3 := reExtVersion.FindStringSubmatch(out)
+	if len(matches3) != 2 {
+		return false, fmt.Errorf("Unable to parse Xen extra version string: %#v", matches3)
+	}
+
+	matches := matches1[1] + "." + matches2[1] + "." + matches3[1]
+
+	node.Attributes["driver.xen"] = "1"
+	node.Attributes["driver.xen.version"] = matches
+
+	return true, nil
+}
+
+/*
+	- TODO
+	- Complete Driver;
+	- cfg in the code;
+	- different toolstack
+
+*/
+
+// Run an existing Xen image. Start() will pull down an existing, valid Xen
+// image and save it to the Drivers Allocation Dir
+func (d *XenTunedDriver) Start(ctx *ExecContext, task *structs.Task) (DriverHandle, error) {
+
+	start_config := time.Now()
+
+	var driverConfig XenTunedDriverConfig
+	if err := mapstructure.WeakDecode(task.Config, &driverConfig); err != nil {
+		return nil, err
+	}
+
+	/*
+		if len(driverConfig.PortMap) > 1 {
+			return nil, fmt.Errorf("Only one port_map block is allowed in the qemu driver config")
+		}
+	*/
+
+	end_config := time.Now()
+
+	// Qemu defaults to 128M of RAM for a given VM. Instead, we force users to
+	// supply a memory size in the tasks resources
+	/*
+		if task.Resources == nil || task.Resources.MemoryMB == 0 {
+			return nil, fmt.Errorf("Missing required Task Resource: Memory")
+		}
+	*/
+
+	start_allocDir := time.Now()
+
+	// Get the tasks local directory.
+	taskDir, ok := ctx.AllocDir.TaskDirs[d.DriverContext.taskName]
+	if !ok {
+		return nil, fmt.Errorf("Could not find task directory for task: %v", d.DriverContext.taskName)
+	}
+
+	end_allocDir := time.Now()
+
+	start_downArtifact := time.Now()
+
+	read, err := os.Open("/root/clickos.cfg")
+	if err != nil {
+		panic(err)
+	}
+	defer read.Close()
+
+	artifactFile := filepath.Join(taskDir, "clickos.cfg")
+
+	write, err := os.Create(artifactFile)
+	if err != nil {
+		panic(err)
+	}
+	defer write.Close()
+
+	_ , err = io.Copy(write, read)
+	if err != nil {
+		panic(err)
+	}
+
+	read, err = os.Open("/root/clickos_x86_64")
+	if err != nil {
+		panic(err)
+	}
+	defer read.Close()
+
+	artifactFile = filepath.Join(taskDir, "clickos_x86_64")
+
+	write, err = os.Create(artifactFile)
+	if err != nil {
+		panic(err)
+	}
+	defer write.Close()
+
+	_ , err = io.Copy(write, read)
+	if err != nil {
+		panic(err)
+	}
+
+	end_downArtificat := time.Now()
+
+	// Parse configuration arguments
+	// Create the base arguments
+	/*
+		accelerator := "tcg"
+		if driverConfig.Accelerator != "" {
+			accelerator = driverConfig.Accelerator
+		}
+		// TODO: Check a lower bounds, e.g. the default 128 of Qemu
+		mem := fmt.Sprintf("%dM", task.Resources.MemoryMB)
+	*/
+
+	start_init_env := time.Now()
+
+	args := []string{
+		"xl",
+		"create",
+		"clickos.cfg",
+	}
+
+	/*
+		// Check the Resources required Networks to add port mappings. If no resources
+		// are required, we assume the VM is a purely compute job and does not require
+		// the outside world to be able to reach it. VMs ran without port mappings can
+		// still reach out to the world, but without port mappings it is effectively
+		// firewalled
+		protocols := []string{"udp", "tcp"}
+		if len(task.Resources.Networks) > 0 && len(driverConfig.PortMap) == 1 {
+			// Loop through the port map and construct the hostfwd string, to map
+			// reserved ports to the ports listenting in the VM
+			// Ex: hostfwd=tcp::22000-:22,hostfwd=tcp::80-:8080
+			var forwarding []string
+			taskPorts := task.Resources.Networks[0].MapLabelToValues(nil)
+			for label, guest := range driverConfig.PortMap[0] {
+				host, ok := taskPorts[label]
+				if !ok {
+					return nil, fmt.Errorf("Unknown port label %q", label)
+				}
+
+				for _, p := range protocols {
+					forwarding = append(forwarding, fmt.Sprintf("hostfwd=%s::%d-:%d", p, host, guest))
+				}
+			}
+
+			if len(forwarding) != 0 {
+				args = append(args,
+					"-netdev",
+					fmt.Sprintf("user,id=user.0,%s", strings.Join(forwarding, ",")),
+					"-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),
+			)
+		}*/
+
+	//d.logger.Printf("[DEBUG] Starting xenVM command: %q", strings.Join(args, " "))
+	bin := "/root/goprojects/bin/nomad"
+
+	pluginLogFile := filepath.Join(taskDir, fmt.Sprintf("%s-executor.out", task.Name))
+	pluginConfig := &plugin.ClientConfig{
+		Cmd: exec.Command(bin, "executor", pluginLogFile),
+	}
+
+	exec, pluginClient, err := createExecutor(pluginConfig, d.config.LogOutput, d.config)
+	if err != nil {
+		return nil, err
+	}
+	executorCtx := &executor.ExecutorContext{
+		TaskEnv:       d.taskEnv,
+		AllocDir:      ctx.AllocDir,
+		TaskName:      task.Name,
+		TaskResources: task.Resources,
+		LogConfig:     task.LogConfig,
+	}
+	ps, err := exec.LaunchCmd(&executor.ExecCommand{Cmd: args[0], Args: args[1:]}, executorCtx)
+	if err != nil {
+		pluginClient.Kill()
+		return nil, fmt.Errorf("error starting process via the plugin: %v", err)
+	}
+	//d.logger.Printf("[DEBUG] Started new XenVM %s using %s\n", vmID, cfgID)
+
+	// Create and Return Handle
+	h := &xenHandle{
+		pluginClient: pluginClient,
+		executor:     exec,
+		userPid:      ps.Pid,
+		allocDir:     ctx.AllocDir,
+		killTimeout:  d.DriverContext.KillTimeout(task),
+		logger:       d.logger,
+		doneCh:       make(chan struct{}),
+		waitCh:       make(chan *cstructs.WaitResult, 1),
+	}
+
+	end_init_env := time.Now()
+
+	h.get_config = end_config.Sub(start_config)
+	h.alloc_dir = end_allocDir.Sub(start_allocDir)
+	h.down_artifact = end_downArtificat.Sub(start_downArtifact)
+	h.init_env = end_init_env.Sub(start_init_env)
+
+	go h.run()
+	return h, nil
+}
+
+type xenTunedId struct {
+	KillTimeout  time.Duration
+	UserPid      int
+	PluginConfig *PluginReattachConfig
+	AllocDir     *allocdir.AllocDir
+}
+
+func (d *XenTunedDriver) Open(ctx *ExecContext, handleID string) (DriverHandle, error) {
+	/*
+		Old code
+		// Find the process
+		cmd, err := executor.OpenId(handleID)
+		if err != nil {
+			return nil, fmt.Errorf("failed to open ID %v: %v", handleID, err)
+		}
+
+		// Return a driver handle
+		h := &execHandle{
+			cmd:    cmd,
+			doneCh: make(chan struct{}),
+			waitCh: make(chan *cstructs.WaitResult, 1),
+		}
+		go h.run()
+		return h, nil
+	*/
+
+	id := &xenTunedId{}
+	if err := json.Unmarshal([]byte(handleID), id); err != nil {
+		return nil, fmt.Errorf("Failed to parse handle '%s': %v", handleID, err)
+	}
+
+	pluginConfig := &plugin.ClientConfig{
+		Reattach: id.PluginConfig.PluginConfig(),
+	}
+
+	executor, pluginClient, err := createExecutor(pluginConfig, d.config.LogOutput, d.config)
+	if err != nil {
+		d.logger.Println("[ERROR] driver.xen: error connecting to plugin so destroying plugin pid and user pid")
+		if e := destroyPlugin(id.PluginConfig.Pid, id.UserPid); e != nil {
+			d.logger.Printf("[ERROR] driver.xen: error destroying plugin and userpid: %v", e)
+		}
+		return nil, fmt.Errorf("error connecting to plugin: %v", err)
+	}
+
+	// Return a driver handle
+	h := &xenTunedHandle{
+		pluginClient: pluginClient,
+		executor:     executor,
+		userPid:      id.UserPid,
+		allocDir:     id.AllocDir,
+		logger:       d.logger,
+		killTimeout:  id.KillTimeout,
+		doneCh:       make(chan struct{}),
+		waitCh:       make(chan *cstructs.WaitResult, 1),
+	}
+	go h.run()
+	return h, nil
+
+}
+
+func (h *xenTunedHandle) ID() string {
+	/*
+		Old code
+		id, _ := h.cmd.ID()
+		return id
+	*/
+	id := xenTunedId{
+		KillTimeout:  h.killTimeout,
+		PluginConfig: NewPluginReattachConfig(h.pluginClient.ReattachConfig()),
+		UserPid:      h.userPid,
+		AllocDir:     h.allocDir,
+	}
+
+	data, err := json.Marshal(id)
+	if err != nil {
+		h.logger.Printf("[ERR] driver.xen: failed to marshal ID to JSON: %s", err)
+	}
+	return string(data)
+}
+
+func (h *xenTunedHandle) WaitCh() chan *cstructs.WaitResult {
+	return h.waitCh
+}
+
+func (h *xenTunedHandle) Update(task *structs.Task) error {
+	/*
+		Old code
+		// Update is not possible
+		return nil
+	*/
+	// Store the updated kill timeout.
+	h.killTimeout = task.KillTimeout
+	h.executor.UpdateLogConfig(task.LogConfig)
+
+	// Update is not possible
+	return nil
+}
+
+// Shut-down command
+func (h *xenTunedHandle) Kill() error {
+	/*
+		Old code
+		h.cmd.Shutdown()
+		select {
+		case <-h.doneCh:
+			return nil
+		case <-time.After(5 * time.Second):
+			return h.cmd.ForceStop()
+		}*/
+	if err := h.executor.ShutDown(); err != nil {
+		if h.pluginClient.Exited() {
+			return nil
+		}
+		return fmt.Errorf("executor Shutdown failed: %v", err)
+	}
+
+	select {
+	case <-h.doneCh:
+		return nil
+	case <-time.After(h.killTimeout):
+		if h.pluginClient.Exited() {
+			return nil
+		}
+		if err := h.executor.Exit(); err != nil {
+			return fmt.Errorf("executor Exit failed: %v", err)
+		}
+
+		return nil
+	}
+}
+
+// Run command
+func (h *xenTunedHandle) run() {
+	/*
+		Old code
+		res := h.cmd.Wait()
+		close(h.doneCh)
+		h.waitCh <- res
+		close(h.waitCh)
+	*/
+	start_spawn := time.Now()
+
+	ps, err := h.executor.Wait()
+
+	end_spawn := time.Now()
+
+	start_clean := time.Now()
+
+	if ps.ExitCode == 0 && err != nil {
+		if e := killProcess(h.userPid); e != nil {
+			h.logger.Printf("[ERROR] driver.xen: error killing user process: %v", e)
+		}
+		if e := h.allocDir.UnmountAll(); e != nil {
+			h.logger.Printf("[ERROR] driver.xen: unmounting dev,proc and alloc dirs failed: %v", e)
+		}
+	}
+	close(h.doneCh)
+	h.waitCh <- &cstructs.WaitResult{ExitCode: ps.ExitCode, Signal: 0, Err: err}
+	close(h.waitCh)
+	h.pluginClient.Kill()
+
+	end_clean := time.Now()
+
+	spawn := end_spawn.Sub(start_spawn)
+	clean := end_clean.Sub(start_clean)
+
+	f1, err := os.OpenFile("get_config.txt", os.O_APPEND|os.O_WRONLY|os.O_CREATE, 0600)
+	if err != nil {
+		panic(err)
+	}
+
+	defer f1.Close()
+
+	f2, err := os.OpenFile("alloc_dir.txt", os.O_APPEND|os.O_WRONLY|os.O_CREATE, 0600)
+	if err != nil {
+		panic(err)
+	}
+
+	defer f2.Close()
+
+	f3, err := os.OpenFile("down_artifact.txt", os.O_APPEND|os.O_WRONLY|os.O_CREATE, 0600)
+	if err != nil {
+		panic(err)
+	}
+
+	defer f3.Close()
+
+	f4, err := os.OpenFile("init_env.txt", os.O_APPEND|os.O_WRONLY|os.O_CREATE, 0600)
+	if err != nil {
+		panic(err)
+	}
+
+	defer f4.Close()
+
+	f5, err := os.OpenFile("spawn.txt", os.O_APPEND|os.O_WRONLY|os.O_CREATE, 0600)
+	if err != nil {
+		panic(err)
+	}
+
+	defer f5.Close()
+
+	f6, err := os.OpenFile("clean.txt", os.O_APPEND|os.O_WRONLY|os.O_CREATE, 0600)
+	if err != nil {
+		panic(err)
+	}
+
+	defer f6.Close()
+
+	if _, err = f1.WriteString(fmt.Sprintf("%v\n", float64(h.get_config.Nanoseconds())/1000000)); err != nil {
+		panic(err)
+	}
+
+	if _, err = f2.WriteString(fmt.Sprintf("%v\n", float64(h.alloc_dir.Nanoseconds())/1000000)); err != nil {
+		panic(err)
+	}
+
+	if _, err = f3.WriteString(fmt.Sprintf("%v\n", float64(h.down_artifact.Nanoseconds())/1000000)); err != nil {
+		panic(err)
+	}
+
+	if _, err = f4.WriteString(fmt.Sprintf("%v\n", float64(h.init_env.Nanoseconds())/1000000)); err != nil {
+		panic(err)
+	}
+
+	if _, err = f5.WriteString(fmt.Sprintf("%v\n", float64(spawn.Nanoseconds())/1000000)); err != nil {
+		panic(err)
+	}
+
+	if _, err = f6.WriteString(fmt.Sprintf("%v\n", float64(clean.Nanoseconds())/1000000)); err != nil {
+		panic(err)
+	}
+
+}
-- 
1.8.1.2


From df967cd30ed24f0500a67b9b471a744076a64dc7 Mon Sep 17 00:00:00 2001
From: Pier Luigi Ventre <pierventre@hotmail.com>
Date: Sat, 2 Jul 2016 15:43:21 +0200
Subject: [PATCH 2/2] Bugfix for Xen tuned

---
 client/driver/xen_tuned.go | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/client/driver/xen_tuned.go b/client/driver/xen_tuned.go
index 6642cc5..5c82bd3 100644
--- a/client/driver/xen_tuned.go
+++ b/client/driver/xen_tuned.go
@@ -89,8 +89,8 @@ func (d *XenTunedDriver) Fingerprint(cfg *config.Config, node *structs.Node) (bo
 
 	matches := matches1[1] + "." + matches2[1] + "." + matches3[1]
 
-	node.Attributes["driver.xen"] = "1"
-	node.Attributes["driver.xen.version"] = matches
+	node.Attributes["driver.xen_tuned"] = "1"
+	node.Attributes["driver.xen_tuned.version"] = matches
 
 	return true, nil
 }
-- 
1.8.1.2