add status command

Gopkg.lock

@@ -89,6 +89,14 @@
   revision = "0360b2af4f38e8d38c7fce2a9f4e702702d73a39"
   version = "v0.0.3"
 
+[[projects]]
+  digest = "1:82b912465c1da0668582a7d1117339c278e786c2536b3c3623029a0c7141c2d0"
+  name = "github.com/mattn/go-runewidth"
+  packages = ["."]
+  pruneopts = ""
+  revision = "ce7b0b5c7b45a81508558cd1dba6bb1e4ddb51bb"
+  version = "v0.0.3"
+
 [[projects]]
   digest = "1:4c23ced97a470b17d9ffd788310502a077b9c1f60221a85563e49696276b4147"
   name = "github.com/matttproud/golang_protobuf_extensions"
@@ -105,6 +113,14 @@
   pruneopts = ""
   revision = "d0303fe809921458f417bcf828397a65db30a7e4"
 
+[[projects]]
+  branch = "master"
+  digest = "1:20a553eff588d7abe1f05addf5f57cdbaef1d0f992427a0099b7eb51274b79cf"
+  name = "github.com/nsf/termbox-go"
+  packages = ["."]
+  pruneopts = ""
+  revision = "b66b20ab708e289ff1eb3e218478302e6aec28ce"
+
 [[projects]]
   digest = "1:7365acd48986e205ccb8652cc746f09c8b7876030d53710ea6ef7d0bd0dcd7ca"
   name = "github.com/pkg/errors"
@@ -246,6 +262,7 @@
     "github.com/kr/pretty",
     "github.com/mattn/go-isatty",
     "github.com/mitchellh/mapstructure",
+    "github.com/nsf/termbox-go",
     "github.com/pkg/errors",
     "github.com/problame/go-netssh",
     "github.com/problame/go-rwccmd",

Gopkg.toml

@@ -63,3 +63,7 @@ ignored = [ "github.com/inconshreveable/mousetrap" ]
 [[constraint]]
   name = "github.com/golang/protobuf"
   version = "1.2.0"
+
+[[constraint]]
+  name = "github.com/nsf/termbox-go"
+  branch = "master"

client/status.go

@@ -0,0 +1,227 @@
+package client
+
+import (
+	"github.com/zrepl/zrepl/config"
+	"github.com/zrepl/zrepl/daemon"
+	"fmt"
+	"github.com/zrepl/zrepl/replication"
+	"github.com/mitchellh/mapstructure"
+	"github.com/zrepl/zrepl/replication/fsrep"
+	"github.com/nsf/termbox-go"
+	"time"
+	"github.com/pkg/errors"
+	"sort"
+	"sync"
+)
+
+type tui struct {
+	x, y int
+	indent int
+
+	lock sync.Mutex //For report and error
+	report map[string]interface{}
+	err error
+}
+
+func newTui() tui {
+	return tui{}
+}
+
+func (t *tui) moveCursor(x, y int) {
+	t.x += x
+	t.y += y
+}
+
+func (t *tui) moveLine(dl int, col int) {
+	t.y += dl
+	t.x = t.indent * 4 + col
+}
+
+func (t *tui) write(text string) {
+	for _, c := range text {
+		termbox.SetCell(t.x, t.y, c, termbox.ColorDefault, termbox.ColorDefault)
+		t.x += 1
+	}
+}
+
+func (t *tui) printf(text string, a ...interface{}) {
+	t.write(fmt.Sprintf(text, a...))
+}
+
+func (t *tui) newline() {
+	t.moveLine(1, 0)
+}
+
+func (t *tui) setIndent(indent int) {
+	t.indent = indent
+	t.moveLine(0, 0)
+}
+
+func (t *tui) addIndent(indent int) {
+	t.indent += indent
+	t.moveLine(0, 0)
+}
+
+
+func RunStatus(config config.Config, args []string) error {
+	httpc, err := controlHttpClient(config.Global.Control.SockPath)
+	if err != nil {
+		return err
+	}
+
+	t := newTui()
+	t.lock.Lock()
+	t.err = errors.New("Got no report yet")
+	t.lock.Unlock()
+
+	err = termbox.Init()
+	if err != nil {
+		return err
+	}
+	defer termbox.Close()
+
+	ticker := time.NewTicker(500 * time.Millisecond)
+	defer ticker.Stop()
+	go func() {
+		for _ = range ticker.C {
+			m := make(map[string]interface{})
+
+			err2 := jsonRequestResponse(httpc, daemon.ControlJobEndpointStatus,
+				struct {}{},
+				&m,
+			)
+
+			t.lock.Lock()
+			t.err = err2
+			t.report = m
+			t.lock.Unlock()
+			t.draw()
+		}
+	}()
+
+	termbox.HideCursor()
+	termbox.Clear(termbox.ColorDefault, termbox.ColorDefault)
+
+	loop:
+	for {
+		switch ev := termbox.PollEvent(); ev.Type {
+		case termbox.EventKey:
+			switch ev.Key {
+			case termbox.KeyEsc:
+				break loop
+			case termbox.KeyCtrlC:
+				break loop
+			}
+		case termbox.EventResize:
+			t.draw()
+		}
+	}
+
+	return nil
+
+}
+
+func (t *tui) draw() {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+
+	termbox.Clear(termbox.ColorDefault, termbox.ColorDefault)
+	t.x = 0
+	t.y = 0
+	t.indent = 0
+
+	if t.err != nil {
+		t.write(t.err.Error())
+	} else {
+		//Iterate over map in alphabetical order
+		keys := make([]string, len(t.report))
+		i := 0
+		for k, _ := range t.report {
+			keys[i] = k
+			i++
+		}
+		sort.Strings(keys)
+
+		for _, k := range keys {
+			v := t.report[k]
+			if len(k) == 0 || k[0] == '_' { //Internal job
+				continue
+			}
+			t.setIndent(0)
+
+			t.printf("Job: %s", k)
+			t.setIndent(1)
+			t.newline()
+
+			if v == nil {
+				t.printf("No report generated yet")
+				t.newline()
+				continue
+			}
+			rep := replication.Report{}
+			err := mapstructure.Decode(v, &rep)
+			if err != nil {
+				t.printf("Failed to decode report: %s", err.Error())
+				t.newline()
+				continue
+			}
+			t.printf("Status:  %s", rep.Status)
+			t.newline()
+			t.printf("Problem: %s", rep.Problem)
+			t.newline()
+
+			for _, fs := range rep.Completed {
+				printFilesystem(fs, t)
+			}
+			if rep.Active != nil {
+				printFilesystem(rep.Active, t)
+			}
+			for _, fs := range rep.Pending {
+				printFilesystem(fs, t)
+			}
+
+		}
+	}
+	termbox.Flush()
+}
+
+func times(str string, n int) (out string) {
+	for i := 0; i < n; i++ {
+		out += str
+	}
+	return
+}
+
+func rightPad(str string, length int, pad string) string {
+	return str + times(pad, length-len(str))
+}
+
+func (t *tui) drawBar(name string, status string, total int, done int) {
+	t.write(rightPad(name, 20, " "))
+	t.write(" ")
+	t.write(rightPad(status, 20, " "))
+
+	if total > 0 {
+		length := 50
+		completedLength := length * done / total
+
+		//FIXME finished bar has 1 off size compared to not finished bar
+		t.write(times("=", completedLength-1))
+		t.write(">")
+		t.write(times("-", length-completedLength))
+
+		t.printf(" %d/%d", done, total)
+	}
+
+	t.newline()
+}
+
+func printFilesystem(rep *fsrep.Report, t *tui) {
+	t.drawBar(rep.Filesystem, rep.Status, len(rep.Completed) + len(rep.Pending), len(rep.Completed))
+	if (rep.Problem != "") {
+		t.addIndent(1)
+		t.printf("Problem: %s", rep.Problem)
+		t.newline()
+		t.addIndent(-1)
+	}
+}

