vrepl: more documentation and fix stats

go/vt/binlog/binlogplayer/binlog_player.go

@@ -79,6 +79,8 @@ type Stats struct {
 
 	SecondsBehindMaster sync2.AtomicInt64
 	History             *history.History
+
+	State sync2.AtomicString
 }
 
 // SetLastPosition sets the last replication position.
@@ -176,17 +178,12 @@ func NewBinlogPlayerTables(dbClient DBClient, tablet *topodatapb.Tablet, tables
 // If an error is encountered, it updates the vreplication state to "Error".
 // If a stop position was specified, and reached, the state is updated to "Stopped".
 func (blp *BinlogPlayer) ApplyBinlogEvents(ctx context.Context) error {
-	if err := SetVReplicationState(blp.dbClient, blp.uid, BlpRunning, ""); err != nil {
+	if err := blp.setVReplicationState(BlpRunning, ""); err != nil {
 		log.Errorf("Error writing Running state: %v", err)
 	}
 
 	if err := blp.applyEvents(ctx); err != nil {
-		msg := err.Error()
-		blp.blplStats.History.Add(&StatsHistoryRecord{
-			Time:    time.Now(),
-			Message: msg,
-		})
-		if err := SetVReplicationState(blp.dbClient, blp.uid, BlpError, msg); err != nil {
+		if err := blp.setVReplicationState(BlpError, err.Error()); err != nil {
 			log.Errorf("Error writing stop state: %v", err)
 		}
 		return err
@@ -241,14 +238,14 @@ func (blp *BinlogPlayer) applyEvents(ctx context.Context) error {
 		case blp.position.Equal(blp.stopPosition):
 			msg := fmt.Sprintf("not starting BinlogPlayer, we're already at the desired position %v", blp.stopPosition)
 			log.Info(msg)
-			if err := SetVReplicationState(blp.dbClient, blp.uid, BlpStopped, msg); err != nil {
+			if err := blp.setVReplicationState(BlpStopped, msg); err != nil {
 				log.Errorf("Error writing stop state: %v", err)
 			}
 			return nil
 		case blp.position.AtLeast(blp.stopPosition):
 			msg := fmt.Sprintf("starting point %v greater than stopping point %v", blp.position, blp.stopPosition)
 			log.Error(msg)
-			if err := SetVReplicationState(blp.dbClient, blp.uid, BlpStopped, msg); err != nil {
+			if err := blp.setVReplicationState(BlpStopped, msg); err != nil {
 				log.Errorf("Error writing stop state: %v", err)
 			}
 			// Don't return an error. Otherwise, it will keep retrying.
@@ -348,7 +345,7 @@ func (blp *BinlogPlayer) applyEvents(ctx context.Context) error {
 					if blp.position.AtLeast(blp.stopPosition) {
 						msg := "Reached stopping position, done playing logs"
 						log.Info(msg)
-						if err := SetVReplicationState(blp.dbClient, blp.uid, BlpStopped, msg); err != nil {
+						if err := blp.setVReplicationState(BlpStopped, msg); err != nil {
 							log.Errorf("Error writing stop state: %v", err)
 						}
 						return nil
@@ -463,6 +460,21 @@ func (blp *BinlogPlayer) writeRecoveryPosition(tx *binlogdatapb.BinlogTransactio
 	return nil
 }
 
+func (blp *BinlogPlayer) setVReplicationState(state, message string) error {
+	if message != "" {
+		blp.blplStats.History.Add(&StatsHistoryRecord{
+			Time:    time.Now(),
+			Message: message,
+		})
+	}
+	blp.blplStats.State.Set(state)
+	query := fmt.Sprintf("update _vt.vreplication set state='%v', message=%v where id=%v", state, encodeString(MessageTruncate(message)), blp.uid)
+	if _, err := blp.dbClient.ExecuteFetch(query, 1); err != nil {
+		return fmt.Errorf("could not set state: %v: %v", query, err)
+	}
+	return nil
+}
+
 // CreateVReplicationTable returns the statements required to create
 // the _vt.vreplication table.
 // id: is an auto-increment column that identifies the stream.
@@ -507,15 +519,6 @@ func AlterVReplicationTable() []string {
 	return []string{"ALTER TABLE _vt.vreplication ADD COLUMN db_name VARBINARY(255) NOT NULL"}
 }
 
-// SetVReplicationState updates the state in the _vt.vreplication table.
-func SetVReplicationState(dbClient DBClient, uid uint32, state, message string) error {
-	query := fmt.Sprintf("update _vt.vreplication set state='%v', message=%v where id=%v", state, encodeString(MessageTruncate(message)), uid)
-	if _, err := dbClient.ExecuteFetch(query, 1); err != nil {
-		return fmt.Errorf("could not set state: %v: %v", query, err)
-	}
-	return nil
-}
-
 // VRSettings contains the settings of a vreplication table.
 type VRSettings struct {
 	StartPos          mysql.Position

go/vt/vttablet/tabletmanager/vreplication/controller.go

@@ -90,6 +90,7 @@ func newController(ctx context.Context, params map[string]string, dbClientFactor
 	ct.id = uint32(id)
 	ct.workflow = params["workflow"]
 
+	blpStats.State.Set(params["state"])
 	// Nothing to do if replication is stopped.
 	if params["state"] == binlogplayer.BlpStopped {
 		ct.cancel = func() {}

go/vt/vttablet/tabletmanager/vreplication/engine.go

@@ -405,6 +405,17 @@ func (vre *Engine) fetchIDs(dbClient binlogplayer.DBClient, selector string) (id
 	return ids, bv, nil
 }
 
+// registerJounral is invoked if any of the vreplication streams encounters a journal event.
+// Multiple registerJournal functions collaborate to converge on the final action.
+// The first invocation creates an entry in vre.journaler. The entry is initialized
+// with the list of participants that also need to converge.
+// The middle invocation happens on the first and subsequent calls: the current participant
+// marks itself as having joined the wait.
+// The final invocation happens for the last participant that joins. Having confirmed
+// that all the participants have joined, transitionJournal is invoked, which deletes
+// all current participant streams and creates new ones to replace them.
+// A unified journal event is identified by the workflow name and journal id.
+// Multiple independent journal events can go through this cycle concurrently.
 func (vre *Engine) registerJournal(journal *binlogdatapb.Journal, id int) error {
 	vre.mu.Lock()
 	defer vre.mu.Unlock()
@@ -417,6 +428,7 @@ func (vre *Engine) registerJournal(journal *binlogdatapb.Journal, id int) error
 	key := fmt.Sprintf("%s:%d", workflow, journal.Id)
 	je, ok := vre.journaler[key]
 	if !ok {
+		// First invocation. Create the entry.
 		log.Infof("Journal encountered: %v", journal)
 		controllerSources := make(map[string]bool)
 		for _, ct := range vre.controllers {
@@ -441,19 +453,25 @@ func (vre *Engine) registerJournal(journal *binlogdatapb.Journal, id int) error
 		vre.journaler[key] = je
 	}
 
+	// Middle invocation. Register yourself
 	ks := fmt.Sprintf("%s:%s", vre.controllers[id].source.Keyspace, vre.controllers[id].source.Shard)
 	log.Infof("Registering id %v against %v", id, ks)
 	je.participants[ks] = id
+	// Check if all participants have joined.
 	for _, pid := range je.participants {
 		if pid == 0 {
 			// Still need to wait.
 			return nil
 		}
 	}
+
+	// Final invocation. Perform the transition.
 	go vre.transitionJournal(key)
 	return nil
 }
 
+// transitionJournal stops all existing participants, deletes their vreplication
+// entries, and creates new ones as instructed by the journal metadata.
 func (vre *Engine) transitionJournal(key string) {
 	vre.mu.Lock()
 	defer vre.mu.Unlock()
@@ -484,6 +502,7 @@ func (vre *Engine) transitionJournal(key string) {
 		return
 	}
 
+	// Use the reference row to copy other fields like cell, tablet_types, etc.
 	params, err := readRow(dbClient, refid)
 	if err != nil {
 		log.Errorf("transitionJournal: %v", err)

go/vt/vttablet/tabletmanager/vreplication/framework_test.go

@@ -216,12 +216,18 @@ func (ftc *fakeTabletConn) StreamHealth(ctx context.Context, callback func(*quer
 	})
 }
 
+// vstreamHook allows you to do work just before calling VStream.
+var vstreamHook func(ctx context.Context)
+
 // VStream directly calls into the pre-initialized engine.
 func (ftc *fakeTabletConn) VStream(ctx context.Context, target *querypb.Target, startPos string, filter *binlogdatapb.Filter, send func([]*binlogdatapb.VEvent) error) error {
 	if target.Keyspace != "vttest" {
 		<-ctx.Done()
 		return io.EOF
 	}
+	if vstreamHook != nil {
+		vstreamHook(ctx)
+	}
 	return streamerEngine.Stream(ctx, startPos, filter, send)
 }
 

go/vt/vttablet/tabletmanager/vreplication/replicator_plan.go

@@ -29,12 +29,19 @@ import (
 	querypb "vitess.io/vitess/go/vt/proto/query"
 )
 
-// ReplicatorPlan is the execution plan for the replicator.
-// The constructor for this is buildReplicatorPlan in table_plan_builder.go
-// The initial build identifies the tables that need to be replicated,
-// and builds partial TablePlan objects for them. The partial plan is used
-// to send streaming requests. As the responses return field info, this
-// information is used to build the final execution plan (buildExecutionPlan).
+// ReplicatorPlan is the execution plan for the replicator. It contains
+// plans for all the tables it's replicating. Every phase of vreplication
+// builds its own instance of the ReplicatorPlan. This is because the plan
+// depends on copyState, which changes on every iteration.
+// The table plans within ReplicatorPlan will not be fully populated because
+// all the information is not available initially.
+// For simplicity, the ReplicatorPlan is immutable.
+// Once we get the field info for a table from the stream response,
+// we'll have all the necessary info to build the final plan.
+// At that time, buildExecutionPlan is invoked, which will make a copy
+// of the TablePlan from ReplicatorPlan, and fill the rest
+// of the members, leaving the original plan unchanged.
+// The constructor is buildReplicatorPlan in table_plan_builder.go
 type ReplicatorPlan struct {
 	VStreamFilter *binlogdatapb.Filter
 	TargetTables  map[string]*TablePlan
@@ -50,6 +57,8 @@ func (rp *ReplicatorPlan) buildExecutionPlan(fieldEvent *binlogdatapb.FieldEvent
 		// Unreachable code.
 		return nil, fmt.Errorf("plan not found for %s", fieldEvent.TableName)
 	}
+	// If Insert is initialized, then it means that we knew the column
+	// names and have already built most of the plan.
 	if prelim.Insert != nil {
 		tplanv := *prelim
 		// We know that we sent only column names, but they may be backticked.
@@ -93,6 +102,7 @@ func (rp *ReplicatorPlan) buildFromFields(tableName string, lastpk *sqltypes.Res
 		}
 		tpb.colExprs = append(tpb.colExprs, cexpr)
 	}
+	// The following actions are a subset of buildTablePlan.
 	if err := tpb.analyzePK(rp.tableKeys); err != nil {
 		return nil, err
 	}
@@ -118,19 +128,35 @@ func (rp *ReplicatorPlan) MarshalJSON() ([]byte, error) {
 	return json.Marshal(&v)
 }
 
-// TablePlan is the execution plan for a table within a player stream.
+// TablePlan is the execution plan for a table within a replicator.
+// If the column names are not known at the time of plan building (like
+// select *), then only TargetName, SendRule and Lastpk are initialized.
+// When the stream returns the field info, those are used as column
+// names to build the final plan.
+// Lastpk comes from copyState. If it's set, then the generated plans
+// are significantly different because any events that fall beyond
+// Lastpk must be excluded.
+// If column names were known upfront, then all fields of TablePlan
+// are built except for Fields. This member is populated only after
+// the field info is received from the stream.
 // The ParsedQuery objects assume that a map of before and after values
 // will be built based on the streaming rows. Before image values will
 // be prefixed with a "b_", and after image values will be prefixed
-// with a "a_".
+// with a "a_". The TablePlan structure is used during all the phases
+// of vreplication: catchup, copy, fastforward, or regular replication.
 type TablePlan struct {
-	TargetName   string
-	SendRule     *binlogdatapb.Rule
-	PKReferences []string
-	// Lastpk is used for delayed generation of replication queries.
-	Lastpk *sqltypes.Result
+	// TargetName, SendRule will always be initialized.
+	// Lastpk will also be initialized if it was specified, and
+	// will be used for building the final plan after field info
+	// is received.
+	TargetName string
+	SendRule   *binlogdatapb.Rule
+	Lastpk     *sqltypes.Result
 	// BulkInsertFront, BulkInsertValues and BulkInsertOnDup are used
-	// by vcopier.
+	// by vcopier. These three parts are combined to build bulk insert
+	// statements. This is functionally equivalent to generating
+	// multiple statements using the "Insert" construct, but much more
+	// efficient for the copy phase.
 	BulkInsertFront  *sqlparser.ParsedQuery
 	BulkInsertValues *sqlparser.ParsedQuery
 	BulkInsertOnDup  *sqlparser.ParsedQuery
@@ -142,30 +168,33 @@ type TablePlan struct {
 	Update *sqlparser.ParsedQuery
 	Delete *sqlparser.ParsedQuery
 	Fields []*querypb.Field
+	// PKReferences is used to check if an event changed
+	// a primary key column (row move).
+	PKReferences []string
 }
 
 // MarshalJSON performs a custom JSON Marshalling.
 func (tp *TablePlan) MarshalJSON() ([]byte, error) {
 	v := struct {
 		TargetName   string
 		SendRule     string
-		PKReferences []string               `json:",omitempty"`
 		InsertFront  *sqlparser.ParsedQuery `json:",omitempty"`
 		InsertValues *sqlparser.ParsedQuery `json:",omitempty"`
 		InsertOnDup  *sqlparser.ParsedQuery `json:",omitempty"`
 		Insert       *sqlparser.ParsedQuery `json:",omitempty"`
 		Update       *sqlparser.ParsedQuery `json:",omitempty"`
 		Delete       *sqlparser.ParsedQuery `json:",omitempty"`
+		PKReferences []string               `json:",omitempty"`
 	}{
 		TargetName:   tp.TargetName,
 		SendRule:     tp.SendRule.Match,
-		PKReferences: tp.PKReferences,
 		InsertFront:  tp.BulkInsertFront,
 		InsertValues: tp.BulkInsertValues,
 		InsertOnDup:  tp.BulkInsertOnDup,
 		Insert:       tp.Insert,
 		Update:       tp.Update,
 		Delete:       tp.Delete,
+		PKReferences: tp.PKReferences,
 	}
 	return json.Marshal(&v)
 }

go/vt/vttablet/tabletmanager/vreplication/replicator_plan_test.go

@@ -35,13 +35,13 @@ type TestReplicatorPlan struct {
 type TestTablePlan struct {
 	TargetName   string
 	SendRule     string
-	PKReferences []string `json:",omitempty"`
 	InsertFront  string   `json:",omitempty"`
 	InsertValues string   `json:",omitempty"`
 	InsertOnDup  string   `json:",omitempty"`
 	Insert       string   `json:",omitempty"`
 	Update       string   `json:",omitempty"`
 	Delete       string   `json:",omitempty"`
+	PKReferences []string `json:",omitempty"`
 }
 
 func TestBuildPlayerPlan(t *testing.T) {

go/vt/vttablet/tabletmanager/vreplication/stats.go

@@ -148,12 +148,6 @@ func (st *vrStats) status() *EngineStatus {
 	status.Controllers = make([]*ControllerStatus, len(st.controllers))
 	i := 0
 	for _, ct := range st.controllers {
-		state := "Running"
-		select {
-		case <-ct.done:
-			state = "Stopped"
-		default:
-		}
 		status.Controllers[i] = &ControllerStatus{
 			Index:               ct.id,
 			Source:              ct.source.String(),
@@ -162,7 +156,7 @@ func (st *vrStats) status() *EngineStatus {
 			SecondsBehindMaster: ct.blpStats.SecondsBehindMaster.Get(),
 			Counts:              ct.blpStats.Timings.Counts(),
 			Rates:               ct.blpStats.Rates.Get(),
-			State:               state,
+			State:               ct.blpStats.State.Get(),
 			SourceTablet:        ct.sourceTablet.Get(),
 			Messages:            ct.blpStats.MessageHistory(),
 		}