runproc sample

client/client.go

@@ -163,4 +163,4 @@ func (c *Client) Get() interface{} {
 }
 
 // Scrub is an Anchor interface method, not applicable to the root-level anchor.
-func (c *Client) Scrub() {}
+func (c *Client) Scrub() {}

client/proc.go

@@ -32,6 +32,7 @@ type Cmd struct {
 	// If Scrub is set, the process element will automatically be removed from its anchor
 	// when the process exits.
 	Scrub bool
+	Name string
 }
 
 func retypeProcStat(c proc.Cmd) Cmd {
@@ -41,6 +42,7 @@ func retypeProcStat(c proc.Cmd) Cmd {
 		Path: c.Path,
 		Args: c.Args,
 		Scrub: c.Scrub,
+		Name: c.Name,
 	}
 }
 
@@ -51,6 +53,7 @@ func (cmd Cmd) retype() proc.Cmd {
 		Path: cmd.Path,
 		Args: cmd.Args,
 		Scrub: cmd.Scrub,
+		Name: cmd.Name,
 	}
 }
 

cmd/circuit/glob_test.go

@@ -15,4 +15,4 @@ import (
 func TestGlob(t *testing.T) {
 	walk, ellipses := parseGlob("/X/hola/petar/...")
 	log.Printf("w=%v ell=%v", walk, ellipses)
-}
+}

cmd/circuit/main.go

@@ -219,6 +219,18 @@ func main() {
 				cli.StringFlag{"hmac", "", "File containing HMAC credentials. Use RC4 encryption."},
 			},
 		},
+		{
+			Name:   "runproc",
+			Usage:  "Run a process element and return output on stdout",
+			Action: runproc,
+			Flags: []cli.Flag{
+				cli.StringFlag{"dial, d", "", "circuit member to dial into"},
+				cli.StringFlag{"discover", "228.8.8.8:8822", "Multicast address for peer server discovery"},
+				cli.BoolFlag{"tag", "tag each output line with the anchor names (hostnames)"},
+				cli.BoolFlag{"all", "run the command across all hosts"},
+				cli.StringFlag{"hmac", "", "File containing HMAC credentials. Use RC4 encryption."},
+			},
+		},
 		{
 			Name:   "signal",
 			Usage:  "Send a signal to a running process",

cmd/circuit/procdkr.go

@@ -8,17 +8,23 @@
 package main
 
 import (
+	"bufio"
 	"encoding/json"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"os"
+	"time"
 
 	"github.com/gocircuit/circuit/client"
 	"github.com/gocircuit/circuit/client/docker"
 
 	"github.com/gocircuit/circuit/github.com/codegangsta/cli"
 )
 
+//make this timeout come from the json input payload
+var timeout = time.Duration(33)
+
 // circuit mkproc /X1234/hola/charlie << EOF
 // { … }
 // EOF
@@ -53,6 +59,131 @@ func mkproc(x *cli.Context) {
 	}
 }
 