daemon/job/push.go

@@ -55,7 +55,18 @@ func PushFromConfig(g config.Global, in *config.PushJob) (j *Push, err error) {
 func (j *Push) Name() string { return j.name }
 
 func (j *Push) Status() interface{} {
-	return nil // FIXME
+	rep := func() *replication.Replication {
+		j.mtx.Lock()
+		defer j.mtx.Unlock()
+		if j.replication == nil {
+			return nil
+		}
+		return j.replication
+	}()
+	if rep == nil {
+		return nil
+	}
+	return rep.Report()
 }
 
 func (j *Push) Run(ctx context.Context) {
@@ -94,11 +105,11 @@ func (j *Push) do(ctx context.Context) {
 	receiver := endpoint.NewRemote(client)
 
 	j.mtx.Lock()
-	rep := replication.NewReplication()
+	j.replication = replication.NewReplication()
 	j.mtx.Unlock()
 
 	ctx = logging.WithSubsystemLoggers(ctx, log)
-	rep.Drive(ctx, sender, receiver)
+	j.replication.Drive(ctx, sender, receiver)
 
 	// Prune sender
 	senderPruner := pruner.NewPruner(10*time.Second, sender, sender, j.keepRulesSender) // FIXME constant

main.go

@@ -5,6 +5,7 @@ import (
 	"github.com/spf13/cobra"
 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon"
+	"github.com/zrepl/zrepl/client"
 	"log"
 	"os"
 )
@@ -40,7 +41,19 @@ var wakeupCmd = &cobra.Command{
 		if err != nil {
 			return err
 		}
-		return RunWakeup(conf, args)
+		return client.RunWakeup(conf, args)
+	},
+}
+
+var statusCmd = &cobra.Command{
+	Use:   "status",
+	Short: "status",
+	RunE: func(cmd *cobra.Command, args []string) error {
+		conf, err := config.ParseConfig(rootArgs.configFile)
+		if err != nil {
+			return err
+		}
+		return client.RunStatus(conf, args)
 	},
 }
 
@@ -53,6 +66,7 @@ func init() {
 	rootCmd.PersistentFlags().StringVar(&rootArgs.configFile, "config", "", "config file path")
 	rootCmd.AddCommand(daemonCmd)
 	rootCmd.AddCommand(wakeupCmd)
+	rootCmd.AddCommand(statusCmd)
 }
 
 func main() {