worker-stats command

Author: meiji163

go/logic/coordinator.go

@@ -30,13 +30,17 @@ type Coordinator struct {
 	// Atomic counter for number of active workers
 	busyWorkers atomic.Int64
 
+	// Mutex protecting currentCoordinates
 	currentCoordinatesMutex sync.Mutex
 	// The binlog coordinates of the low water mark transaction.
 	currentCoordinates mysql.BinlogCoordinates
 
 	// Mutex to protect the fields below
 	mu sync.Mutex
 
+	// list of workers
+	workers []*Worker
+
 	// The low water mark. This is the sequence number of the last job that has been committed.
 	lowWaterMark int64
 
@@ -56,13 +60,25 @@ type Coordinator struct {
 	finishedMigrating atomic.Bool
 }
 
+// Worker takes jobs from the Coordinator and applies the job's DML events.
 type Worker struct {
 	id          int
 	coordinator *Coordinator
+	eventQueue  chan *replication.BinlogEvent
 
-	executedJobs int
+	executedJobs     atomic.Int64
+	dmlEventsApplied atomic.Int64
+	waitTimeNs       atomic.Int64
+	busyTimeNs       atomic.Int64
+}
 
-	eventQueue chan *replication.BinlogEvent
+type stats struct {
+	dmlRate          float64
+	trxRate          float64
+	dmlEventsApplied int64
+	executedJobs     int64
+	busyTime         time.Duration
+	waitTime         time.Duration
 }
 
 func (w *Worker) ProcessEvents() error {
@@ -86,21 +102,15 @@ func (w *Worker) ProcessEvents() error {
 		// Wait for conditions to be met
 		waitChannel := w.coordinator.WaitForTransaction(gtidEvent.LastCommitted)
 		if waitChannel != nil {
-			//fmt.Printf("Worker %d - transaction %d waiting for transaction: %d\n", w.id, gtidEvent.SequenceNumber, gtidEvent.LastCommitted)
-			t := time.Now()
+			waitStart := time.Now()
 			<-waitChannel
-			timeWaited := time.Since(t)
-			w.coordinator.migrationContext.Log.Infof(
-				"Worker %d waited for transaction %d for: %d\n",
-				w.id, gtidEvent.LastCommitted, timeWaited)
+			timeWaited := time.Since(waitStart)
+			w.waitTimeNs.Add(timeWaited.Nanoseconds())
 		}
 
 		// Process the transaction
-
 		var changelogEvent *binlog.BinlogDMLEvent
-
 		dmlEvents := make([]*binlog.BinlogDMLEvent, 0, int(atomic.LoadInt64(&w.coordinator.migrationContext.DMLBatchSize)))
-
 	events:
 		for {
 			ev := <-w.eventQueue
@@ -166,21 +176,21 @@ func (w *Worker) ProcessEvents() error {
 						dmlEvents = append(dmlEvents, dmlEvent)
 
 						if len(dmlEvents) == cap(dmlEvents) {
-							err := w.coordinator.applier.ApplyDMLEventQueries(dmlEvents)
-							if err != nil {
-								//TODO(meiji163) add retry
+							if err := w.applyDMLEvents(dmlEvents); err != nil {
 								w.coordinator.migrationContext.Log.Errore(err)
-								w.coordinator.migrationContext.PanicAbort <- err
 							}
 							dmlEvents = dmlEvents[:0]
 						}
 					}
 				}
 			case *replication.XIDEvent:
 				if len(dmlEvents) > 0 {
-					w.coordinator.applier.ApplyDMLEventQueries(dmlEvents)
+					if err := w.applyDMLEvents(dmlEvents); err != nil {
+						w.coordinator.migrationContext.Log.Errore(err)
+					}
 				}
 
+				w.executedJobs.Add(1)
 				break events
 			}
 		}
@@ -190,19 +200,33 @@ func (w *Worker) ProcessEvents() error {
 		// Did we see a changelog event?
 		// Handle it now
 		if changelogEvent != nil {
+			// wait for all transactions before this point
 			waitChannel = w.coordinator.WaitForTransaction(gtidEvent.SequenceNumber - 1)
 			if waitChannel != nil {
+				waitStart := time.Now()
 				<-waitChannel
+				w.waitTimeNs.Add(time.Since(waitStart).Nanoseconds())
 			}
 			w.coordinator.HandleChangeLogEvent(changelogEvent)
 		}
 
-		w.executedJobs += 1
 		w.coordinator.workerQueue <- w
 		w.coordinator.busyWorkers.Add(-1)
 	}
 }
 
+func (w *Worker) applyDMLEvents(dmlEvents []*binlog.BinlogDMLEvent) error {
+	busyStart := time.Now()
+	err := w.coordinator.applier.ApplyDMLEventQueries(dmlEvents)
+	if err != nil {
+		//TODO(meiji163) add retry
+		return err
+	}
+	w.busyTimeNs.Add(time.Since(busyStart).Nanoseconds())
+	w.dmlEventsApplied.Add(int64(len(dmlEvents)))
+	return nil
+}
+
 func NewCoordinator(migrationContext *base.MigrationContext, applier *Applier, onChangelogEvent func(dmlEvent *binlog.BinlogDMLEvent) error) *Coordinator {
 	connectionConfig := migrationContext.InspectorConnectionConfig
 
@@ -338,7 +362,6 @@ func (c *Coordinator) ProcessEventsUntilDrained() error {
 				if c.finishedMigrating.Load() {
 					return nil
 				}
-				//				c.migrationContext.Log.Infof("Received event: %T - %+v", ev.Event, ev.Event)
 
 				switch binlogEvent := ev.Event.(type) {
 				case *replication.GTIDEvent:
@@ -358,13 +381,10 @@ func (c *Coordinator) ProcessEventsUntilDrained() error {
 				worker := <-c.workerQueue
 				c.busyWorkers.Add(1)
 
-				// c.migrationContext.Log.Infof("Submitting job %d to worker", ev.Event.(*replication.GTIDEvent).SequenceNumber)
 				worker.eventQueue <- ev
 
 				ev = <-c.events
 
-				// c.migrationContext.Log.Infof("Received event: %T - %+v", ev.Event, ev.Event)
-
 				switch binlogEvent := ev.Event.(type) {
 				case *replication.QueryEvent:
 					if bytes.Equal([]byte("BEGIN"), binlogEvent.Query) {
@@ -381,9 +401,6 @@ func (c *Coordinator) ProcessEventsUntilDrained() error {
 			events:
 				for {
 					ev = <-c.events
-
-					// c.migrationContext.Log.Infof("Received event: %T - %+v", ev.Event, ev.Event)
-
 					switch ev.Event.(type) {
 					case *replication.RowsEvent:
 						worker.eventQueue <- ev
@@ -402,8 +419,6 @@ func (c *Coordinator) ProcessEventsUntilDrained() error {
 				busyWorkers := c.busyWorkers.Load()
 				if busyWorkers == 0 {
 					return nil
-				} else {
-					//c.migrationContext.Log.Infof("%d/%d workers are busy\n", busyWorkers, cap(c.workerQueue))
 				}
 			}
 		}
@@ -414,11 +429,35 @@ func (c *Coordinator) InitializeWorkers(count int) {
 	c.workerQueue = make(chan *Worker, count)
 	for i := 0; i < count; i++ {
 		w := &Worker{id: i, coordinator: c, eventQueue: make(chan *replication.BinlogEvent, 1000)}
+
+		c.mu.Lock()
+		c.workers = append(c.workers, w)
+		c.mu.Unlock()
+
 		c.workerQueue <- w
 		go w.ProcessEvents()
 	}
 }
 
+func (c *Coordinator) GetWorkerStats() []stats {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	statSlice := make([]stats, 0, len(c.workers))
+	for _, w := range c.workers {
+		stat := stats{}
+		stat.dmlEventsApplied = w.dmlEventsApplied.Load()
+		stat.executedJobs = w.executedJobs.Load()
+		stat.busyTime = time.Duration(w.busyTimeNs.Load())
+		stat.waitTime = time.Duration(w.waitTimeNs.Load())
+		if stat.busyTime.Milliseconds() > 0 {
+			stat.dmlRate = 1000.0 * float64(stat.dmlEventsApplied) / float64(stat.busyTime.Milliseconds())
+			stat.trxRate = 1000.0 * float64(stat.executedJobs) / float64(stat.busyTime.Milliseconds())
+		}
+		statSlice = append(statSlice, stat)
+	}
+	return statSlice
+}
+
 func (c *Coordinator) WaitForTransaction(lastCommitted int64) chan struct{} {
 	c.mu.Lock()
 	defer c.mu.Unlock()
@@ -438,11 +477,8 @@ func (c *Coordinator) WaitForTransaction(lastCommitted int64) chan struct{} {
 }
 
 func (c *Coordinator) HandleChangeLogEvent(event *binlog.BinlogDMLEvent) {
-	//c.migrationContext.Log.Infof("Coordinator: Handling changelog event: %+v\n", event)
-
 	c.mu.Lock()
 	defer c.mu.Unlock()
-
 	c.onChangelogEvent(event)
 }
 

go/logic/migrator.go

@@ -456,6 +456,7 @@ waitForGhostTable:
 		return err
 	}
 	this.printStatus(ForcePrintStatusRule)
+	this.printWorkerStats()
 
 	if this.migrationContext.IsCountingTableRows() {
 		this.migrationContext.Log.Info("stopping query for exact row count, because that can accidentally lock out the cut over")
@@ -762,10 +763,13 @@ func (this *Migrator) atomicCutOver() (err error) {
 
 // initiateServer begins listening on unix socket/tcp for incoming interactive commands
 func (this *Migrator) initiateServer() (err error) {
-	var f printStatusFunc = func(rule PrintStatusRule, writer io.Writer) {
+	var printStatus printStatusFunc = func(rule PrintStatusRule, writer io.Writer) {
 		this.printStatus(rule, writer)
 	}
-	this.server = NewServer(this.migrationContext, this.hooksExecutor, f)
+	var printWorkers printWorkersFunc = func(writer io.Writer) {
+		this.printWorkerStats(writer)
+	}
+	this.server = NewServer(this.migrationContext, this.hooksExecutor, printStatus, printWorkers)
 	if err := this.server.BindSocketFile(); err != nil {
 		return err
 	}
@@ -1022,6 +1026,29 @@ func (this *Migrator) shouldPrintMigrationStatusHint(rule PrintStatusRule, elaps
 	return shouldPrint
 }
 
+// printWorkerStats prints cumulative stats from the trxCoordinator workers.
+func (this *Migrator) printWorkerStats(writers ...io.Writer) {
+	writers = append(writers, os.Stdout)
+	mw := io.MultiWriter(writers...)
+
+	busyWorkers := this.trxCoordinator.busyWorkers.Load()
+	totalWorkers := cap(this.trxCoordinator.workerQueue)
+	fmt.Fprintf(mw, "# %d/%d workers are busy\n", busyWorkers, totalWorkers)
+
+	stats := this.trxCoordinator.GetWorkerStats()
+	for id, stat := range stats {
+		fmt.Fprintf(mw,
+			"Worker %d; Waited: %s; Busy: %s; DML Applied: %d (%.2f/s), Trx Applied: %d (%.2f/s)\n",
+			id,
+			base.PrettifyDurationOutput(stat.waitTime),
+			base.PrettifyDurationOutput(stat.busyTime),
+			stat.dmlEventsApplied,
+			stat.dmlRate,
+			stat.executedJobs,
+			stat.trxRate)
+	}
+}
+
 // printStatus prints the progress status, and optionally additionally detailed
 // dump of configuration.
 // `rule` indicates the type of output expected.

go/logic/server.go

@@ -32,6 +32,7 @@ var (
 )
 
 type printStatusFunc func(PrintStatusRule, io.Writer)
+type printWorkersFunc func(io.Writer)
 
 // Server listens for requests on a socket file or via TCP
 type Server struct {
@@ -40,14 +41,16 @@ type Server struct {
 	tcpListener      net.Listener
 	hooksExecutor    *HooksExecutor
 	printStatus      printStatusFunc
+	printWorkers     printWorkersFunc
 	isCPUProfiling   int64
 }
 
-func NewServer(migrationContext *base.MigrationContext, hooksExecutor *HooksExecutor, printStatus printStatusFunc) *Server {
+func NewServer(migrationContext *base.MigrationContext, hooksExecutor *HooksExecutor, printStatus printStatusFunc, printWorkers printWorkersFunc) *Server {
 	return &Server{
 		migrationContext: migrationContext,
 		hooksExecutor:    hooksExecutor,
 		printStatus:      printStatus,
+		printWorkers:     printWorkers,
 	}
 }
 
@@ -232,6 +235,9 @@ help                                 # This message
 		return ForcePrintStatusOnlyRule, nil
 	case "info", "status":
 		return ForcePrintStatusAndHintRule, nil
+	case "worker-stats":
+		this.printWorkers(writer)
+		return NoPrintStatusRule, nil
 	case "cpu-profile":
 		cpuProfile, err := this.runCPUProfile(arg)
 		if err == nil {