func (c *ExecCommand) createSessionLocal() (string, error) {
	session := c.client.Session()
	se := consulapi.SessionEntry{
		Name:     "Remote Exec",
		Behavior: consulapi.SessionBehaviorDelete,
		TTL:      rExecTTL,
	}
	id, _, err := session.Create(&se, nil)
	return id, err
}
createSessionLocal is used to create a new session in a local datacenter
This is simpler since we can use the local agent to create the session.
package command
import (
	"bytes"
	"encoding/json"
	"flag"
	"fmt"
	"io"
	"os"
	"path"
	"regexp"
	"strconv"
	"strings"
	"time"
	"unicode"
	consulapi "github.com/socketplane/socketplane/Godeps/_workspace/src/github.com/hashicorp/consul/api"
	"github.com/socketplane/socketplane/Godeps/_workspace/src/github.com/mitchellh/cli"
)
const (
	rExecPrefix = "_rexec"
	rExecFileName = "job"
	rExecAckSuffix = "/ack"
	rExecExitSuffix = "/exit"
	rExecOutputDivider = "/out/"
	rExecReplicationWait = 200 * time.Millisecond
	rExecQuietWait = 2 * time.Second
	rExecTTL = "15s"
	rExecRenewInterval = 5 * time.Second
)
type rExecConf struct {
	datacenter string
	prefix     string
	foreignDC bool
	localDC   string
	localNode string
	node    string
	service string
	tag     string
	wait     time.Duration
	replWait time.Duration
	cmd    string
	script []byte
	verbose bool
}
type rExecEvent struct {
	Prefix  string
	Session string
}
type rExecSpec struct {
	Command string `json:",omitempty"`
	Script []byte `json:",omitempty"`
	Wait time.Duration
}
type rExecAck struct {
	Node string
}
type rExecHeart struct {
	Node string
}
type rExecOutput struct {
	Node   string
	Output []byte
}
type rExecExit struct {
	Node string
	Code int
}
type ExecCommand struct {
	ShutdownCh <-chan struct{}
	Ui         cli.Ui
	conf       rExecConf
	client     *consulapi.Client
	sessionID  string
	stopCh     chan struct{}
}
func (c *ExecCommand) Run(args []string) int {
	cmdFlags := flag.NewFlagSet("exec", flag.ContinueOnError)
	cmdFlags.Usage = func() { c.Ui.Output(c.Help()) }
	cmdFlags.StringVar(&c.conf.datacenter, "datacenter", "", "")
	cmdFlags.StringVar(&c.conf.node, "node", "", "")
	cmdFlags.StringVar(&c.conf.service, "service", "", "")
	cmdFlags.StringVar(&c.conf.tag, "tag", "", "")
	cmdFlags.StringVar(&c.conf.prefix, "prefix", rExecPrefix, "")
	cmdFlags.DurationVar(&c.conf.replWait, "wait-repl", rExecReplicationWait, "")
	cmdFlags.DurationVar(&c.conf.wait, "wait", rExecQuietWait, "")
	cmdFlags.BoolVar(&c.conf.verbose, "verbose", false, "")
	httpAddr := HTTPAddrFlag(cmdFlags)
	if err := cmdFlags.Parse(args); err != nil {
		return 1
	}
	c.conf.cmd = strings.Join(cmdFlags.Args(), " ")
	if c.conf.cmd == "-" {
		c.conf.cmd = ""
		var buf bytes.Buffer
		_, err := io.Copy(&buf, os.Stdin)
		if err != nil {
			c.Ui.Error(fmt.Sprintf("Failed to read stdin: %v", err))
			c.Ui.Error("")
			c.Ui.Error(c.Help())
			return 1
		}
		c.conf.script = buf.Bytes()
	}
	if c.conf.cmd == "" && len(c.conf.script) == 0 {
		c.Ui.Error("Must specify a command to execute")
		c.Ui.Error("")
		c.Ui.Error(c.Help())
		return 1
	}
	if err := c.conf.validate(); err != nil {
		c.Ui.Error(err.Error())
		return 1
	}
	client, err := HTTPClientDC(*httpAddr, c.conf.datacenter)
	if err != nil {
		c.Ui.Error(fmt.Sprintf("Error connecting to Consul agent: %s", err))
		return 1
	}
	info, err := client.Agent().Self()
	if err != nil {
		c.Ui.Error(fmt.Sprintf("Error querying Consul agent: %s", err))
		return 1
	}
	c.client = client
	if c.conf.datacenter != "" && c.conf.datacenter != info["Config"]["Datacenter"] {
		if c.conf.verbose {
			c.Ui.Info("Remote exec in foreign datacenter, using Session TTL")
		}
		c.conf.foreignDC = true
		c.conf.localDC = info["Config"]["Datacenter"].(string)
		c.conf.localNode = info["Config"]["NodeName"].(string)
	}
	spec, err := c.makeRExecSpec()
	if err != nil {
		c.Ui.Error(fmt.Sprintf("Failed to create job spec: %s", err))
		return 1
	}
	c.sessionID, err = c.createSession()
	if err != nil {
		c.Ui.Error(fmt.Sprintf("Failed to create session: %s", err))
		return 1
	}
	defer c.destroySession()
	if c.conf.verbose {
		c.Ui.Info(fmt.Sprintf("Created remote execution session: %s", c.sessionID))
	}
	if err := c.uploadPayload(spec); err != nil {
		c.Ui.Error(fmt.Sprintf("Failed to create job file: %s", err))
		return 1
	}
	defer c.destroyData()
	if c.conf.verbose {
		c.Ui.Info(fmt.Sprintf("Uploaded remote execution spec"))
	}
	select {
	case <-time.After(c.conf.replWait):
	case <-c.ShutdownCh:
		return 1
	}
	id, err := c.fireEvent()
	if err != nil {
		c.Ui.Error(fmt.Sprintf("Failed to fire event: %s", err))
		return 1
	}
	if c.conf.verbose {
		c.Ui.Info(fmt.Sprintf("Fired remote execution event: %s", id))
	}
	return c.waitForJob()
}
func (c *ExecCommand) waitForJob() int {
	defer c.destroySession()
	start := time.Now()
	ackCh := make(chan rExecAck, 128)
	heartCh := make(chan rExecHeart, 128)
	outputCh := make(chan rExecOutput, 128)
	exitCh := make(chan rExecExit, 128)
	doneCh := make(chan struct{})
	errCh := make(chan struct{}, 1)
	defer close(doneCh)
	go c.streamResults(doneCh, ackCh, heartCh, outputCh, exitCh, errCh)
	target := &TargettedUi{Ui: c.Ui}
	var ackCount, exitCount, badExit int
OUTER:
	for {
		waitIntv := c.conf.wait
		if ackCount > exitCount {
			waitIntv *= 2
		}
		select {
		case e := <-ackCh:
			ackCount++
			if c.conf.verbose {
				target.Target = e.Node
				target.Info("acknowledged")
			}
		case h := <-heartCh:
			if c.conf.verbose {
				target.Target = h.Node
				target.Info("heartbeat received")
			}
		case e := <-outputCh:
			target.Target = e.Node
			target.Output(string(e.Output))
		case e := <-exitCh:
			exitCount++
			target.Target = e.Node
			target.Info(fmt.Sprintf("finished with exit code %d", e.Code))
			if e.Code != 0 {
				badExit++
			}
		case <-time.After(waitIntv):
			c.Ui.Info(fmt.Sprintf("%d / %d node(s) completed / acknowledged", exitCount, ackCount))
			if c.conf.verbose {
				c.Ui.Info(fmt.Sprintf("Completed in %0.2f seconds",
					float64(time.Now().Sub(start))/float64(time.Second)))
			}
			break OUTER
		case <-errCh:
			return 1
		case <-c.ShutdownCh:
			return 1
		}
	}
	if badExit > 0 {
		return 2
	}
	return 0
}
func (c *ExecCommand) streamResults(doneCh chan struct{}, ackCh chan rExecAck, heartCh chan rExecHeart,
	outputCh chan rExecOutput, exitCh chan rExecExit, errCh chan struct{}) {
	kv := c.client.KV()
	opts := consulapi.QueryOptions{WaitTime: c.conf.wait}
	dir := path.Join(c.conf.prefix, c.sessionID) + "/"
	seen := make(map[string]struct{})
	for {
		select {
		case <-doneCh:
			return
		default:
		}
		keys, qm, err := kv.Keys(dir, "", &opts)
		if err != nil {
			c.Ui.Error(fmt.Sprintf("Failed to read results: %s", err))
			goto ERR_EXIT
		}
		if qm.LastIndex == opts.WaitIndex {
			continue
		}
		opts.WaitIndex = qm.LastIndex
		for _, key := range keys {
			if _, ok := seen[key]; ok {
				continue
			}
			seen[key] = struct{}{}
			full := key
			key = strings.TrimPrefix(key, dir)
			switch {
			case key == rExecFileName:
				continue
			case strings.HasSuffix(key, rExecAckSuffix):
				ackCh <- rExecAck{Node: strings.TrimSuffix(key, rExecAckSuffix)}
			case strings.HasSuffix(key, rExecExitSuffix):
				pair, _, err := kv.Get(full, nil)
				if err != nil {
					c.Ui.Error(fmt.Sprintf("Failed to read key '%s': %v", full, err))
					continue
				}
				code, err := strconv.ParseInt(string(pair.Value), 10, 32)
				if err != nil {
					c.Ui.Error(fmt.Sprintf("Failed to parse exit code '%s': %v", pair.Value, err))
					continue
				}
				exitCh <- rExecExit{
					Node: strings.TrimSuffix(key, rExecExitSuffix),
					Code: int(code),
				}
			case strings.LastIndex(key, rExecOutputDivider) != -1:
				pair, _, err := kv.Get(full, nil)
				if err != nil {
					c.Ui.Error(fmt.Sprintf("Failed to read key '%s': %v", full, err))
					continue
				}
				idx := strings.LastIndex(key, rExecOutputDivider)
				node := key[:idx]
				if len(pair.Value) == 0 {
					heartCh <- rExecHeart{Node: node}
				} else {
					outputCh <- rExecOutput{Node: node, Output: pair.Value}
				}
			default:
				c.Ui.Error(fmt.Sprintf("Unknown key '%s', ignoring.", key))
			}
		}
	}
ERR_EXIT:
	select {
	case errCh <- struct{}{}:
	default:
	}
}
func (conf *rExecConf) validate() error {
	if conf.node != "" {
		if _, err := regexp.Compile(conf.node); err != nil {
			return fmt.Errorf("Failed to compile node filter regexp: %v", err)
		}
	}
	if conf.service != "" {
		if _, err := regexp.Compile(conf.service); err != nil {
			return fmt.Errorf("Failed to compile service filter regexp: %v", err)
		}
	}
	if conf.tag != "" {
		if _, err := regexp.Compile(conf.tag); err != nil {
			return fmt.Errorf("Failed to compile tag filter regexp: %v", err)
		}
	}
	if conf.tag != "" && conf.service == "" {
		return fmt.Errorf("Cannot provide tag filter without service filter.")
	}
	return nil
}
func (c *ExecCommand) createSession() (string, error) {
	var id string
	var err error
	if c.conf.foreignDC {
		id, err = c.createSessionForeign()
	} else {
		id, err = c.createSessionLocal()
	}
	if err == nil {
		c.stopCh = make(chan struct{})
		go c.renewSession(id, c.stopCh)
	}
	return id, err
}