Merge branch 'logging-interface-contrib' into master

Author: Shlomi Noach

doc/command-line-flags.md

@@ -18,7 +18,7 @@ If, for some reason, you do not wish `gh-ost` to connect to a replica, you may c
 
 ### approve-renamed-columns
 
-When your migration issues a column rename (`change column old_name new_name ...`) `gh-ost` analyzes the statement to try an associate the old column name with new column name. Otherwise the new structure may also look like some column was dropped and another was added.
+When your migration issues a column rename (`change column old_name new_name ...`) `gh-ost` analyzes the statement to try and associate the old column name with new column name. Otherwise the new structure may also look like some column was dropped and another was added.
 
 `gh-ost` will print out what it thinks the _rename_ implied, but will not issue the migration unless you provide with `--approve-renamed-columns`.
 
@@ -161,7 +161,7 @@ List of metrics and threshold values; topping the threshold of any will cause th
 
 ### migrate-on-replica
 
-Typically `gh-ost` is used to migrate tables on a master. If you wish to only perform the migration in full on a replica, connect `gh-ost` to said replica and pass `--migrate-on-replica`. `gh-ost` will briefly connect to the master but other issue no changes on the master. Migration will be fully executed on the replica, while making sure to maintain a small replication lag.
+Typically `gh-ost` is used to migrate tables on a master. If you wish to only perform the migration in full on a replica, connect `gh-ost` to said replica and pass `--migrate-on-replica`. `gh-ost` will briefly connect to the master but otherwise will make no changes on the master. Migration will be fully executed on the replica, while making sure to maintain a small replication lag.
 
 ### postpone-cut-over-flag-file
 

doc/hooks.md

@@ -65,6 +65,8 @@ The following variables are available on all hooks:
 - `GH_OST_ELAPSED_COPY_SECONDS` - row-copy time (excluding startup, row-count and postpone time)
 - `GH_OST_ESTIMATED_ROWS` - estimated total rows in table
 - `GH_OST_COPIED_ROWS` - number of rows copied by `gh-ost`
+- `GH_OST_INSPECTED_LAG` - lag in seconds (floating point) of inspected server
+- `GH_OST_PROGRESS` - progress pct ([0..100], floating point) of migration
 - `GH_OST_MIGRATED_HOST`
 - `GH_OST_INSPECTED_HOST`
 - `GH_OST_EXECUTING_HOST`

go/base/context.go

