PBM-1534 progress indicators

cmd/pbm/backup.go

@@ -20,6 +20,7 @@ import (
 	"github.com/percona/percona-backup-mongodb/pbm/defs"
 	"github.com/percona/percona-backup-mongodb/pbm/errors"
 	"github.com/percona/percona-backup-mongodb/pbm/log"
+	"github.com/percona/percona-backup-mongodb/pbm/progress"
 	"github.com/percona/percona-backup-mongodb/pbm/storage"
 	"github.com/percona/percona-backup-mongodb/pbm/topo"
 	"github.com/percona/percona-backup-mongodb/pbm/util"
@@ -388,6 +389,11 @@ type bcpDesc struct {
 	StorageType        storage.Type    `json:"storage_type,omitempty" yaml:"storage_type,omitempty"`
 	OPID               string          `json:"opid" yaml:"opid"`
 	Type               defs.BackupType `json:"type" yaml:"type"`
+	StartTS            int64           `json:"start_ts" yaml:"-"`
+	StartTime          string          `json:"start" yaml:"start"`
+	FinishTime         *string         `json:"finish,omitempty" yaml:"finish,omitempty"`
+	Duration           int64           `json:"duration" yaml:"-"`
+	DurationH          string          `json:"duration_h" yaml:"duration"`
 	LastWriteTS        int64           `json:"last_write_ts" yaml:"-"`
 	LastTransitionTS   int64           `json:"last_transition_ts" yaml:"-"`
 	LastWriteTime      string          `json:"last_write_time" yaml:"last_write_time"`
@@ -484,6 +490,8 @@ func describeBackup(
 		StorageType:        bcp.Store.Type,
 		OPID:               bcp.OPID,
 		Type:               bcp.Type,
+		StartTS:            bcp.StartTS,
+		StartTime:          time.Unix(bcp.StartTS, 0).UTC().Format(time.RFC3339),
 		Namespaces:         bcp.Namespaces,
 		SelUserAndRoles:    bcp.SelUsersAndRoles,
 		MongoVersion:       bcp.MongoVersion,
@@ -499,6 +507,11 @@ func describeBackup(
 		SizeUncompressed:   bcp.SizeUncompressed,
 		HSizeUncompressed:  byteCountIEC(bcp.SizeUncompressed),
 	}
+	rv.Duration = operationDurationSeconds(bcp.Status, bcp.StartTS, bcp.LastTransitionTS)
+	rv.DurationH = progress.FormatDuration(time.Duration(rv.Duration) * time.Second)
+	if isTerminalStatus(bcp.Status) {
+		rv.FinishTime = util.Ref(time.Unix(bcp.LastTransitionTS, 0).UTC().Format(time.RFC3339))
+	}
 	if bcp.SizeUncompressed > 0 {
 		rv.HSizeUncompressed = byteCountIEC(bcp.SizeUncompressed)
 	}

cmd/pbm/main.go

@@ -1202,6 +1202,7 @@
 	Err         error            `json:"-"`
 	ErrString   string           `json:"error,omitempty"`
 	RestoreTS   int64            `json:"restoreTo"`
+	Duration    int64            `json:"duration,omitempty"`
 	PBMVersion  string           `json:"pbmVersion"`
 	Type        defs.BackupType  `json:"type"`
 	SrcBackup   string           `json:"src"`
@@ -1224,6 +1225,24 @@
 	return strings.TrimSuffix(t, "Z")
 }
 
+func isTerminalStatus(status defs.Status) bool {
+	return !status.IsRunning() || status == defs.StatusPartlyDone
+}
+
+func operationDurationSeconds(status defs.Status, start, transition int64) int64 {
+	if start <= 0 {
+		return 0
+	}
+	end := transition
+	if !isTerminalStatus(status) {
+		end = time.Now().Unix()
+	}
+	if end < start {
+		return 0
+	}
+	return end - start
+}
+
 type outMsg struct {
 	Msg string `json:"msg"`
 }
@@ -1244,7 +1263,7 @@
 func (c outCaption) MarshalJSON() ([]byte, error) {
 	var b bytes.Buffer
 	b.WriteString("{")
 	b.WriteString(fmt.Sprintf("\"%s\":", c.k))
 	err := json.NewEncoder(&b).Encode(c.v)
 	if err != nil {
 		return nil, err

cmd/pbm/restore.go

@@ -21,6 +21,7 @@ import (
 	"github.com/percona/percona-backup-mongodb/pbm/defs"
 	"github.com/percona/percona-backup-mongodb/pbm/errors"
 	"github.com/percona/percona-backup-mongodb/pbm/log"
+	"github.com/percona/percona-backup-mongodb/pbm/progress"
 	"github.com/percona/percona-backup-mongodb/pbm/restore"
 	"github.com/percona/percona-backup-mongodb/pbm/storage"
 	"github.com/percona/percona-backup-mongodb/pbm/topo"
@@ -728,6 +729,8 @@ type describeRestoreResult struct {
 	StartTS            *int64           `json:"start_ts,omitempty" yaml:"-"`
 	StartTime          *string          `json:"start,omitempty" yaml:"start,omitempty"`
 	FinishTime         *string          `json:"finish,omitempty" yaml:"finish,omitempty"`
+	Duration           int64            `json:"duration,omitempty" yaml:"-"`
+	DurationH          string           `json:"duration_h,omitempty" yaml:"duration,omitempty"`
 	PITR               *int64           `json:"ts_to_restore,omitempty" yaml:"-"`
 	PITRTime           *string          `json:"time_to_restore,omitempty" yaml:"time_to_restore,omitempty"`
 	LastTransitionTS   int64            `json:"last_transition_ts" yaml:"-"`
@@ -804,7 +807,9 @@ func describeRestore(
 	res.LastTransitionTS = meta.LastTransitionTS
 	res.LastTransitionTime = time.Unix(res.LastTransitionTS, 0).UTC().Format(time.RFC3339)
 	res.StartTime = util.Ref(time.Unix(meta.StartTS, 0).UTC().Format(time.RFC3339))
-	if meta.Status == defs.StatusDone {
+	res.Duration = operationDurationSeconds(meta.Status, meta.StartTS, meta.LastTransitionTS)
+	res.DurationH = progress.FormatDuration(time.Duration(res.Duration) * time.Second)
+	if isTerminalStatus(meta.Status) {
 		res.FinishTime = util.Ref(time.Unix(meta.LastTransitionTS, 0).UTC().Format(time.RFC3339))
 	}
 	if meta.Status == defs.StatusError {

cmd/pbm/status.go

@@ -19,6 +19,8 @@ import (
 	"github.com/percona/percona-backup-mongodb/pbm/errors"
 	"github.com/percona/percona-backup-mongodb/pbm/log"
 	"github.com/percona/percona-backup-mongodb/pbm/oplog"
+	"github.com/percona/percona-backup-mongodb/pbm/progress"
+	"github.com/percona/percona-backup-mongodb/pbm/restore"
 	"github.com/percona/percona-backup-mongodb/pbm/slicer"
 	"github.com/percona/percona-backup-mongodb/pbm/storage"
 	"github.com/percona/percona-backup-mongodb/pbm/topo"
@@ -372,11 +374,13 @@ LOOP:
 }
 
 type currOp struct {
-	Type    ctrl.Command `json:"type,omitempty"`
-	OPID    string       `json:"opID,omitempty"`
-	Name    string       `json:"name,omitempty"`
-	StartTS int64        `json:"startTS,omitempty"`
-	Status  string       `json:"status,omitempty"`
+	Type     ctrl.Command       `json:"type,omitempty"`
+	OPID     string             `json:"opID,omitempty"`
+	Name     string             `json:"name,omitempty"`
+	StartTS  int64              `json:"startTS,omitempty"`
+	Duration int64              `json:"duration,omitempty"`
+	Status   string             `json:"status,omitempty"`
+	Progress *progress.Progress `json:"progress,omitempty"`
 }
 
 func (c currOp) String() string {
@@ -388,10 +392,14 @@ func (c currOp) String() string {
 	default:
 		return fmt.Sprintf("%s [op id: %s]", c.Type, c.OPID)
 	case ctrl.CmdBackup, ctrl.CmdRestore:
-		return fmt.Sprintf("%s \"%s\", started at %s. Status: %s. [op id: %s]",
+		s := fmt.Sprintf("%s \"%s\", started at %s. Status: %s. Duration: %s. [op id: %s]",
 			c.Type, c.Name, time.Unix((c.StartTS), 0).UTC().Format("2006-01-02T15:04:05Z"),
-			c.Status, c.OPID,
+			c.Status, progress.FormatDuration(time.Duration(c.Duration)*time.Second), c.OPID,
 		)
+		if c.Progress != nil {
+			s += ". Progress: " + c.Progress.StringAt(time.Now().Unix())
+		}
+		return s
 	}
 }
 
@@ -418,6 +426,8 @@ func getCurrOps(ctx context.Context, pbm *sdk.Client) (fmt.Stringer, error) {
 
 		r.Name = bcp.Name
 		r.StartTS = bcp.StartTS
+		r.Duration = durationSeconds(bcp.StartTS, 0)
+		r.Progress = backupProgress(bcp)
 
 		switch bcp.Status {
 		case defs.StatusRunning:
@@ -435,6 +445,8 @@ func getCurrOps(ctx context.Context, pbm *sdk.Client) (fmt.Stringer, error) {
 
 		r.Name = rst.Backup
 		r.StartTS = rst.StartTS
+		r.Duration = durationSeconds(rst.StartTS, 0)
+		r.Progress = restoreProgress(rst)
 
 		switch rst.Status {
 		case defs.StatusRunning:
@@ -449,6 +461,73 @@ func getCurrOps(ctx context.Context, pbm *sdk.Client) (fmt.Stringer, error) {
 	return r, nil
 }
 
+func backupProgress(bcp *backup.BackupMeta) *progress.Progress {
+	p := combineProgress(bcp.Progress, func(yield func(*progress.Progress)) {
+		for i := range bcp.Replsets {
+			yield(bcp.Replsets[i].Progress)
+		}
+	})
+	return p
+}
+
+func restoreProgress(rst *restore.RestoreMeta) *progress.Progress {
+	p := combineProgress(rst.Progress, func(yield func(*progress.Progress)) {
+		for i := range rst.Replsets {
+			yield(rst.Replsets[i].Progress)
+		}
+	})
+	return p
+}
+
+func combineProgress(fallback *progress.Progress, each func(func(*progress.Progress))) *progress.Progress {
+	var rv *progress.Progress
+	each(func(p *progress.Progress) {
+		if p == nil {
+			return
+		}
+		if rv == nil {
+			cp := *p
+			rv = &cp
+			return
+		}
+		if p.StartedAt > 0 && (rv.StartedAt == 0 || p.StartedAt < rv.StartedAt) {
+			rv.StartedAt = p.StartedAt
+		}
+		if p.UpdatedAt > rv.UpdatedAt {
+			rv.UpdatedAt = p.UpdatedAt
+		}
+		rv.DoneBytes += p.DoneBytes
+		rv.TotalBytes += p.TotalBytes
+		rv.DoneItems += p.DoneItems
+		rv.TotalItems += p.TotalItems
+	})
+	if rv != nil {
+		return rv
+	}
+	return fallback
+}
+
+func durationSeconds(start, end int64) int64 {
+	if start <= 0 {
+		return 0
+	}
+	if end <= 0 {
+		end = time.Now().Unix()
+	}
+	if end < start {
+		return 0
+	}
+	return end - start
+}
+
+func backupDuration(bcp backup.BackupMeta, now int64) int64 {
+	end := bcp.LastTransitionTS
+	if bcp.Status.IsRunning() {
+		end = now
+	}
+	return durationSeconds(bcp.StartTS, end)
+}
+
 type storageStat struct {
 	Type     string         `json:"type"`
 	Path     string         `json:"path"`
@@ -482,9 +561,9 @@ func (s storageStat) String() string {
 		return a.RestoreTS > b.RestoreTS
 	})
 
-	ret += fmt.Sprintf("  %-24s  %-10s  %-12s  %-20s  %-5s  %-4s  %-19s  %s\n",
-		"NAME", "SIZE", "TYPE", "PROFILE", "SEL", "BASE", "RESTORE TIME", "STATUS")
-	ret += fmt.Sprintf("  %s\n", strings.Repeat("-", 24+10+12+20+5+4+19+6+(7*2)))
+	ret += fmt.Sprintf("  %-24s  %-10s  %-12s  %-20s  %-5s  %-4s  %-19s  %-10s  %s\n",
+		"NAME", "SIZE", "TYPE", "PROFILE", "SEL", "BASE", "RESTORE TIME", "DURATION", "STATUS")
+	ret += fmt.Sprintf("  %s\n", strings.Repeat("-", 24+10+12+20+5+4+19+10+6+(8*2)))
 
 	for i := range s.Snapshot {
 		ss := &s.Snapshot[i]
@@ -523,14 +602,15 @@ func (s storageStat) String() string {
 			status = strings.TrimRight(status[:maxStatusLen-3], " ") + "..."
 		}
 
-		ret += fmt.Sprintf("  %-24s  %-10s  %-12s  %-20s  %-5s  %-4s  %-19s  %s\n",
+		ret += fmt.Sprintf("  %-24s  %-10s  %-12s  %-20s  %-5s  %-4s  %-19s  %-10s  %s\n",
 			ss.Name,
 			storage.PrettySize(ss.Size),
 			bcpType,
 			profile,
 			selective,
 			base,
 			fmtTS(ss.RestoreTS),
+			progress.FormatDuration(time.Duration(ss.Duration)*time.Second),
 			status)
 	}
 
@@ -625,6 +705,7 @@ func getStorageStat(
 			SrcBackup:  bcp.SrcBackup,
 			Profile:    bcp.Store.Name,
 			StoreName:  bcp.Store.Name,
+			Duration:   backupDuration(bcp, int64(now.T)),
 		}
 		if err := bcp.Error(); err != nil {
 			snpsht.Err = err

pbm/backup/backup.go

@@ -17,6 +17,7 @@
 	"github.com/percona/percona-backup-mongodb/pbm/lock"
 	"github.com/percona/percona-backup-mongodb/pbm/log"
 	"github.com/percona/percona-backup-mongodb/pbm/oplog"
+	"github.com/percona/percona-backup-mongodb/pbm/progress"
 	"github.com/percona/percona-backup-mongodb/pbm/storage"
 	"github.com/percona/percona-backup-mongodb/pbm/topo"
 	"github.com/percona/percona-backup-mongodb/pbm/util"
@@ -185,6 +186,16 @@
 //
 //nolint:nonamedreturns
 func (b *Backup) Run(ctx context.Context, bcp *ctrl.BackupCmd, opid ctrl.OPID, l log.LogEvent) (err error) {
+	opStarted := time.Now()
+	defer func() {
+		elapsed := progress.FormatDuration(time.Since(opStarted))
+		if err != nil {
+			l.Info("backup failed after %s: %v", elapsed, err)
+			return
+		}
+		l.Info("backup completed after %s", elapsed)
+	}()
+
 	inf, err := topo.GetNodeInfoExt(ctx, b.nodeConn)
 	if err != nil {
 		return errors.Wrap(err, "get cluster info")
@@ -384,6 +395,7 @@
 		// PBM-1114: update file metadata with the same values as in database
 		unix := time.Now().Unix()
 		bcpm.Status = defs.StatusDone
+		bcpm.Progress = nil
 		bcpm.LastTransitionTS = unix
 		bcpm.Conditions = append(bcpm.Conditions, Condition{
 			Timestamp: unix,
@@ -549,7 +561,7 @@
 		for _, shard := range bmeta.Replsets {
 			if shard.Name == sh.RS {
 				// Check if node is alive.
 				// Nodes in terminal states (done, error, cancelled) may have
 				// already released their locks. Their status is handled by
 				// the switch below.
 				if shard.Status.IsRunning() {

pbm/backup/logical.go

@@ -20,6 +20,7 @@ import (
 	"github.com/percona/percona-backup-mongodb/pbm/defs"
 	"github.com/percona/percona-backup-mongodb/pbm/errors"
 	"github.com/percona/percona-backup-mongodb/pbm/log"
+	"github.com/percona/percona-backup-mongodb/pbm/progress"
 	"github.com/percona/percona-backup-mongodb/pbm/snapshot"
 	"github.com/percona/percona-backup-mongodb/pbm/storage"
 	"github.com/percona/percona-backup-mongodb/pbm/topo"
@@ -47,6 +48,7 @@ func (b *Backup) doLogical(
 	}
 
 	sizeHints := make(map[string]int64, len(nssSize))
+	totalSizeHint := int64(0)
 	for ns, cs := range nssSize {
 		if bcp.Compression == compress.CompressionTypeNone {
 			// Uncompressed dump: the output size matches the logical BSON size.
@@ -55,6 +57,7 @@ func (b *Backup) doLogical(
 			// Compressed: WiredTiger on-disk size approximates compressed output.
 			sizeHints[ns] = cs.StorageSize
 		}
+		totalSizeHint += sizeHints[ns]
 	}
 
 	rsMeta.Status = defs.StatusRunning
@@ -64,6 +67,11 @@ func (b *Backup) doLogical(
 	if err != nil {
 		return errors.Wrap(err, "add shard's metadata")
 	}
+	reporter := progress.NewReporter(ctx, l, time.Minute, totalSizeHint, int64(len(nssSize)),
+		func(ctx context.Context, p progress.Progress) error {
+			return SetRSProgress(ctx, b.leadConn, bcp.Name, rsMeta.Name, p)
+		})
+	defer reporter.Close("backup transfer finished")
 
 	if inf.IsLeader() {
 		err := b.reconcileStatus(ctx,
@@ -178,7 +186,7 @@ func (b *Backup) doLogical(
 		}
 	}
 
-	snapshotSize, err := snapshot.UploadDump(ctx,
+	snapshotSize, err := snapshot.UploadDumpWithProgress(ctx,
 		func(newFile archive.NewWriter) error {
 			bcp, err := archive.NewBackup(ctx, archive.BackupOptions{
 				Client:        b.nodeConn,
@@ -202,10 +210,19 @@ func (b *Backup) doLogical(
 			return stg.Save(filepath, r, storage.Size(sizeHints[ns]))
 		},
 		bcp.Compression,
-		bcp.CompressionLevel)
+		bcp.CompressionLevel,
+		func(ns string, bytes int64, done bool) {
+			reporter.AddBytes(bytes)
+			if done && ns != archive.MetaFileV2 {
+				reporter.AddItems(1)
+			}
+		})
 	if err != nil {
 		return errors.Wrap(err, "dump")
 	}
+	if err := reporter.Flush(); err != nil {
+		l.Warning("update progress: %v", err)
+	}
 
 	err = archive.GenerateV1FromV2(ctx, stg, bcp.Name, rsMeta.Name)
 	if err != nil {