add metrics

- send_by_timeout/size_full
- dest_queue2_size

Author: ffffwh

drivers/mysql/common/models.go

@@ -46,6 +46,7 @@ type ThroughputStat struct {
 type BufferStat struct {
 	BinlogEventQueueSize    int
 	ExtractorTxQueueSize    int
+	ApplierMsgQueueSize     int
 	ApplierTxQueueSize      int
 	SendByTimeout           int
 	SendBySizeFull          int

drivers/mysql/handle.go

@@ -134,7 +134,8 @@ func (h *taskHandle) emitStats(ru *common.TaskStatistics) {
 		metrics.SetGaugeWithLabels([]string{"memory.full_kb_est"}, float32(ru.MemoryStat.Full) * srcFullFactor / 1024, labels)
 		metrics.SetGaugeWithLabels([]string{"memory.incr_kb_est"}, float32(ru.MemoryStat.Incr) * srcIncrFactor / 1024, labels)
 	case common.TaskTypeDest:
-		metrics.SetGaugeWithLabels([]string{"buffer", "dest_queue_size"}, float32(ru.BufferStat.ApplierTxQueueSize), labels)
+		metrics.SetGaugeWithLabels([]string{"buffer", "dest_queue_size"}, float32(ru.BufferStat.ApplierMsgQueueSize), labels)
+		metrics.SetGaugeWithLabels([]string{"buffer", "dest_queue2_size"}, float32(ru.BufferStat.ApplierTxQueueSize), labels)
 
 		metrics.SetGaugeWithLabels([]string{"memory.full_kb_est"}, float32(ru.MemoryStat.Full) * dstFullFactor / 1024, labels)
 		metrics.SetGaugeWithLabels([]string{"memory.incr_kb_est"}, float32(ru.MemoryStat.Incr) * dstIncrFactor / 1024, labels)

drivers/mysql/mysql/applier.go

@@ -791,13 +791,17 @@ func (a *Applier) ApplyEventQueries(db *gosql.DB, entry *common.DumpEntry) error
 func (a *Applier) Stats() (*common.TaskStatistics, error) {
 	a.logger.Debug("Stats")
 	var totalDeltaCopied int64
-	var lenApplyDataEntryQueue int
-	var capApplyDataEntryQueue int
+	var lenApplierMsgQueue int
+	var capApplierMsgQueue int
+	var lenApplierTxQueue int
+	var capApplierTxQueue int
 	var delay int64
 	if a.ai != nil {
 		totalDeltaCopied = a.ai.TotalDeltaCopied
-		lenApplyDataEntryQueue = len(a.ai.incrBytesQueue)
-		capApplyDataEntryQueue = cap(a.ai.incrBytesQueue)
+		lenApplierMsgQueue = len(a.ai.incrBytesQueue)
+		capApplierMsgQueue = cap(a.ai.incrBytesQueue)
+		lenApplierTxQueue = len(a.ai.binlogEntryQueue)
+		capApplierTxQueue = cap(a.ai.binlogEntryQueue)
 		delay = a.ai.timestampCtx.GetDelay()
 	}
 	totalRowsReplay := a.TotalRowsReplayed
@@ -813,7 +817,8 @@ func (a *Applier) Stats() (*common.TaskStatistics, error) {
 		if a.mysqlContext.Gtid != "" {
 			// Done copying rows. The totalRowsCopied value is the de-facto number of rows,
 			// and there is no further need to keep updating the value.
-			backlog = fmt.Sprintf("%d/%d", lenApplyDataEntryQueue, capApplyDataEntryQueue)
+			backlog = fmt.Sprintf("%d/%d", lenApplierMsgQueue + lenApplierTxQueue,
+				capApplierMsgQueue + capApplierTxQueue)
 		} else {
 			backlog = fmt.Sprintf("%d/%d", len(a.fullBytesQueue), cap(a.fullBytesQueue))
 		}
@@ -861,7 +866,8 @@ func (a *Applier) Stats() (*common.TaskStatistics, error) {
 			RetrievedGtidSet:   "",
 		},
 		BufferStat: common.BufferStat{
-			ApplierTxQueueSize: lenApplyDataEntryQueue,
+			ApplierMsgQueueSize: lenApplierMsgQueue,
+			ApplierTxQueueSize:  lenApplierTxQueue,
 		},
 		Timestamp: time.Now().UTC().UnixNano(),
 		DelayCount: &common.DelayCount{

drivers/mysql/mysql/extractor.go

@@ -923,6 +923,8 @@ func (e *Extractor) StreamEvents() error {
 						"groupMaxSize", e.mysqlContext.GroupMaxSize,
 						"Entries.len", len(entries.Entries))
 
+					e.sendBySizeFullCounter += 1
+
 					err := sendEntriesAndClear()
 					if err != nil {
 						e.onError(TaskStateDead, err)
@@ -939,6 +941,7 @@ func (e *Extractor) StreamEvents() error {
 				if nEntries > 0 {
 					e.logger.Debug("incr. send by timeout.", "entriesSize", entriesSize,
 						"timeout", e.mysqlContext.GroupTimeout)
+					e.sendByTimeoutCounter += 1
 					err := sendEntriesAndClear()
 					if err != nil {
 						e.onError(TaskStateDead, err)