dumpling: add recent speed to dump status

dumpling/export/dump.go

@@ -55,6 +55,8 @@ type Dumper struct {
 	selectTiDBTableRegionFunc     func(tctx *tcontext.Context, conn *BaseConn, meta TableMeta) (pkFields []string, pkVals [][]string, err error)
 	totalTables                   int64
 	charsetAndDefaultCollationMap map[string]string
+
+	statusRecorder *StatusRecorder
 }
 
 // NewDumper returns a new Dumper
@@ -78,6 +80,7 @@ func NewDumper(ctx context.Context, conf *Config) (*Dumper, error) {
 		conf:                      conf,
 		cancelCtx:                 cancelFn,
 		selectTiDBTableRegionFunc: selectTiDBTableRegion,
+		statusRecorder:            NewStatusRecorder(),
 	}
 
 	var err error

dumpling/export/status.go

@@ -4,6 +4,7 @@ package export
 
 import (
 	"fmt"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -48,6 +49,7 @@ type DumpStatus struct {
 	FinishedRows      float64
 	EstimateTotalRows float64
 	TotalTables       int64
+	CurrentSpeedBPS   int64
 }
 
 // GetStatus returns the status of dumping by reading metrics.
@@ -58,6 +60,7 @@ func (d *Dumper) GetStatus() *DumpStatus {
 	ret.FinishedBytes = ReadGauge(d.metrics.finishedSizeGauge)
 	ret.FinishedRows = ReadGauge(d.metrics.finishedRowsGauge)
 	ret.EstimateTotalRows = ReadCounter(d.metrics.estimateTotalRowsCounter)
+	ret.CurrentSpeedBPS = d.statusRecorder.GetSpeed(int64(ret.FinishedBytes))
 	return ret
 }
 
@@ -72,3 +75,46 @@ func calculateTableCount(m DatabaseTables) int {
 	}
 	return cnt
 }
+
+// StatusRecorder record the finished bytes and calculate its speed.
+type StatusRecorder struct {
+	mu             sync.Mutex
+	lastFinished   int64
+	lastUpdateTime time.Time
+	speedBPS       int64
+}
+
+// NewStatusRecorder new a StatusRecorder.
+func NewStatusRecorder() *StatusRecorder {
+	return &StatusRecorder{
+		lastUpdateTime: time.Now(),
+	}
+}
+
+// GetSpeed calculate status speed.
+func (s *StatusRecorder) GetSpeed(finished int64) int64 {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if finished <= s.lastFinished {
+		// for finished bytes does not get forwarded, use old speed to avoid
+		// display zero. We may find better strategy in future.
+		return s.speedBPS
+	}
+
+	now := time.Now()
+	elapsed := int64(now.Sub(s.lastUpdateTime).Seconds())
+	if elapsed == 0 {
+		elapsed = 1
+	}
+	currentSpeed := (finished - s.lastFinished) / elapsed
+	if currentSpeed == 0 {
+		currentSpeed = 1
+	}
+
+	s.lastFinished = finished
+	s.lastUpdateTime = now
+	s.speedBPS = currentSpeed
+
+	return currentSpeed
+}

dumpling/export/status_test.go

@@ -10,7 +10,7 @@ import (
 
 func TestGetParameters(t *testing.T) {
 	conf := defaultConfigForTest(t)
-	d := &Dumper{conf: conf}
+	d := &Dumper{conf: conf, statusRecorder: NewStatusRecorder()}
 	d.metrics = newMetrics(conf.PromFactory, nil)
 
 	mid := d.GetStatus()