()
| 1452 | } |
| 1453 | |
| 1454 | func (e *Extractor) Stats() (*common.TaskStatistics, error) { |
| 1455 | totalRowsCopied := atomic.LoadInt64(&e.TotalRowsCopied) |
| 1456 | rowsEstimate := atomic.LoadInt64(&e.mysqlContext.RowsEstimate) |
| 1457 | deltaEstimate := atomic.LoadInt64(&e.mysqlContext.DeltaEstimate) |
| 1458 | if atomic.LoadInt64(&e.rowCopyCompleteFlag) == 1 { |
| 1459 | // Done copying rows. The totalRowsCopied value is the de-facto number of rows, |
| 1460 | // and there is no further need to keep updating the value. |
| 1461 | rowsEstimate = totalRowsCopied |
| 1462 | } |
| 1463 | var progressPct float64 |
| 1464 | if rowsEstimate == 0 { |
| 1465 | progressPct = 0.0 |
| 1466 | } else { |
| 1467 | progressPct = 100.0 * float64(totalRowsCopied) / float64(rowsEstimate) |
| 1468 | } |
| 1469 | |
| 1470 | var etaSeconds float64 = math.MaxFloat64 |
| 1471 | var eta string |
| 1472 | eta = "N/A" |
| 1473 | |
| 1474 | if progressPct >= 100.0 { |
| 1475 | eta = "0s" |
| 1476 | e.mysqlContext.Stage = common.StageMasterHasSentAllBinlogToSlave |
| 1477 | } else if progressPct >= 1.0 { |
| 1478 | elapsedRowCopySeconds := e.mysqlContext.ElapsedRowCopyTime().Seconds() |
| 1479 | totalExpectedSeconds := elapsedRowCopySeconds * float64(rowsEstimate) / float64(totalRowsCopied) |
| 1480 | etaSeconds = totalExpectedSeconds - elapsedRowCopySeconds |
| 1481 | if etaSeconds >= 0 { |
| 1482 | etaDuration := time.Duration(etaSeconds) * time.Second |
| 1483 | eta = base.PrettifyDurationOutput(etaDuration) |
| 1484 | } else { |
| 1485 | eta = "0s" |
| 1486 | } |
| 1487 | } |
| 1488 | |
| 1489 | extractedTxCount := e.binlogReader.GetExtractedTxCount() |
| 1490 | taskResUsage := common.TaskStatistics{ |
| 1491 | ExecMasterRowCount: totalRowsCopied, |
| 1492 | ExecMasterTxCount: deltaEstimate, |
| 1493 | ReadMasterRowCount: rowsEstimate, |
| 1494 | ReadMasterTxCount: deltaEstimate, |
| 1495 | ProgressPct: strconv.FormatFloat(progressPct, 'f', 1, 64), |
| 1496 | ETA: eta, |
| 1497 | Backlog: fmt.Sprintf("%d/%d", len(e.dataChannel), cap(e.dataChannel)), |
| 1498 | Stage: e.mysqlContext.Stage, |
| 1499 | BufferStat: common.BufferStat{ |
| 1500 | BinlogEventQueueSize: e.binlogReader.GetQueueSize(), |
| 1501 | ExtractorTxQueueSize: len(e.dataChannel), |
| 1502 | SendByTimeout: e.sendByTimeoutCounter, |
| 1503 | SendBySizeFull: e.sendBySizeFullCounter, |
| 1504 | }, |
| 1505 | DelayCount: &common.DelayCount{ |
| 1506 | Num: 0, |
| 1507 | Time: e.timestampCtx.GetDelay(), |
| 1508 | }, |
| 1509 | Timestamp: time.Now().UTC().UnixNano(), |
| 1510 | MemoryStat: common.MemoryStat{ |
| 1511 | Full: *e.memory1, |
nothing calls this directly
no test coverage detected