From 3ca5f89fa52aa76d4ef0725ddad1eb85e13dc5f7 Mon Sep 17 00:00:00 2001 From: Abeyu M Date: Tue, 8 Oct 2019 13:49:15 -0400 Subject: [PATCH] add migrationcontext to gomysql_reader --- go/binlog/gomysql_reader.go | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/go/binlog/gomysql_reader.go b/go/binlog/gomysql_reader.go index 41d1ead..bc80cb5 100644 --- a/go/binlog/gomysql_reader.go +++ b/go/binlog/gomysql_reader.go @@ -13,13 +13,13 @@ import ( "github.com/github/gh-ost/go/mysql" "github.com/github/gh-ost/go/sql" - "github.com/outbrain/golib/log" gomysql "github.com/siddontang/go-mysql/mysql" "github.com/siddontang/go-mysql/replication" "golang.org/x/net/context" ) type GoMySQLReader struct { + migrationContext *base.MigrationContext connectionConfig *mysql.ConnectionConfig binlogSyncer *replication.BinlogSyncer binlogStreamer *replication.BinlogStreamer @@ -30,6 +30,7 @@ type GoMySQLReader struct { func NewGoMySQLReader(migrationContext *base.MigrationContext) (binlogReader *GoMySQLReader, err error) { binlogReader = &GoMySQLReader{ + migrationContext: migrationContext, connectionConfig: migrationContext.InspectorConnectionConfig, currentCoordinates: mysql.BinlogCoordinates{}, currentCoordinatesMutex: &sync.Mutex{}, @@ -57,11 +58,11 @@ func NewGoMySQLReader(migrationContext *base.MigrationContext) (binlogReader *Go // ConnectBinlogStreamer func (this *GoMySQLReader) ConnectBinlogStreamer(coordinates mysql.BinlogCoordinates) (err error) { if coordinates.IsEmpty() { - return log.Errorf("Empty coordinates at ConnectBinlogStreamer()") + return this.migrationContext.Log.Errorf("Empty coordinates at ConnectBinlogStreamer()") } this.currentCoordinates = coordinates - log.Infof("Connecting binlog streamer at %+v", this.currentCoordinates) + this.migrationContext.Log.Infof("Connecting binlog streamer at %+v", this.currentCoordinates) // Start sync with specified binlog file and position this.binlogStreamer, err = this.binlogSyncer.StartSync(gomysql.Position{this.currentCoordinates.LogFile, uint32(this.currentCoordinates.LogPos)}) @@ -78,7 +79,7 @@ func (this *GoMySQLReader) GetCurrentBinlogCoordinates() *mysql.BinlogCoordinate // StreamEvents func (this *GoMySQLReader) handleRowsEvent(ev *replication.BinlogEvent, rowsEvent *replication.RowsEvent, entriesChannel chan<- *BinlogEntry) error { if this.currentCoordinates.SmallerThanOrEquals(&this.LastAppliedRowsEventHint) { - log.Debugf("Skipping handled query at %+v", this.currentCoordinates) + this.migrationContext.Log.Debugf("Skipping handled query at %+v", this.currentCoordinates) return nil } @@ -147,14 +148,14 @@ func (this *GoMySQLReader) StreamEvents(canStopStreaming func() bool, entriesCha defer this.currentCoordinatesMutex.Unlock() this.currentCoordinates.LogFile = string(rotateEvent.NextLogName) }() - log.Infof("rotate to next log from %s:%d to %s", this.currentCoordinates.LogFile, int64(ev.Header.LogPos), rotateEvent.NextLogName) + this.migrationContext.Log.Infof("rotate to next log from %s:%d to %s", this.currentCoordinates.LogFile, int64(ev.Header.LogPos), rotateEvent.NextLogName) } else if rowsEvent, ok := ev.Event.(*replication.RowsEvent); ok { if err := this.handleRowsEvent(ev, rowsEvent, entriesChannel); err != nil { return err } } } - log.Debugf("done streaming events") + this.migrationContext.Log.Debugf("done streaming events") return nil }