@@ -7,7 +7,7 @@ package base
 
 import (
 	"fmt"
-	"github.com/outbrain/golib/log"
+	"math"
 	"os"
 	"regexp"
 	"strings"
@@ -19,6 +19,7 @@ import (
 
 	"github.com/github/gh-ost/go/mysql"
 	"github.com/github/gh-ost/go/sql"
+	"github.com/outbrain/golib/log"
 
 	"gopkg.in/gcfg.v1"
 	gcfgscanner "gopkg.in/gcfg.v1/scanner"
@@ -175,6 +176,7 @@ type MigrationContext struct {
 	pointOfInterestTime                    time.Time
 	pointOfInterestTimeMutex               *sync.Mutex
 	CurrentLag                             int64
+	currentProgress                        uint64
 	ThrottleHTTPStatusCode                 int64
 	controlReplicasLagResult               mysql.ReplicationLagResult
 	TotalRowsCopied                        int64
@@ -449,6 +451,20 @@ func (this *MigrationContext) MarkRowCopyEndTime() {
 	this.RowCopyEndTime = time.Now()
 }
 
+func (this *MigrationContext) GetCurrentLagDuration() time.Duration {
+	return time.Duration(atomic.LoadInt64(&this.CurrentLag))
+}
+
+func (this *MigrationContext) GetProgressPct() float64 {
+	return math.Float64frombits(atomic.LoadUint64(&this.currentProgress))
+}
+
+func (this *MigrationContext) SetProgressPct(progressPct float64) {
+	atomic.StoreUint64(&this.currentProgress, math.Float64bits(progressPct))
+}
+
+// math.Float64bits([f=0..100])
+
 // GetTotalRowsCopied returns the accurate number of rows being copied (affected)
 // This is not exactly the same as the rows being iterated via chunks, but potentially close enough
 func (this *MigrationContext) GetTotalRowsCopied() int64 {

go/logic/hooks.go

@@ -63,6 +63,8 @@ func (this *HooksExecutor) applyEnvironmentVariables(extraVariables ...string) [
 	env = append(env, fmt.Sprintf("GH_OST_MIGRATED_HOST=%s", this.migrationContext.GetApplierHostname()))
 	env = append(env, fmt.Sprintf("GH_OST_INSPECTED_HOST=%s", this.migrationContext.GetInspectorHostname()))
 	env = append(env, fmt.Sprintf("GH_OST_EXECUTING_HOST=%s", this.migrationContext.Hostname))
+	env = append(env, fmt.Sprintf("GH_OST_INSPECTED_LAG=%f", this.migrationContext.GetCurrentLagDuration().Seconds()))
+	env = append(env, fmt.Sprintf("GH_OST_PROGRESS=%f", this.migrationContext.GetProgressPct()))
 	env = append(env, fmt.Sprintf("GH_OST_HOOKS_HINT=%s", this.migrationContext.HooksHintMessage))
 	env = append(env, fmt.Sprintf("GH_OST_HOOKS_HINT_OWNER=%s", this.migrationContext.HooksHintOwner))
 	env = append(env, fmt.Sprintf("GH_OST_HOOKS_HINT_TOKEN=%s", this.migrationContext.HooksHintToken))

go/logic/migrator.go

@@ -893,6 +893,8 @@ func (this *Migrator) printStatus(rule PrintStatusRule, writers ...io.Writer) {
 	} else {
 		progressPct = 100.0 * float64(totalRowsCopied) / float64(rowsEstimate)
 	}
+	// we take the opportunity to update migration context with progressPct
+	this.migrationContext.SetProgressPct(progressPct)
 	// Before status, let's see if we should print a nice reminder for what exactly we're doing here.
 	shouldPrintMigrationStatusHint := (elapsedSeconds%600 == 0)
 	if rule == ForcePrintStatusAndHintRule {
@@ -909,7 +911,7 @@ func (this *Migrator) printStatus(rule PrintStatusRule, writers ...io.Writer) {
 	eta := "N/A"
 	if progressPct >= 100.0 {
 		eta = "due"
-	} else if progressPct >= 1.0 {
+	} else if progressPct >= 0.1 {
 		elapsedRowCopySeconds := this.migrationContext.ElapsedRowCopyTime().Seconds()
 		totalExpectedSeconds := elapsedRowCopySeconds * float64(rowsEstimate) / float64(totalRowsCopied)
 		etaSeconds = totalExpectedSeconds - elapsedRowCopySeconds
@@ -956,12 +958,13 @@ func (this *Migrator) printStatus(rule PrintStatusRule, writers ...io.Writer) {
 
 	currentBinlogCoordinates := *this.eventsStreamer.GetCurrentBinlogCoordinates()
 
-	status := fmt.Sprintf("Copy: %d/%d %.1f%%; Applied: %d; Backlog: %d/%d; Time: %+v(total), %+v(copy); streamer: %+v; State: %s; ETA: %s",
+	status := fmt.Sprintf("Copy: %d/%d %.1f%%; Applied: %d; Backlog: %d/%d; Time: %+v(total), %+v(copy); streamer: %+v; Lag: %.2fs, State: %s; ETA: %s",
 		totalRowsCopied, rowsEstimate, progressPct,
 		atomic.LoadInt64(&this.migrationContext.TotalDMLEventsApplied),
 		len(this.applyEventsQueue), cap(this.applyEventsQueue),
 		base.PrettifyDurationOutput(elapsedTime), base.PrettifyDurationOutput(this.migrationContext.ElapsedRowCopyTime()),
 		currentBinlogCoordinates,
+		this.migrationContext.GetCurrentLagDuration().Seconds(),
 		state,
 		eta,
 	)