inject dummy statements when vreplication transaction_timestamp is stale

shlomi-noach · shlomi-noach · commit 7e2a0748a50d · 2021-02-16T17:05:54.000+02:00
Signed-off-by: Shlomi Noach &lt;2607934+shlomi-noach@users.noreply.github.com&gt;
diff --git a/go/vt/vttablet/onlineddl/executor.go b/go/vt/vttablet/onlineddl/executor.go
@@ -140,6 +140,8 @@ type Executor struct {
 	lastMigrationUUID     string
 	tickReentranceFlag    int64
 
+	requestForDummyInjection int64
+
 	ticks             *timer.Timer
 	isOpen            bool
 	schemaInitialized bool
@@ -387,6 +389,24 @@ func (e *Executor) tableExists(ctx context.Context, tableName string) (bool, err
 	return (row != nil), nil
 }
 
+// injectDummyStatements issues several no-op statements on the backend MySQL server. This is done to
+// ensure binary logs are being written, to solve a possible scenario for vreplication-based migrations
+// where the server is otherwise completely stale. Vreplication does not write heartbeats like gh-ost does,
+// and if the server is stale (and assuming lag-throttler is not enabled) then binary logs may be completely silent,
+// which makes it impossible to know when to cut-over.
+func (e *Executor) injectDummyStatements(ctx context.Context) (err error) {
+	conn, err := dbconnpool.NewDBConnection(ctx, e.env.Config().DB.DbaWithDB())
+	if err != nil {
+		return err
+	}
+	defer conn.Close()
+
+	if _, err := conn.ExecuteFetch(sqlDummyDropView, 0, false); err != nil {
+		return err
+	}
+	return nil
+}
+
 func (e *Executor) parseAlterOptions(ctx context.Context, onlineDDL *schema.OnlineDDL) string {
 	// Temporary hack (2020-08-11)
 	// Because sqlparser does not do full blown ALTER TABLE parsing,
@@ -1525,19 +1545,26 @@ func (e *Executor) isVReplMigrationReadyToCutOver(ctx context.Context, s *VReplS
 			}
 			return diff
 		}
+		timeNow := time.Now()
 		timeUpdated := time.Unix(s.timeUpdated, 0)
-		if durationDiff(time.Now(), timeUpdated) > cutOverThreshold {
+		if durationDiff(timeNow, timeUpdated) > cutOverThreshold {
 			return false, nil
 		}
 		// Let's look at transaction timestamp. This gets written by any ongoing
 		// writes on the server (whether on this table or any other table)
 		transactionTimestamp := time.Unix(s.transactionTimestamp, 0)
-		if durationDiff(transactionTimestamp, timeUpdated) > cutOverThreshold {
+		if diff := durationDiff(timeNow, transactionTimestamp); diff > cutOverThreshold {
 			// There's two ways the diff can be high:
 			// 1. High workload: vreplication is unable to apply events in a timely manner. This is
 			//    a normal situation and is why we're testing transaction_timestamp
 			// 2. Lack of writes on the server. Possibly the server is stale. In which case, cut-over
 			//    should be good to go. But we need (TODO) some mechanism to inform us that this is indeed the case.
+
+			// To handle (2), we request an injection of a dummy transaction, heuristically in good time for the next check.
+			// To not overdo it, and avoid spamming the server with dummy statements, we only request injection if lag is very high.
+			if diff > *migrationCheckInterval {
+				atomic.StoreInt64(&e.requestForDummyInjection, 1)
+			}
 			return false, nil
 		}
 	}
@@ -1793,6 +1820,18 @@ func (e *Executor) onMigrationCheckTick() {
 		log.Error(err)
 		return
 	}
+
+	if atomic.LoadInt64(&e.requestForDummyInjection) > 0 {
+		// This is a special scenario. VReplication-based online-DDL notices transaction timestamp is old. Either it
+		// is very busy, or the server is compeltely stale. In th elatter case, it requests dummy injection of
+		// statements. We comply here.
+		// We inject a few well-timed statements, scheduled to be fresh by *next iteration*, to be intercepted by e.reviewRunningMigrations(ctx), following
+		atomic.StoreInt64(&e.requestForDummyInjection, 0)
+		intervals := []time.Duration{*migrationCheckInterval - 2*time.Second, *migrationCheckInterval - time.Second, *migrationCheckInterval}
+		for _, d := range intervals {
+			time.AfterFunc(d, func() { e.injectDummyStatements(ctx) })
+		}
+	}
 	if err := e.retryTabletFailureMigrations(ctx); err != nil {
 		log.Error(err)
 	}
diff --git a/go/vt/vttablet/onlineddl/schema.go b/go/vt/vttablet/onlineddl/schema.go
@@ -245,7 +245,8 @@ const (
 			_vt.copy_state
 		WHERE vrepl_id=%a
 		`
-	sqlSwapTables = "RENAME TABLE `%a` TO `%a`, `%a` TO `%a`, `%a` TO `%a`"
+	sqlSwapTables    = "RENAME TABLE `%a` TO `%a`, `%a` TO `%a`, `%a` TO `%a`"
+	sqlDummyDropView = "DROP VIEW IF EXISTS `_vt:onlineddl:executor:inject:binlog:entry`"
 )
 
 const (

Original file line number	Diff line number	Diff line change
`@@ -245,7 +245,8 @@ const (`
`245`	`245`	`_vt.copy_state`
`246`	`246`	`WHERE vrepl_id=%a`
`247`	`247`	`
`248`		- sqlSwapTables = "RENAME TABLE `%a` TO `%a`, `%a` TO `%a`, `%a` TO `%a`"
	`248`	+ sqlSwapTables = "RENAME TABLE `%a` TO `%a`, `%a` TO `%a`, `%a` TO `%a`"
	`249`	+ sqlDummyDropView = "DROP VIEW IF EXISTS `_vt:onlineddl:executor:inject:binlog:entry`"
`249`	`250`	`)`
`250`	`251`
`251`	`252`	`const (`