MCPcopy Create free account
hub / github.com/actiontech/dtle / OnApplierRotate

Method OnApplierRotate

driver/mysql/binlog/binlog_reader.go:1554–1601  ·  view source on GitHub ↗
(binlogFile string)

Source from the content-addressed store, hash-verified

1552}
1553
1554func (b *BinlogReader) OnApplierRotate(binlogFile string) {
1555 logger := b.logger.Named("OnApplierRotate")
1556 if !b.mysqlContext.BinlogRelay {
1557 // do nothing if BinlogRelay is not enabled
1558 return
1559 }
1560
1561 wrappingDir := b.getBinlogDir()
1562 fs, err := os.ReadDir(wrappingDir)
1563 if err != nil {
1564 logger.Error("ReadDir error", "dir", wrappingDir, "err", err)
1565 return
1566 }
1567
1568 dir := ""
1569
1570 for i := range fs {
1571 // https://pingcap.com/docs-cn/v3.0/reference/tools/data-migration/relay-log/
1572 // <server-uuid>.<subdir-seq-number>
1573 // currently there should only be .000001, but we loop to the last one.
1574 if fs[i].IsDir() {
1575 dir = filepath.Join(wrappingDir, fs[i].Name())
1576 } else {
1577 // meta-files like server-uuid.index
1578 }
1579 }
1580
1581 if dir == "" {
1582 logger.Warn("OnApplierRotate: no sub dir", "wrappingDir", wrappingDir)
1583 return
1584 }
1585
1586 realBinlogFile := normalizeBinlogFilename(binlogFile)
1587
1588 cmp, err := dmstreamer.CollectBinlogFilesCmp(dir, realBinlogFile, dmstreamer.FileCmpLess)
1589 if err != nil {
1590 logger.Error("err at cmp", "err", err)
1591 }
1592 b.logger.Debug("cmp", "cmp", cmp)
1593 for i := range cmp {
1594 f := filepath.Join(dir, cmp[i])
1595 b.logger.Info("will remove", "file", f)
1596 err := os.Remove(f)
1597 if err != nil {
1598 b.logger.Error("error when removing binlog", "file", f)
1599 }
1600 }
1601}
1602
1603func normalizeBinlogFilename(name string) string {
1604 // See `posUUIDSuffixSeparator` in pingcap/dm.

Callers 1

initNatsPubClientMethod · 0.80

Calls 2

getBinlogDirMethod · 0.95
normalizeBinlogFilenameFunction · 0.85

Tested by

no test coverage detected