+func doRun(x *cli.Context, c *client.Client, cmd client.Cmd, path string) {
+
+	w2, _ := parseGlob(path)
+	a2 := c.Walk(w2)
+	_runproc(a2, cmd, x.Bool("tag"))
+
+}
+
+func runproc(x *cli.Context) {
+	defer func() {
+		if r := recover(); r != nil {
+			fatalf("error, likely due to missing server or misspelled anchor: %v", r)
+		}
+	}()
+	c := dial(x)
+	args := x.Args()
+
+	if len(args) != 1 && !x.Bool("all") {
+		fatalf("runproc needs an anchor argument or use the --all flag to do the entire circuit")
+	}
+	buf, _ := ioutil.ReadAll(os.Stdin)
+	var cmd client.Cmd
+	if err := json.Unmarshal(buf, &cmd); err != nil {
+		fatalf("command json not parsing: %v", err)
+	}
+	if x.Bool("scrub") {
+		cmd.Scrub = true
+	}
+
+	el := string("")
+	if len(cmd.Name) > 0 {
+		el = cmd.Name
+	} else {
+		el = cmd.Path
+	}
+
+	if x.Bool("all") {
+
+		w, _ := parseGlob("/")
+
+		anchor := c.Walk(w)
+
+		done := make(chan bool, 10)
+		procs := 0
+
+		for _, a := range anchor.View() {
+
+			procs++
+
+			go func(x *cli.Context, cmd client.Cmd, a string, done chan bool) {
+
+				doRun(x, c, cmd, a)
+				done <- true
+
+			}(x, cmd, a.Path()+"/"+el, done)
+
+		}
+
+		for ; procs > 0 ; procs--  {
+
+			select {
+			case <-done:
+				continue
+			case <-time.After(time.Second * timeout):
+				fmt.Fprintln(os.Stderr, "timeout waiting for the command")
+			}
+
+		}
+
+	} else {
+
+		doRun(x, c, cmd, args[0]+"/"+el)
+
+	}
+
+}
+
+func _runproc(a client.Anchor, cmd client.Cmd, tags bool) {
+
+	p, err := a.MakeProc(cmd)
+	if err != nil {
+		fatalf("mkproc error: %s", err)
+	}
+
+	q := p.Stdin()
+	if err := q.Close(); err != nil {
+		fatalf("error closing stdin: %v", err)
+	}
+
+	if tags {
+
+		done := make(chan bool)
+
+		go func(a string, r io.Reader, w io.Writer, d chan bool) {
+
+			scanner := bufio.NewScanner(r)
+			for scanner.Scan() {
+				fmt.Fprintf(w, "%s %s\n", a, scanner.Text())
+			}
+			if err := scanner.Err(); err != nil {
+				d <- false
+			}
+
+			d <- true
+
+		}(a.Path(), p.Stdout(), os.Stdout, done)
+
+		select {
+		case rv := <-done:
+			if !rv {
+				fmt.Fprintln(os.Stderr, "error prefixing the data")
+			}
+		case <-time.After(time.Second * timeout):
+			fmt.Fprintln(os.Stderr, "timeout waiting for data")
+		}
+
+	} else {
+
+		io.Copy(os.Stdout, p.Stdout())
+
+	}
+
+	a.Scrub()
+}
+
 func mkdkr(x *cli.Context) {
 	defer func() {
 		if r := recover(); r != nil {
@@ -92,7 +223,9 @@ func sgnl(x *cli.Context) {
 		fatalf("signal needs an anchor and a signal name arguments")
 	}
 	w, _ := parseGlob(args[1])
-	u, ok := c.Walk(w).Get().(interface{Signal(string) error})
+	u, ok := c.Walk(w).Get().(interface {
+		Signal(string) error
+	})
 	if !ok {
 		fatalf("anchor is not a process or a docker container")
 	}

element/proc/cmd.go

@@ -18,6 +18,7 @@ type Cmd struct {
 	Path string   `json:"path"`
 	Args []string `json:"args"`
 	Scrub bool `json:"scrub"`
+	Name string `json:"name"`
 }
 
 func ParseCmd(src string) (*Cmd, error) {

element/proc/proc.go

@@ -45,6 +45,7 @@ type proc struct {
 		sync.Mutex
 		cmd exec.Cmd
 		scrb bool
+		name string
 		abr chan<- struct{}
 		wait chan<- error
 		exit error // exit set by waiter
@@ -66,6 +67,7 @@ func MakeProc(cmd Cmd) Proc {
 	p.cmd.cmd.Dir = cmd.Dir
 	bin := strings.TrimSpace(cmd.Path)
 	p.cmd.cmd.Path = bin
+	p.cmd.name = cmd.Name
 	p.cmd.cmd.Args = append([]string{bin}, cmd.Args...)
 	p.cmd.scrb = cmd.Scrub
 	// exec
@@ -148,6 +150,7 @@ func (p *proc) GetCmd() Cmd {
 		Path: p.cmd.cmd.Path,
 		Args: p.cmd.cmd.Args[1:],
 		Scrub: p.cmd.scrb,
+		Name: p.cmd.name,
 	}
 }
 
@@ -177,6 +180,7 @@ func (p *proc) peek() Stat {
 			Path: p.cmd.cmd.Path,
 			Args: p.cmd.cmd.Args[1:],
 			Scrub: p.cmd.scrb,
+			Name: p.cmd.name,
 		},
 		Exit: p.cmd.exit,
 		Phase: p.phase().String(),