zrepl/client/status.go

548 lines
12 KiB
Go
Raw Normal View History

2018-08-29 19:18:54 +02:00
package client
import (
"fmt"
"github.com/gdamore/tcell/termbox"
2018-08-29 19:18:54 +02:00
"github.com/pkg/errors"
"github.com/spf13/pflag"
"github.com/zrepl/yaml-config"
"github.com/zrepl/zrepl/cli"
2018-08-29 22:06:24 +02:00
"github.com/zrepl/zrepl/daemon"
"github.com/zrepl/zrepl/daemon/job"
"github.com/zrepl/zrepl/daemon/pruner"
"github.com/zrepl/zrepl/replication"
2018-08-29 22:06:24 +02:00
"github.com/zrepl/zrepl/replication/fsrep"
"io"
"math"
"net/http"
"os"
2018-08-29 19:18:54 +02:00
"sort"
"strings"
2018-08-29 19:18:54 +02:00
"sync"
2018-08-29 22:06:24 +02:00
"time"
2018-08-29 19:18:54 +02:00
)
type tui struct {
2018-08-29 22:06:24 +02:00
x, y int
2018-08-29 19:18:54 +02:00
indent int
2018-08-29 22:06:24 +02:00
lock sync.Mutex //For report and error
report map[string]job.Status
2018-08-29 22:06:24 +02:00
err error
2018-08-29 19:18:54 +02:00
}
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
2018-08-29 22:06:24 +02:00
t.x = t.indent*4 + col
2018-08-29 19:18:54 +02:00
}
func (t *tui) write(text string) {
for _, c := range text {
if c == '\n' {
t.newline()
continue
}
2018-08-29 19:18:54 +02:00
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)
}
var statusFlags struct {
2018-09-23 14:44:53 +02:00
Raw bool
}
var StatusCmd = &cli.Subcommand{
Use: "status",
Short: "show job activity or dump as JSON for monitoring",
SetupFlags: func(f *pflag.FlagSet) {
f.BoolVar(&statusFlags.Raw, "raw", false, "dump raw status description from zrepl daemon")
},
Run: runStatus,
}
func runStatus(s *cli.Subcommand, args []string) error {
httpc, err := controlHttpClient(s.Config().Global.Control.SockPath)
2018-08-29 19:18:54 +02:00
if err != nil {
return err
}
if statusFlags.Raw {
2018-09-23 14:44:53 +02:00
resp, err := httpc.Get("http://unix"+daemon.ControlJobEndpointStatus)
if err != nil {
return err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
fmt.Fprintf(os.Stderr, "Received error response:\n")
io.CopyN(os.Stderr, resp.Body, 4096)
return errors.Errorf("exit")
}
if _, err := io.Copy(os.Stdout, resp.Body); err != nil {
return err
}
return nil
}
2018-08-29 19:18:54 +02:00
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()
2018-08-29 22:06:24 +02:00
update := func() {
m := make(map[string]job.Status)
2018-08-29 22:06:24 +02:00
err2 := jsonRequestResponse(httpc, daemon.ControlJobEndpointStatus,
struct{}{},
&m,
)
t.lock.Lock()
t.err = err2
t.report = m
t.lock.Unlock()
t.draw()
}
update()
2018-08-29 19:18:54 +02:00
ticker := time.NewTicker(500 * time.Millisecond)
defer ticker.Stop()
go func() {
for _ = range ticker.C {
2018-08-29 22:06:24 +02:00
update()
2018-08-29 19:18:54 +02:00
}
}()
termbox.HideCursor()
termbox.Clear(termbox.ColorDefault, termbox.ColorDefault)
2018-08-29 22:06:24 +02:00
loop:
2018-08-29 19:18:54 +02:00
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 || daemon.IsInternalJobName(k) { //Internal job
2018-08-29 19:18:54 +02:00
continue
}
t.setIndent(0)
t.printf("Job: %s", k)
t.setIndent(1)
t.newline()
t.printf("Type: %s", v.Type)
t.setIndent(1)
t.newline()
if v.Type != job.TypePush && v.Type != job.TypePull {
t.printf("No status representation for job type '%s', dumping as YAML", v.Type)
t.newline()
asYaml, err := yaml.Marshal(v.JobSpecific)
if err != nil {
t.printf("Error marshaling status to YAML: %s", err)
t.newline()
continue
}
t.write(string(asYaml))
t.newline()
continue
}
2018-08-29 19:18:54 +02:00
pushStatus, ok := v.JobSpecific.(*job.ActiveSideStatus)
if !ok || pushStatus == nil {
t.printf("ActiveSideStatus is null")
2018-08-29 19:18:54 +02:00
t.newline()
continue
}
t.printf("Replication:")
t.newline()
t.addIndent(1)
t.renderReplicationReport(pushStatus.Replication)
t.addIndent(-1)
t.printf("Pruning Sender:")
2018-08-29 19:18:54 +02:00
t.newline()
t.addIndent(1)
t.renderPrunerReport(pushStatus.PruningSender)
t.addIndent(-1)
2018-08-29 19:18:54 +02:00
t.printf("Pruning Receiver:")
t.newline()
t.addIndent(1)
t.renderPrunerReport(pushStatus.PruningReceiver)
t.addIndent(-1)
}
}
termbox.Flush()
}
func (t *tui) renderReplicationReport(rep *replication.Report) {
if rep == nil {
t.printf("...\n")
return
}
all := make([]*fsrep.Report, 0, len(rep.Completed)+len(rep.Pending) + 1)
all = append(all, rep.Completed...)
all = append(all, rep.Pending...)
if rep.Active != nil {
all = append(all, rep.Active)
}
sort.Slice(all, func(i, j int) bool {
return all[i].Filesystem < all[j].Filesystem
})
state, err := replication.StateString(rep.Status)
if err != nil {
t.printf("Status: %q (parse error: %q)\n", rep.Status, err)
return
}
t.printf("Status: %s", state)
t.newline()
if rep.Problem != "" {
t.printf("Problem: %s", rep.Problem)
t.newline()
}
if rep.SleepUntil.After(time.Now()) &&
state & ^(replication.ContextDone|replication.Completed) != 0 {
t.printf("Sleeping until %s (%s left)\n", rep.SleepUntil, rep.SleepUntil.Sub(time.Now()))
}
if state != replication.Planning && state != replication.PlanningError {
// Progress: [---------------]
sumUpFSRep := func(rep *fsrep.Report) (transferred, total int64) {
for _, s := range rep.Pending {
transferred += s.Bytes
total += s.ExpectedBytes
}
for _, s := range rep.Completed {
transferred += s.Bytes
total += s.ExpectedBytes
}
return
2018-08-29 19:18:54 +02:00
}
var transferred, total int64
for _, fs := range all {
fstx, fstotal := sumUpFSRep(fs)
transferred += fstx
total += fstotal
}
t.write("Progress: ")
t.drawBar(80, transferred, total)
t.write(fmt.Sprintf(" %s / %s", ByteCountBinary(transferred), ByteCountBinary(total)))
t.newline()
2018-08-29 19:18:54 +02:00
}
var maxFSLen int
for _, fs := range all {
if len(fs.Filesystem) > maxFSLen {
maxFSLen = len(fs.Filesystem)
}
}
for _, fs := range all {
printFilesystemStatus(fs, t, fs == rep.Active, maxFSLen)
}
}
func (t *tui) renderPrunerReport(r *pruner.Report) {
if r == nil {
t.printf("...\n")
return
}
state, err := pruner.StateString(r.State)
if err != nil {
t.printf("Status: %q (parse error: %q)\n", r.State, err)
return
}
t.printf("Status: %s", state)
t.newline()
if r.Error != "" {
t.printf("Error: %s\n", r.Error)
}
if r.SleepUntil.After(time.Now()) {
t.printf("Sleeping until %s (%s left)\n", r.SleepUntil, r.SleepUntil.Sub(time.Now()))
}
type commonFS struct {
*pruner.FSReport
completed bool
}
all := make([]commonFS, 0, len(r.Pending) + len(r.Completed))
for i := range r.Pending {
all = append(all, commonFS{&r.Pending[i], false})
}
for i := range r.Completed {
all = append(all, commonFS{&r.Completed[i], true})
}
switch state {
case pruner.Plan: fallthrough
case pruner.PlanWait: fallthrough
case pruner.ErrPerm:
return
}
if len(all) == 0 {
t.printf("nothing to do\n")
return
}
var totalDestroyCount, completedDestroyCount int
var maxFSname int
for _, fs := range all {
totalDestroyCount += len(fs.DestroyList)
if fs.completed {
completedDestroyCount += len(fs.DestroyList)
}
if maxFSname < len(fs.Filesystem) {
maxFSname = len(fs.Filesystem)
}
}
// global progress bar
progress := int(math.Round(80 * float64(completedDestroyCount) / float64(totalDestroyCount)))
t.write("Progress: ")
t.write("[")
t.write(times("=", progress))
t.write(">")
t.write(times("-", 80 - progress))
t.write("]")
t.printf(" %d/%d snapshots", completedDestroyCount, totalDestroyCount)
t.newline()
sort.SliceStable(all, func(i, j int) bool {
return strings.Compare(all[i].Filesystem, all[j].Filesystem) == -1
})
// Draw a table-like representation of 'all'
for _, fs := range all {
t.write(rightPad(fs.Filesystem, maxFSname, " "))
t.write(" ")
if fs.Error != "" {
t.printf("ERROR: %s\n", fs.Error) // whitespace is padding
continue
}
pruneRuleActionStr := fmt.Sprintf("(destroy %d of %d snapshots)",
len(fs.DestroyList), len(fs.SnapshotList))
if fs.completed {
t.printf( "Completed %s\n", pruneRuleActionStr)
continue
}
t.write("Pending ") // whitespace is padding 10
if len(fs.DestroyList) == 1 {
t.write(fs.DestroyList[0].Name)
} else {
t.write(pruneRuleActionStr)
}
t.newline()
}
2018-08-29 19:18:54 +02:00
}
2018-08-30 15:20:44 +02:00
const snapshotIndent = 1
func calculateMaxFSLength(all []*fsrep.Report) (maxFS, maxStatus int) {
2018-08-30 15:20:44 +02:00
for _, e := range all {
if len(e.Filesystem) > maxFS {
maxFS = len(e.Filesystem)
2018-08-30 15:20:44 +02:00
}
all2 := make([]*fsrep.StepReport, 0, len(e.Pending) + len(e.Completed))
all2 = append(all2, e.Pending...)
all2 = append(all2, e.Completed...)
2018-08-30 15:20:44 +02:00
for _, e2 := range all2 {
elen := len(e2.Problem) + len(e2.From) + len(e2.To) + 60 // random spacing, units, labels, etc
if elen > maxStatus {
maxStatus = elen
2018-08-30 15:20:44 +02:00
}
}
}
return
2018-08-30 15:20:44 +02:00
}
2018-08-29 19:18:54 +02:00
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 {
2018-08-30 14:07:28 +02:00
if len(str) > length {
return str[:length]
}
2018-08-29 19:18:54 +02:00
return str + times(pad, length-len(str))
}
func leftPad(str string, length int, pad string) string {
if len(str) > length {
return str[len(str)-length:]
}
return times(pad, length-len(str)) + str
}
func (t *tui) drawBar(length int, bytes, totalBytes int64) {
var completedLength int
if totalBytes > 0 {
completedLength = int(int64(length) * bytes / totalBytes)
2018-08-30 14:07:28 +02:00
if completedLength > length {
completedLength = length
}
} else if totalBytes == bytes {
completedLength = length
}
2018-08-29 19:18:54 +02:00
t.write("[")
t.write(times("=", completedLength))
t.write(">")
t.write(times("-", length-completedLength))
t.write("]")
2018-08-29 19:18:54 +02:00
}
2018-08-30 14:07:28 +02:00
func StringStepState(s fsrep.StepState) string {
switch s {
case fsrep.StepReplicationReady: return "Ready"
case fsrep.StepReplicationRetry: return "Retry"
case fsrep.StepMarkReplicatedReady: return "MarkReady"
case fsrep.StepMarkReplicatedRetry: return "MarkRetry"
case fsrep.StepPermanentError: return "PermanentError"
case fsrep.StepCompleted: return "Completed"
default:
return fmt.Sprintf("UNKNOWN %d", s)
}
}
func filesystemStatusString(rep *fsrep.Report, active bool, fsWidth int) (line string, bytes, totalBytes int64) {
bytes = int64(0)
totalBytes = int64(0)
2018-08-29 22:06:24 +02:00
for _, s := range rep.Pending {
bytes += s.Bytes
totalBytes += s.ExpectedBytes
2018-08-29 22:06:24 +02:00
}
for _, s := range rep.Completed {
bytes += s.Bytes
totalBytes += s.ExpectedBytes
2018-08-29 22:06:24 +02:00
}
next := ""
if rep.Problem != "" {
next = " problem: " + rep.Problem
} else if len(rep.Pending) > 0 {
if rep.Pending[0].From != "" {
next = fmt.Sprintf(" next: %s => %s", rep.Pending[0].From, rep.Pending[0].To)
} else {
next = fmt.Sprintf(" next: %s (full)", rep.Pending[0].To)
}
}
status := fmt.Sprintf("%s (step %d/%d, %s/%s)%s",
rep.Status,
len(rep.Completed), len(rep.Pending) + len(rep.Completed),
ByteCountBinary(bytes), ByteCountBinary(totalBytes),
next,
)
activeIndicator := " "
if active {
activeIndicator = "*"
}
line = fmt.Sprintf("%s %s %s",
activeIndicator,
rightPad(rep.Filesystem, fsWidth, " "),
status)
return line, bytes, totalBytes
}
func printFilesystemStatus(rep *fsrep.Report, t *tui, active bool, maxFS int) {
totalStatus, _, _ := filesystemStatusString(rep, active, maxFS)
t.write(totalStatus)
t.newline()
2018-08-29 22:06:24 +02:00
}
func ByteCountBinary(b int64) string {
const unit = 1024
if b < unit {
return fmt.Sprintf("%d B", b)
}
div, exp := int64(unit), 0
for n := b / unit; n >= unit; n /= unit {
div *= unit
exp++
}
return fmt.Sprintf("%.1f %ciB", float64(b)/float64(div), "KMGTPE"[exp])
}