Execute hook on every batch insert retry

Co-authored-by: Bastian Bartmann <bastian.bartmann@shopify.com>

Author: grodowski

go/logic/hooks.go

@@ -28,6 +28,7 @@ const (
 	onInteractiveCommand = "gh-ost-on-interactive-command"
 	onSuccess            = "gh-ost-on-success"
 	onFailure            = "gh-ost-on-failure"
+	onBatchCopyRetry     = "gh-ost-on-batch-copy-retry"
 	onStatus             = "gh-ost-on-status"
 	onStopReplication    = "gh-ost-on-stop-replication"
 	onStartReplication   = "gh-ost-on-start-replication"
@@ -77,6 +78,7 @@ func (this *HooksExecutor) applyEnvironmentVariables(extraVariables ...string) [
 // executeHook executes a command, and sets relevant environment variables
 // combined output & error are printed to the configured writer.
 func (this *HooksExecutor) executeHook(hook string, extraVariables ...string) error {
+	this.migrationContext.Log.Infof("executing hook: %+v", hook)
 	cmd := exec.Command(hook)
 	cmd.Env = this.applyEnvironmentVariables(extraVariables...)
 
@@ -123,6 +125,10 @@ func (this *HooksExecutor) onBeforeRowCopy() error {
 	return this.executeHooks(onBeforeRowCopy)
 }
 
+func (this *HooksExecutor) onBatchCopyRetry() error {
+	return this.executeHooks(onBatchCopyRetry)
+}
+
 func (this *HooksExecutor) onRowCopyComplete() error {
 	return this.executeHooks(onRowCopyComplete)
 }

go/logic/migrator.go

@@ -130,6 +130,18 @@ func (this *Migrator) sleepWhileTrue(operation func() (bool, error)) error {
 	}
 }
 
+func (this *Migrator) retryBatchCopyWithHooks(operation func() error, notFatalHint ...bool) (err error) {
+	wrappedOperation := func() error {
+		if err := operation(); err != nil {
+			this.hooksExecutor.onBatchCopyRetry()
+			return err
+		}
+		return nil
+	}
+
+	return this.retryOperation(wrappedOperation, notFatalHint...)
+}
+
 // retryOperation attempts up to `count` attempts at running given function,
 // exiting as soon as it returns with non-error.
 func (this *Migrator) retryOperation(operation func() error, notFatalHint ...bool) (err error) {
@@ -1232,28 +1244,28 @@ func (this *Migrator) iterateChunks() error {
 			return nil
 		}
 		copyRowsFunc := func() error {
-			if atomic.LoadInt64(&this.rowCopyCompleteFlag) == 1 || atomic.LoadInt64(&hasNoFurtherRangeFlag) == 1 {
-				// Done.
-				// There's another such check down the line
-				return nil
-			}
-
-			// When hasFurtherRange is false, original table might be write locked and CalculateNextIterationRangeEndValues would hangs forever
-
-			hasFurtherRange := false
-			expectedRangeSize := int64(0)
-			if err := this.retryOperation(func() (e error) {
-				hasFurtherRange, expectedRangeSize, e = this.applier.CalculateNextIterationRangeEndValues()
-				return e
-			}); err != nil {
-				return terminateRowIteration(err)
-			}
-			if !hasFurtherRange {
-				atomic.StoreInt64(&hasNoFurtherRangeFlag, 1)
-				return terminateRowIteration(nil)
-			}
 			// Copy task:
 			applyCopyRowsFunc := func() error {
+				if atomic.LoadInt64(&this.rowCopyCompleteFlag) == 1 || atomic.LoadInt64(&hasNoFurtherRangeFlag) == 1 {
+					// Done.
+					// There's another such check down the line
+					return nil
+				}
+
+				// When hasFurtherRange is false, original table might be write locked and CalculateNextIterationRangeEndValues would hangs forever
+
+				hasFurtherRange := false
+				// TODO: figure out how to rewrite this double retry?
+				if err := this.retryOperation(func() (e error) {
+					hasFurtherRange, e = this.applier.CalculateNextIterationRangeEndValues()
+					return e
+				}); err != nil {
+					return terminateRowIteration(err)
+				}
+				if !hasFurtherRange {
+					atomic.StoreInt64(&hasNoFurtherRangeFlag, 1)
+					return terminateRowIteration(nil)
+				}
 				if atomic.LoadInt64(&this.rowCopyCompleteFlag) == 1 {
 					// No need for more writes.
 					// This is the de-facto place where we avoid writing in the event of completed cut-over.
@@ -1286,7 +1298,7 @@ func (this *Migrator) iterateChunks() error {
 				atomic.AddInt64(&this.migrationContext.Iteration, 1)
 				return nil
 			}
-			if err := this.retryOperation(applyCopyRowsFunc); err != nil {
+			if err := this.retryBatchCopyWithHooks(applyCopyRowsFunc); err != nil {
 				return terminateRowIteration(err)
 			}
 			return nil