()
| 255 | } |
| 256 | |
| 257 | func (e *ExtractorOracle) Stats() (*common.TaskStatistics, error) { |
| 258 | totalRowsCopied := atomic.LoadInt64(&e.TotalRowsCopied) |
| 259 | rowsEstimate := atomic.LoadInt64(&e.mysqlContext.RowsEstimate) |
| 260 | deltaEstimate := atomic.LoadInt64(&e.mysqlContext.DeltaEstimate) |
| 261 | if atomic.LoadInt64(&e.rowCopyCompleteFlag) == 1 { |
| 262 | // Done copying rows. The totalRowsCopied value is the de-facto number of rows, |
| 263 | // and there is no further need to keep updating the value. |
| 264 | rowsEstimate = totalRowsCopied |
| 265 | } |
| 266 | var progressPct float64 |
| 267 | if rowsEstimate == 0 { |
| 268 | progressPct = 0.0 |
| 269 | } else { |
| 270 | progressPct = 100.0 * float64(totalRowsCopied) / float64(rowsEstimate) |
| 271 | } |
| 272 | |
| 273 | var etaSeconds float64 = math.MaxFloat64 |
| 274 | var eta string |
| 275 | eta = "N/A" |
| 276 | |
| 277 | if progressPct >= 100.0 { |
| 278 | eta = "0s" |
| 279 | e.mysqlContext.Stage = common.StageMasterHasSentAllBinlogToSlave |
| 280 | } else if progressPct >= 1.0 { |
| 281 | elapsedRowCopySeconds := e.mysqlContext.ElapsedRowCopyTime().Seconds() |
| 282 | totalExpectedSeconds := elapsedRowCopySeconds * float64(rowsEstimate) / float64(totalRowsCopied) |
| 283 | etaSeconds = totalExpectedSeconds - elapsedRowCopySeconds |
| 284 | if etaSeconds >= 0 { |
| 285 | etaDuration := time.Duration(etaSeconds) * time.Second |
| 286 | eta = base.PrettifyDurationOutput(etaDuration) |
| 287 | } else { |
| 288 | eta = "0s" |
| 289 | } |
| 290 | } |
| 291 | |
| 292 | var extractedTxCount uint32 |
| 293 | if e.LogMinerStream != nil { |
| 294 | extractedTxCount = atomic.LoadUint32(&e.LogMinerStream.OracleTxNum) |
| 295 | } |
| 296 | |
| 297 | taskResUsage := common.TaskStatistics{ |
| 298 | ExecMasterRowCount: totalRowsCopied, |
| 299 | ExecMasterTxCount: deltaEstimate, |
| 300 | ReadMasterRowCount: rowsEstimate, |
| 301 | ReadMasterTxCount: deltaEstimate, |
| 302 | ProgressPct: strconv.FormatFloat(progressPct, 'f', 1, 64), |
| 303 | ETA: eta, |
| 304 | Backlog: fmt.Sprintf("%d/%d", len(e.dataChannel), cap(e.dataChannel)), |
| 305 | Stage: e.mysqlContext.Stage, |
| 306 | BufferStat: common.BufferStat{ |
| 307 | BinlogEventQueueSize: e.binlogReader.GetQueueSize(), |
| 308 | ExtractorTxQueueSize: len(e.dataChannel), |
| 309 | SendByTimeout: e.sendByTimeoutCounter, |
| 310 | SendBySizeFull: e.sendBySizeFullCounter, |
| 311 | }, |
| 312 | DelayCount: &common.DelayCount{ |
| 313 | Num: 0, |
| 314 | //Time: e.timestampCtx.GetDelay(), |
nothing calls this directly
no test coverage detected