hooks: reporting GH_OST_ETA_SECONDS. ETA stored as part of migration context
Signed-off-by: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
This commit is contained in:
parent
2b5cf78b4d
commit
54000ab516
@ -182,6 +182,7 @@ type MigrationContext struct {
|
|||||||
lastHeartbeatOnChangelogMutex *sync.Mutex
|
lastHeartbeatOnChangelogMutex *sync.Mutex
|
||||||
CurrentLag int64
|
CurrentLag int64
|
||||||
currentProgress uint64
|
currentProgress uint64
|
||||||
|
etaNanoseonds int64
|
||||||
ThrottleHTTPStatusCode int64
|
ThrottleHTTPStatusCode int64
|
||||||
controlReplicasLagResult mysql.ReplicationLagResult
|
controlReplicasLagResult mysql.ReplicationLagResult
|
||||||
TotalRowsCopied int64
|
TotalRowsCopied int64
|
||||||
@ -474,6 +475,14 @@ func (this *MigrationContext) SetProgressPct(progressPct float64) {
|
|||||||
atomic.StoreUint64(&this.currentProgress, math.Float64bits(progressPct))
|
atomic.StoreUint64(&this.currentProgress, math.Float64bits(progressPct))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (this *MigrationContext) GetETADuration() time.Duration {
|
||||||
|
return time.Duration(atomic.LoadInt64(&this.etaNanoseonds))
|
||||||
|
}
|
||||||
|
|
||||||
|
func (this *MigrationContext) SetETADuration(etaDuration time.Duration) {
|
||||||
|
atomic.StoreInt64(&this.etaNanoseonds, etaDuration.Nanoseconds())
|
||||||
|
}
|
||||||
|
|
||||||
// math.Float64bits([f=0..100])
|
// math.Float64bits([f=0..100])
|
||||||
|
|
||||||
// GetTotalRowsCopied returns the accurate number of rows being copied (affected)
|
// GetTotalRowsCopied returns the accurate number of rows being copied (affected)
|
||||||
|
@ -66,6 +66,7 @@ func (this *HooksExecutor) applyEnvironmentVariables(extraVariables ...string) [
|
|||||||
env = append(env, fmt.Sprintf("GH_OST_INSPECTED_LAG=%f", this.migrationContext.GetCurrentLagDuration().Seconds()))
|
env = append(env, fmt.Sprintf("GH_OST_INSPECTED_LAG=%f", this.migrationContext.GetCurrentLagDuration().Seconds()))
|
||||||
env = append(env, fmt.Sprintf("GH_OST_HEARTBEAT_LAG=%f", this.migrationContext.TimeSinceLastHeartbeatOnChangelog().Seconds()))
|
env = append(env, fmt.Sprintf("GH_OST_HEARTBEAT_LAG=%f", this.migrationContext.TimeSinceLastHeartbeatOnChangelog().Seconds()))
|
||||||
env = append(env, fmt.Sprintf("GH_OST_PROGRESS=%f", this.migrationContext.GetProgressPct()))
|
env = append(env, fmt.Sprintf("GH_OST_PROGRESS=%f", this.migrationContext.GetProgressPct()))
|
||||||
|
env = append(env, fmt.Sprintf("GH_OST_ETA_SECONDS=%f", this.migrationContext.GetETADuration().Seconds()))
|
||||||
env = append(env, fmt.Sprintf("GH_OST_HOOKS_HINT=%s", this.migrationContext.HooksHintMessage))
|
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_OWNER=%s", this.migrationContext.HooksHintOwner))
|
||||||
env = append(env, fmt.Sprintf("GH_OST_HOOKS_HINT_TOKEN=%s", this.migrationContext.HooksHintToken))
|
env = append(env, fmt.Sprintf("GH_OST_HOOKS_HINT_TOKEN=%s", this.migrationContext.HooksHintToken))
|
||||||
|
@ -939,20 +939,29 @@ func (this *Migrator) printStatus(rule PrintStatusRule, writers ...io.Writer) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
var etaSeconds float64 = math.MaxFloat64
|
var etaSeconds float64 = math.MaxFloat64
|
||||||
eta := "N/A"
|
var etaDuration = time.Duration(math.MaxInt64)
|
||||||
if progressPct >= 100.0 {
|
if progressPct >= 100.0 {
|
||||||
eta = "due"
|
etaDuration = 0
|
||||||
} else if progressPct >= 0.1 {
|
} else if progressPct >= 0.1 {
|
||||||
elapsedRowCopySeconds := this.migrationContext.ElapsedRowCopyTime().Seconds()
|
elapsedRowCopySeconds := this.migrationContext.ElapsedRowCopyTime().Seconds()
|
||||||
totalExpectedSeconds := elapsedRowCopySeconds * float64(rowsEstimate) / float64(totalRowsCopied)
|
totalExpectedSeconds := elapsedRowCopySeconds * float64(rowsEstimate) / float64(totalRowsCopied)
|
||||||
etaSeconds = totalExpectedSeconds - elapsedRowCopySeconds
|
etaSeconds = totalExpectedSeconds - elapsedRowCopySeconds
|
||||||
if etaSeconds >= 0 {
|
if etaSeconds >= 0 {
|
||||||
etaDuration := time.Duration(etaSeconds) * time.Second
|
etaDuration = time.Duration(etaSeconds) * time.Second
|
||||||
eta = base.PrettifyDurationOutput(etaDuration)
|
|
||||||
} else {
|
} else {
|
||||||
eta = "due"
|
etaDuration = 0
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
this.migrationContext.SetETADuration(etaDuration)
|
||||||
|
var eta string
|
||||||
|
switch etaDuration {
|
||||||
|
case 0:
|
||||||
|
eta = "due"
|
||||||
|
case time.Duration(math.MaxInt64):
|
||||||
|
eta = "N/A"
|
||||||
|
default:
|
||||||
|
eta = base.PrettifyDurationOutput(etaDuration)
|
||||||
|
}
|
||||||
|
|
||||||
state := "migrating"
|
state := "migrating"
|
||||||
if atomic.LoadInt64(&this.migrationContext.CountingRowsFlag) > 0 && !this.migrationContext.ConcurrentCountTableRows {
|
if atomic.LoadInt64(&this.migrationContext.CountingRowsFlag) > 0 && !this.migrationContext.ConcurrentCountTableRows {
|
||||||
|
Loading…
Reference in New Issue
Block a user