聊聊kingbus的DumpBinlogAt

序

本文主要研究一下 kingbus 的 DumpBinlogAt

DumpBinlogAt

kingbus/server/binlog_server.go

//DumpBinlogAt implements dump binlog event by slave executed gtid set
func (s *BinlogServer) DumpBinlogAt(ctx context.Context,
    startRaftIndex uint64, slaveGtids *gomysql.MysqlGTIDSet,
    eventC chan<- *storagepb.BinlogEvent, errorC chan<- error) error {
    var inExcludeGroup = false

    //new a binlog event reader from startRaftIndex, then send event to slave one by one
    reader, err := s.store.NewEntryReaderAt(startRaftIndex)
    if err != nil {log.Log.Errorf("NewEntryReaderAt error,err:%s,raftIndex:%d", err, startRaftIndex)
        return err
    }

    nextRaftIndex := reader.NextRaftIndex()
    log.Log.Infof("DumpBinlogAt:raftIndex:%d,slaveGtids:%s", nextRaftIndex, slaveGtids.String())
    go func() {
        for {
            //the next read raftIndex must be little than AppliedIndex
            if nextRaftIndex <= s.kingbusInfo.AppliedIndex() {raftEntry, err := reader.GetNext()
                if err != nil {
                    log.Log.Errorf("reader.GetNext error,err:%s,nextRaftIndex:%d,AppliedIndex:%d",
                        err, nextRaftIndex, s.kingbusInfo.AppliedIndex())
                    select {
                    case errorC <- err:
                    default:
                    }
                    return //need quit
                }
                nextRaftIndex = reader.NextRaftIndex()

                //this entry is not binlog event
                if utils.IsBinlogEvent(raftEntry) == false {continue}
                event := utils.DecodeBinlogEvent(raftEntry)
                //filter the event in slave gtids,if the event has send to slave
                inExcludeGroup = s.skipEvent(event, slaveGtids, inExcludeGroup)
                if inExcludeGroup {continue}

                select {
                case eventC <- event:
                case <-ctx.Done():
                    log.Log.Errorf("binlog server receive cancel, need quit,err:%s", ctx.Err())
                    select {case errorC <- ctx.Err():
                    default:
                    }
                    return //need quit
                }
            } else {
                select {case <-s.broadcast.Receive():
                    break
                case <-ctx.Done():
                    log.Log.Errorf("binlog server receive cancel, need quit,err:%s", ctx.Err())
                    select {case errorC <- ctx.Err():
                    default:
                    }
                    return //need quit
                }
            }
        }
    }()

    return nil
}

• DumpBinlogAt 方法通过 s.store.NewEntryReaderAt(startRaftIndex) 获取 reader，然后获取 nextRaftIndex，之后通过 utils.DecodeBinlogEvent(raftEntry) 获取 event，然后通过 s.skipEvent(event, slaveGtids, inExcludeGroup) 来判断是否该 skip，之后将 event 写入到 eventC

NewEntryReaderAt

kingbus/storage/disk_storage.go

//NewEntryReaderAt create a DiskEntryReader at raftIndex
func (s *DiskStorage) NewEntryReaderAt(raftIndex uint64) (EntryReader, error) {err := s.checkRaftIndex(raftIndex)
    if err != nil {log.Log.Errorf("checkRaftIndex error,err:%s,raftIndex:%d", err, raftIndex)
        return nil, err
    }

    reader := new(DiskEntryReader)
    reader.indexReadAt = raftIndex
    reader.store = s

    return reader, nil
}

• NewEntryReaderAt 方法先通过 s.checkRaftIndex(raftIndex) 校验一下 index，然后创建 DiskEntryReader，设置 indexReadAt 为 raftIndex

skipEvent

kingbus/server/binlog_server.go

//skipEvent filter the event has been executed by slave
func (s *BinlogServer) skipEvent(event *storagepb.BinlogEvent, slaveGtids *gomysql.MysqlGTIDSet, inExcludeGroup bool) bool {switch replication.EventType(event.Type) {
    case replication.GTID_EVENT:
        //remove header
        eventBody := event.Data[replication.EventHeaderSize:]
        //remove crc32
        eventBody = eventBody[:len(eventBody)-replication.BinlogChecksumLength]

        gtidEvent := &replication.GTIDEvent{}
        if err := gtidEvent.Decode(eventBody); err != nil {log.Log.Errorf("Decode gtid event error,err:%s", err)
            return true
        }
        u, err := uuid.FromBytes(gtidEvent.SID)
        if err != nil {log.Log.Errorf("FromBytes error,err:%s,sid:%v", err, gtidEvent.SID)
            return true
        }
        gtidStr := fmt.Sprintf("%s:%d", u.String(), gtidEvent.GNO)
        currentGtidset, err := gomysql.ParseMysqlGTIDSet(gtidStr)
        if err != nil {log.Log.Errorf("ParseMysqlGTIDSet error,err:%s,gtid:%s", err, gtidStr)
            return true
        }
        return slaveGtids.Contain(currentGtidset)
    case replication.ROTATE_EVENT:
        return false
    }
    return inExcludeGroup
}

• skipEvent 方法根据 replication.EventType(event.Type) 类型来判断，对于能够取到 currentGtidset 的通过 slaveGtids.Contain(currentGtidset) 判断，对于 replication.ROTATE_EVENT 返回 false

小结

DumpBinlogAt 方法通过 s.store.NewEntryReaderAt(startRaftIndex) 获取 reader，然后获取 nextRaftIndex，之后通过 utils.DecodeBinlogEvent(raftEntry) 获取 event，然后通过 s.skipEvent(event, slaveGtids, inExcludeGroup) 来判断是否该 skip，之后将 event 写入到 eventC

doc

• binlog_server