ETA based on recent activity

go/logic/progress.go

@@ -0,0 +1,105 @@
+/*
+   Copyright 2016 GitHub Inc.
+	 See https://github.com/github/gh-ost/blob/master/LICENSE
+*/
+
+package logic
+
+import (
+	"sync"
+	"time"
+
+	"github.com/github/gh-ost/go/base"
+)
+
+const maxHistoryDuration time.Duration = time.Hour
+
+type ProgressState struct {
+	mark       time.Time
+	rowsCopied int64
+}
+
+func NewProgressState(rowsCopied int64) *ProgressState {
+	result := &ProgressState{
+		mark:       time.Now(),
+		rowsCopied: rowsCopied,
+	}
+	return result
+}
+
+type ProgressHistory struct {
+	migrationContext *base.MigrationContext
+	history          [](*ProgressState)
+	historyMutex     *sync.Mutex
+}
+
+func NewProgressHistory() *ProgressHistory {
+	result := &ProgressHistory{
+		history:          make([](*ProgressState), 0),
+		historyMutex:     &sync.Mutex{},
+		migrationContext: base.GetMigrationContext(),
+	}
+	return result
+}
+
+func (this *ProgressHistory) oldestState() *ProgressState {
+	if len(this.history) == 0 {
+		return nil
+	}
+	return this.history[0]
+}
+
+func (this *ProgressHistory) newestState() *ProgressState {
+	if len(this.history) == 0 {
+		return nil
+	}
+	return this.history[len(this.history)-1]
+}
+
+func (this *ProgressHistory) oldestMark() (mark time.Time) {
+	if oldest := this.oldestState(); oldest != nil {
+		return oldest.mark
+	}
+	return mark
+}
+
+func (this *ProgressHistory) markState() {
+	this.historyMutex.Lock()
+	defer this.historyMutex.Unlock()
+
+	state := NewProgressState(this.migrationContext.GetTotalRowsCopied())
+	this.history = append(this.history, state)
+	for time.Since(this.oldestMark()) > maxHistoryDuration {
+		if len(this.history) == 0 {
+			return
+		}
+		this.history = this.history[1:]
+	}
+}
+
+// hasEnoughData tells us whether there's at all enough information to predict an ETA
+// this function is not concurrent-safe
+func (this *ProgressHistory) hasEnoughData() bool {
+	oldest := this.oldestState()
+	if oldest == nil {
+		return false
+	}
+	newest := this.newestState()
+
+	if !oldest.mark.Before(newest.mark) {
+		// single point in time; cannot extrapolate
+		return false
+	}
+	if oldest.rowsCopied == newest.rowsCopied {
+		// Nothing really happened; cannot extrapolate
+		return false
+	}
+	return true
+}
+
+func (this *ProgressHistory) getETA() (eta time.Time) {
+	if !this.hasEnoughData() {
+		return eta
+	}
+	return eta
+}

go/logic/progress_test.go

@@ -0,0 +1,59 @@
+/*
+   Copyright 2016 GitHub Inc.
+	 See https://github.com/github/gh-ost/blob/master/LICENSE
+*/
+
+package logic
+
+import (
+	"testing"
+	"time"
+
+	"github.com/outbrain/golib/log"
+	test "github.com/outbrain/golib/tests"
+)
+
+func init() {
+	log.SetLevel(log.ERROR)
+}
+
+func TestNewProgressHistory(t *testing.T) {
+	progressHistory := NewProgressHistory()
+	test.S(t).ExpectEquals(len(progressHistory.history), 0)
+}
+
+func TestMarkState(t *testing.T) {
+	{
+		progressHistory := NewProgressHistory()
+		test.S(t).ExpectEquals(len(progressHistory.history), 0)
+	}
+	{
+		progressHistory := NewProgressHistory()
+		progressHistory.markState()
+		progressHistory.markState()
+		progressHistory.markState()
+		progressHistory.markState()
+		progressHistory.markState()
+		test.S(t).ExpectEquals(len(progressHistory.history), 5)
+	}
+	{
+		progressHistory := NewProgressHistory()
+		progressHistory.markState()
+		progressHistory.markState()
+		progressHistory.history[0].mark = time.Now().Add(-2 * time.Hour)
+		progressHistory.markState()
+		progressHistory.markState()
+		progressHistory.markState()
+		test.S(t).ExpectEquals(len(progressHistory.history), 4)
+	}
+}
+
+func TestOldestMark(t *testing.T) {
+	{
+		progressHistory := NewProgressHistory()
+		oldestState := progressHistory.oldestState()
+		test.S(t).ExpectTrue(oldestState == nil)
+		oldestMark := progressHistory.oldestMark()
+		test.S(t).ExpectTrue(oldestMark.IsZero())
+	}
+}