// This file contains utility methods to create binlog replication
// packets. They are mostly used for testing.
// NewMySQL56BinlogFormat returns a typical BinlogFormat for MySQL 5.6.
547071
func NewMySQL56BinlogFormat() BinlogFormat {
	return BinlogFormat{
		FormatVersion:     4,
		ServerVersion:     "5.6.33-0ubuntu0.14.04.1-log",
		HeaderLength:      19,
		ChecksumAlgorithm: BinlogChecksumAlgCRC32, // most commonly used.
		HeaderSizes: []byte{
			56, 13, 0, 8, 0, 18, 0, 4, 4, 4,
			4, 18, 0, 0, 92, 0, 4, 26, 8, 0,
			0, 0, 8, 8, 8, 2, 0, 0, 0, 10,
			10, 10, 25, 25, 0},
	}
}
969896
func (h binlogRecover) ServeHTTP(w http.ResponseWriter, req *http.Request) {
	binloginfo.DisableSkipBinlogFlag()
}
1015996
func (info *BinlogInfo) WriteBinlog(clusterID uint64) error {
	skip := atomic.LoadUint32(&skipBinlog)
	if skip > 0 {
		metrics.CriticalErrorCounter.Add(1)
		return nil
	}

	if info.Client == nil {
		return errors.New("pumps client is nil")
	}

	// it will retry in PumpsClient if write binlog fail.
	err := info.Client.WriteBinlog(info.Data)
	if err != nil {
		logutil.Logger(context.Background()).Error("write binlog failed", zap.Error(err))
		if atomic.LoadUint32(&ignoreError) == 1 {
			logutil.Logger(context.Background()).Error("write binlog fail but error ignored")
			metrics.CriticalErrorCounter.Add(1)
			// If error happens once, we'll stop writing binlog.
			atomic.CompareAndSwapUint32(&skipBinlog, skip, skip+1)
			return nil
		}

		if strings.Contains(err.Error(), "received message larger than max") {
			// This kind of error is not critical, return directly.
			return errors.Errorf("binlog data is too large (%s)", err.Error())
		}

		return terror.ErrCritical.GenWithStackByArgs(err)
	}

	return nil
}
1194570
func NewLog(out io.Writer) *Log {
	d := new(Log)
	d.Logger = log.New(out, "[ORM]", 1e9)
	return d
}
367767
func Uint64EncodeSql(b encoding2.BinaryWriter, num uint64) {
	numVal, _ := BuildValue(num)
	numVal.EncodeSql(b)
}
679316
func NewRawPacket(bytes ...[]byte) *Packet {
	p := &Packet{b: make([]byte, 0, MaxEIRPacketLength)}
	for _, b := range bytes {
		p.b = append(p.b, b...)
	}
	return p
}
789150
func connectForReplication(cp *mysql.ConnParams) (*mysql.Conn, error) {
	params, err := dbconfigs.WithCredentials(cp)
	if err != nil {
		return nil, err
	}

	ctx := context.Background()
	conn, err := mysql.Connect(ctx, params)
	if err != nil {
		return nil, err
	}

	// Tell the server that we understand the format of events
	// that will be used if binlog_checksum is enabled on the server.
	if _, err := conn.ExecuteFetch("SET @master_binlog_checksum=@@global.binlog_checksum", 0, false); err != nil {
		return nil, fmt.Errorf("failed to set @master_binlog_checksum=@@global.binlog_checksum: %v", err)
	}

	return conn, nil
}
312545
func (mysqlFlavor) readBinlogEvent(c *Conn) (BinlogEvent, error) {
	result, err := c.ReadPacket()
	if err != nil {
		return nil, err
	}
	switch result[0] {
	case EOFPacket:
		return nil, NewSQLError(CRServerLost, SSUnknownSQLState, "%v", io.EOF)
	case ErrPacket:
		return nil, ParseErrorPacket(result)
	}
	return NewMysql56BinlogEvent(result[1:]), nil
}
702066
func NewFormatLogger(out io.Writer, flags FlagSet, e Emitter) *Logger {
	return &Logger{
		out:   out,
		flags: uint64(flags),
		e:     e,
	}
}
566269
func NewFileLog(cfg FileLogConfig) (*FileLog, error) {
	if err := cfg.CheckAndSetDefaults(); err != nil {
		return nil, trace.Wrap(err)
	}
	f := &FileLog{
		FileLogConfig: cfg,
		Entry: log.WithFields(log.Fields{
			trace.Component: teleport.ComponentAuditLog,
		}),
	}
	return f, nil
}
338519
func NewDecayedLog(dbPath string,
	notifier chainntnfs.ChainNotifier) *DecayedLog {

	// Use default path for log database
	if dbPath == "" {
		dbPath = defaultDbDirectory
	}

	return &DecayedLog{
		dbPath:   dbPath,
		notifier: notifier,
		quit:     make(chan struct{}),
	}
}
611392
func (sc *SlaveConnection) StartBinlogDumpFromCurrent(ctx context.Context) (mysql.Position, <-chan mysql.BinlogEvent, error) {
	ctx, sc.cancel = context.WithCancel(ctx)

	masterPosition, err := sc.Conn.MasterPosition()
	if err != nil {
		return mysql.Position{}, nil, fmt.Errorf("failed to get master position: %v", err)
	}

	c, err := sc.StartBinlogDumpFromPosition(ctx, masterPosition)
	return masterPosition, c, err
}
1367695
func (vre *Engine) executeFetchMaybeCreateTable(dbClient binlogplayer.DBClient, query string, maxrows int) (qr *sqltypes.Result, err error) {
	qr, err = dbClient.ExecuteFetch(query, maxrows)

	if err == nil {
		return
	}

	// If it's a bad table or db, it could be because _vt.vreplication wasn't created.
	// In that case we can try creating it again.
	merr, isSQLErr := err.(*mysql.SQLError)
	if !isSQLErr || !(merr.Num == mysql.ERNoSuchTable || merr.Num == mysql.ERBadDb || merr.Num == mysql.ERBadFieldError) {
		return qr, err
	}

	log.Info("Looks like _vt.vreplication table may not exist. Trying to recreate... ")
	if merr.Num == mysql.ERNoSuchTable || merr.Num == mysql.ERBadDb {
		for _, query := range binlogplayer.CreateVReplicationTable() {
			if _, merr := dbClient.ExecuteFetch(query, 0); merr != nil {
				log.Warningf("Failed to ensure _vt.vreplication table exists: %v", merr)
				return nil, err
			}
		}
	}
	if merr.Num == mysql.ERBadFieldError {
		log.Info("Adding column to table _vt.vreplication")
		for _, query := range binlogplayer.AlterVReplicationTable() {
			if _, merr := dbClient.ExecuteFetch(query, 0); merr != nil {
				merr, isSQLErr := err.(*mysql.SQLError)
				if !isSQLErr || !(merr.Num == mysql.ERDupFieldName) {
					log.Warningf("Failed to alter _vt.vreplication table: %v", merr)
					return nil, err
				}
			}
		}
	}
	return dbClient.ExecuteFetch(query, maxrows)
}
1101264
func newBytesBinParser(b []byte) *binParser {
	p := &binParser{bufio.NewReader(bytes.NewReader(b)), 0, 0}
	if p.r == nil {
		return nil
	}
	return p
}
1319802
func NewLoggerWithFormatter(level string, formatter logrus.Formatter) (*logrus.Logger, error) {
	logger := logrus.New()
	// If there is an environment override, use that
	envLogLevel := os.Getenv(envVarLogLevel)
	if envLogLevel != "" {
		level = envLogLevel
	}

	logLevel, err := logrus.ParseLevel(level)
	if err != nil {
		return nil, err
	}

	logger.Level = logLevel
	if nil != formatter {
		logger.Formatter = formatter
	}
	logger.Out = os.Stdout
	return logger, nil
}
1185169
func NewIntVarEvent(f BinlogFormat, s *FakeBinlogStream, typ byte, value uint64) BinlogEvent {
	length := 9
	data := make([]byte, length)

	data[0] = typ
	data[1] = byte(value)
	data[2] = byte(value >> 8)
	data[3] = byte(value >> 16)
	data[4] = byte(value >> 24)
	data[5] = byte(value >> 32)
	data[6] = byte(value >> 40)
	data[7] = byte(value >> 48)
	data[8] = byte(value >> 56)

	ev := s.Packetize(f, eIntVarEvent, 0, data)
	return NewMysql56BinlogEvent(ev)
}
1282971
func NewInvalidQueryEvent(f BinlogFormat, s *FakeBinlogStream) BinlogEvent {
	length := 100
	data := make([]byte, length)
	data[4+4] = 200 // > 100

	ev := s.Packetize(f, eQueryEvent, 0, data)
	return NewMysql56BinlogEvent(ev)
}