Remote runner web

core/local.go

@@ -11,14 +11,14 @@ import (
 const READ_BUFFER_SIZE = 1024
 
 type Local struct {
-	ChStdOut chan []byte
-	ChStdErr chan []byte
+	chStdOut chan []byte
+	chStdErr chan []byte
 }
 
 func NewLocalRunner() (*Local, error) {
 	lr := Local{
-		ChStdOut: make(chan []byte),
-		ChStdErr: make(chan []byte),
+		chStdOut: make(chan []byte),
+		chStdErr: make(chan []byte),
 	}
 
 	return &lr, nil
@@ -62,12 +62,12 @@ func (l *Local) Run(task Task) error {
 			}
 
 			if outBytes > 0 {
-				l.ChStdOut <- bufferStdOut[:outBytes]
+				l.ChStdOut() <- bufferStdOut[:outBytes]
 			}
 		}
 
 		if listeningErr {
-			errBytes, err := stdOut.Read(bufferStdErr)
+			errBytes, err := stdErr.Read(bufferStdErr)
 			if err != nil {
 				if err == io.EOF {
 					listeningErr = false
@@ -77,7 +77,7 @@ func (l *Local) Run(task Task) error {
 			}
 
 			if errBytes > 0 {
-				l.ChStdErr <- bufferStdErr[:errBytes]
+				l.ChStdErr() <- bufferStdErr[:errBytes]
 			}
 		}
 
@@ -97,10 +97,18 @@ func (l *Local) Close() {
 func LogOutput(runner *Local) {
 	for {
 		select {
-		case out := <-runner.ChStdOut:
+		case out := <-runner.chStdOut:
 			log.Printf("%s", out)
-		case out := <-runner.ChStdErr:
+		case out := <-runner.chStdErr:
 			log.Printf("%s", out)
 		}
 	}
 }
+
+func (l *Local) ChStdOut() chan []byte {
+	return l.chStdOut
+}
+
+func (l *Local) ChStdErr() chan []byte {
+	return l.chStdOut
+}

core/remote.go

@@ -6,7 +6,7 @@ import (
 	"fmt"
 	"io/ioutil"
 	"log"
-	"os"
+	"io"
 	"strings"
 
 	"golang.org/x/crypto/ssh"
@@ -17,7 +17,9 @@ import (
 // Remote describes a runner which runs task
 // on a remote system via SSH.
 type Remote struct {
-	Client *ssh.Client
+	chStdOut chan []byte
+	chStdErr chan []byte
+	Client   *ssh.Client
 }
 
 // NewRemoteRunner creates a new runner which runs
@@ -33,6 +35,8 @@ func NewRemoteRunner(options ConfigServer) (*Remote, error) {
 	}
 
 	return &Remote{
+		chStdOut: make(chan []byte),
+		chStdErr: make(chan []byte),
 		Client: client,
 	}, nil
 }
@@ -49,12 +53,61 @@ func (r *Remote) Run(task Task) error {
 
 	cmd := fmt.Sprintf("%v %v", task.Name(), strings.Join(task.Args(), " "))
 
-	session.Stdout = os.Stdout
-	session.Stderr = os.Stderr
+	stdOut, err := session.StdoutPipe()
+	if err != nil {
+		return err
+	}
+
+	stdErr, err := session.StderrPipe()
+	if err != nil {
+		return err
+	}
+
 	if err := session.Start(cmd); err != nil {
 		return err
 	}
 
+	// COULDDO: Might be slicker to spin up two async processes that communicate back
+	bufferStdOut := make([]byte, READ_BUFFER_SIZE)
+	bufferStdErr := make([]byte, READ_BUFFER_SIZE)
+	listeningOut := true
+	listeningErr := true
+	for {
+		if listeningOut {
+			outBytes, err := stdOut.Read(bufferStdOut)
+			if err != nil {
+				if err == io.EOF {
+					listeningOut = false
+				} else {
+					return err
+				}
+			}
+
+			if outBytes > 0 {
+				r.ChStdOut() <- bufferStdOut[:outBytes]
+			}
+		}
+
+		if listeningErr {
+			errBytes, err := stdErr.Read(bufferStdErr)
+			if err != nil {
+				if err == io.EOF {
+					listeningErr = false
+				} else {
+					return err
+				}
+			}
+
+			if errBytes > 0 {
+				r.ChStdErr() <- bufferStdErr[:errBytes]
+			}
+		}
+
+		if !listeningOut && !listeningErr { // both complete
+			break
+		}
+	}
+
 	return session.Wait()
 }
 
@@ -117,3 +170,11 @@ func loadKey(file string) (interface{}, error) {
 
 	return key, nil
 }
+
+func (r *Remote) ChStdOut() chan []byte {
+	return r.chStdOut
+}
+
+func (r *Remote) ChStdErr() chan []byte {
+	return r.chStdOut
+}

core/runner.go

@@ -1,6 +1,26 @@
 package core
 
 type Runner interface {
+	ChStdOut() chan []byte
+	ChStdErr() chan []byte
 	Run(Task) error
 	Close()
 }
+
+func GetRunner(host ConfigServer) (Runner, error) {
+	var runner Runner
+	var err error
+	if host.RunLocally {
+		runner, err = NewLocalRunner()
+		if err != nil {
+			return nil, err
+		}
+	} else {
+		runner, err = NewRemoteRunner(host)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return runner, nil
+}

daemon/config/test-tasks.yaml

@@ -1,27 +1,10 @@
 ---
-vars:
-  mailgun_user: "uservaluehere"
-  mailgun_key: "keyvaluehere"
 servergroups:
-  localhost:
   web:
-    - host: 10.10.0.0
-      username: "root"
-      password: "password"
-      vars:
-        mailgun_user: "user2valuehere"
-        mailgun_key: "key2valuehere"
-    - host: 10.10.0.1
-      username: "root"
-      password: "password"
-      key_path: "/var/ssh/keys/"
-  database:
-    - host: 10.10.0.2
-      username: "root"
-      password: "password"
-      vars:
-        mailgun_user: "user2valuehere"
-        mailgun_key: "key2valuehere"
+    - host: localhost
+      username: "vagrant"
+      password: "vagrant"
+      port: 2222
 tasks:
   sayhello:
     steps: