diff --git a/go/base/context.go b/go/base/context.go index f12cd43..d523274 100644 --- a/go/base/context.go +++ b/go/base/context.go @@ -736,3 +736,20 @@ func (this *MigrationContext) ReadConfigFile() error { return nil } + +func (this *MigrationContext) PanicAbortIfTableError(err error) { + if err == nil { + return + } + if strings.Contains(err.Error(), mysql.Error1146TableDoesntExist) || strings.Contains(err.Error(), mysql.Error1017CantFindFile) { + this.PanicAbortOnError(err) + } + // otherwise irrelevant error and we do not panic +} + +func (this *MigrationContext) PanicAbortOnError(err error) { + if err == nil { + return + } + this.PanicAbort <- err +} diff --git a/go/logic/applier.go b/go/logic/applier.go index 5fb795b..bc48242 100644 --- a/go/logic/applier.go +++ b/go/logic/applier.go @@ -916,6 +916,8 @@ func (this *Applier) AtomicCutoverRename(sessionIdChan chan int64, tablesRenamed ) log.Infof("Issuing and expecting this to block: %s", query) if _, err := tx.Exec(query); err != nil { + this.migrationContext.PanicAbortIfTableError(err) + tablesRenamed <- err return log.Errore(err) } diff --git a/go/logic/migrator.go b/go/logic/migrator.go index 86de5ac..533ecf1 100644 --- a/go/logic/migrator.go +++ b/go/logic/migrator.go @@ -144,7 +144,7 @@ func (this *Migrator) retryOperation(operation func() error, notFatalHint ...boo // there's an error. Let's try again. } if len(notFatalHint) == 0 { - this.migrationContext.PanicAbort <- err + this.migrationContext.PanicAbortOnError(err) } return err } @@ -172,7 +172,7 @@ func (this *Migrator) retryOperationWithExponentialBackoff(operation func() erro } } if len(notFatalHint) == 0 { - this.migrationContext.PanicAbort <- err + this.migrationContext.PanicAbortOnError(err) } return err } @@ -191,14 +191,14 @@ func (this *Migrator) executeAndThrottleOnError(operation func() error) (err err // consumes and drops any further incoming events that may be left hanging. func (this *Migrator) consumeRowCopyComplete() { if err := <-this.rowCopyComplete; err != nil { - this.migrationContext.PanicAbort <- err + this.migrationContext.PanicAbortOnError(err) } atomic.StoreInt64(&this.rowCopyCompleteFlag, 1) this.migrationContext.MarkRowCopyEndTime() go func() { for err := range this.rowCopyComplete { if err != nil { - this.migrationContext.PanicAbort <- err + this.migrationContext.PanicAbortOnError(err) } } }() @@ -620,10 +620,12 @@ func (this *Migrator) atomicCutOver() (err error) { tableUnlocked := make(chan error, 2) go func() { if err := this.applier.AtomicCutOverMagicLock(lockOriginalSessionIdChan, tableLocked, okToUnlockTable, tableUnlocked); err != nil { + this.migrationContext.PanicAbortIfTableError(err) log.Errore(err) } }() if err := <-tableLocked; err != nil { + this.migrationContext.PanicAbortIfTableError(err) return log.Errore(err) } lockOriginalSessionId := <-lockOriginalSessionIdChan @@ -631,6 +633,7 @@ func (this *Migrator) atomicCutOver() (err error) { // At this point we know the original table is locked. // We know any newly incoming DML on original table is blocked. if err := this.waitForEventsUpToLock(); err != nil { + this.migrationContext.PanicAbortIfTableError(err) return log.Errore(err) } @@ -644,6 +647,7 @@ func (this *Migrator) atomicCutOver() (err error) { go func() { if err := this.applier.AtomicCutoverRename(renameSessionIdChan, tablesRenamed); err != nil { // Abort! Release the lock + this.migrationContext.PanicAbortIfTableError(err) atomic.StoreInt64(&tableRenameKnownToHaveFailed, 1) okToUnlockTable <- true } @@ -996,9 +1000,8 @@ func (this *Migrator) initiateStreaming() error { go func() { log.Debugf("Beginning streaming") - err := this.eventsStreamer.StreamEvents(this.canStopStreaming) - if err != nil { - this.migrationContext.PanicAbort <- err + if err := this.eventsStreamer.StreamEvents(this.canStopStreaming); err != nil { + this.migrationContext.PanicAbortOnError(err) } log.Debugf("Done streaming") }() diff --git a/go/logic/server.go b/go/logic/server.go index 774c4ab..7c50d80 100644 --- a/go/logic/server.go +++ b/go/logic/server.go @@ -342,7 +342,7 @@ help # This message return NoPrintStatusRule, err } err := fmt.Errorf("User commanded 'panic'. I will now panic, without cleanup. PANIC!") - this.migrationContext.PanicAbort <- err + this.migrationContext.PanicAbortOnError(err) return NoPrintStatusRule, err } default: diff --git a/go/logic/throttler.go b/go/logic/throttler.go index 6f9f4bb..7e0df4f 100644 --- a/go/logic/throttler.go +++ b/go/logic/throttler.go @@ -209,6 +209,7 @@ func (this *Throttler) collectControlReplicasLag() { lagResult := &mysql.ReplicationLagResult{Key: connectionConfig.Key} go func() { lagResult.Lag, lagResult.Err = readReplicaLag(connectionConfig) + this.migrationContext.PanicAbortIfTableError(lagResult.Err) lagResults <- lagResult }() } @@ -317,7 +318,7 @@ func (this *Throttler) collectGeneralThrottleMetrics() error { // Regardless of throttle, we take opportunity to check for panic-abort if this.migrationContext.PanicFlagFile != "" { if base.FileExists(this.migrationContext.PanicFlagFile) { - this.migrationContext.PanicAbort <- fmt.Errorf("Found panic-file %s. Aborting without cleanup", this.migrationContext.PanicFlagFile) + this.migrationContext.PanicAbortOnError(fmt.Errorf("Found panic-file %s. Aborting without cleanup", this.migrationContext.PanicFlagFile)) } } @@ -340,7 +341,7 @@ func (this *Throttler) collectGeneralThrottleMetrics() error { } if criticalLoadMet && this.migrationContext.CriticalLoadIntervalMilliseconds == 0 { - this.migrationContext.PanicAbort <- fmt.Errorf("critical-load met: %s=%d, >=%d", variableName, value, threshold) + this.migrationContext.PanicAbortOnError(fmt.Errorf("critical-load met: %s=%d, >=%d", variableName, value, threshold)) } if criticalLoadMet && this.migrationContext.CriticalLoadIntervalMilliseconds > 0 { log.Errorf("critical-load met once: %s=%d, >=%d. Will check again in %d millis", variableName, value, threshold, this.migrationContext.CriticalLoadIntervalMilliseconds) @@ -348,7 +349,7 @@ func (this *Throttler) collectGeneralThrottleMetrics() error { timer := time.NewTimer(time.Millisecond * time.Duration(this.migrationContext.CriticalLoadIntervalMilliseconds)) <-timer.C if criticalLoadMetAgain, variableName, value, threshold, _ := this.criticalLoadIsMet(); criticalLoadMetAgain { - this.migrationContext.PanicAbort <- fmt.Errorf("critical-load met again after %d millis: %s=%d, >=%d", this.migrationContext.CriticalLoadIntervalMilliseconds, variableName, value, threshold) + this.migrationContext.PanicAbortOnError(fmt.Errorf("critical-load met again after %d millis: %s=%d, >=%d", this.migrationContext.CriticalLoadIntervalMilliseconds, variableName, value, threshold)) } }() } diff --git a/go/mysql/utils.go b/go/mysql/utils.go index 17bb5fc..a14eb94 100644 --- a/go/mysql/utils.go +++ b/go/mysql/utils.go @@ -18,6 +18,11 @@ import ( "github.com/outbrain/golib/sqlutils" ) +const ( + Error1017CantFindFile = "Error 1017:" + Error1146TableDoesntExist = "Error 1146:" +) + const MaxTableNameLength = 64 const MaxReplicationPasswordLength = 32