Merge branch 'master' into iss909

This commit is contained in:
Rashiq 2021-03-08 18:20:03 +01:00 committed by GitHub
commit db513872c1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 13 additions and 6 deletions

View File

@ -13,6 +13,7 @@ import (
"time" "time"
gosql "database/sql" gosql "database/sql"
"github.com/github/gh-ost/go/mysql" "github.com/github/gh-ost/go/mysql"
) )
@ -62,7 +63,7 @@ func StringContainsAll(s string, substrings ...string) bool {
return nonEmptyStringsFound return nonEmptyStringsFound
} }
func ValidateConnection(db *gosql.DB, connectionConfig *mysql.ConnectionConfig, migrationContext *MigrationContext) (string, error) { func ValidateConnection(db *gosql.DB, connectionConfig *mysql.ConnectionConfig, migrationContext *MigrationContext, name string) (string, error) {
versionQuery := `select @@global.version` versionQuery := `select @@global.version`
var port, extraPort int var port, extraPort int
var version string var version string
@ -86,7 +87,7 @@ func ValidateConnection(db *gosql.DB, connectionConfig *mysql.ConnectionConfig,
} }
if connectionConfig.Key.Port == port || (extraPort > 0 && connectionConfig.Key.Port == extraPort) { if connectionConfig.Key.Port == port || (extraPort > 0 && connectionConfig.Key.Port == extraPort) {
migrationContext.Log.Infof("connection validated on %+v", connectionConfig.Key) migrationContext.Log.Infof("%s connection validated on %+v", name, connectionConfig.Key)
return version, nil return version, nil
} else if extraPort == 0 { } else if extraPort == 0 {
return "", fmt.Errorf("Unexpected database port reported: %+v", port) return "", fmt.Errorf("Unexpected database port reported: %+v", port)

View File

@ -57,6 +57,7 @@ type Applier struct {
singletonDB *gosql.DB singletonDB *gosql.DB
migrationContext *base.MigrationContext migrationContext *base.MigrationContext
finishedMigrating int64 finishedMigrating int64
name string
} }
func NewApplier(migrationContext *base.MigrationContext) *Applier { func NewApplier(migrationContext *base.MigrationContext) *Applier {
@ -64,6 +65,7 @@ func NewApplier(migrationContext *base.MigrationContext) *Applier {
connectionConfig: migrationContext.ApplierConnectionConfig, connectionConfig: migrationContext.ApplierConnectionConfig,
migrationContext: migrationContext, migrationContext: migrationContext,
finishedMigrating: 0, finishedMigrating: 0,
name: "applier",
} }
} }
@ -78,11 +80,11 @@ func (this *Applier) InitDBConnections() (err error) {
return err return err
} }
this.singletonDB.SetMaxOpenConns(1) this.singletonDB.SetMaxOpenConns(1)
version, err := base.ValidateConnection(this.db, this.connectionConfig, this.migrationContext) version, err := base.ValidateConnection(this.db, this.connectionConfig, this.migrationContext, this.name)
if err != nil { if err != nil {
return err return err
} }
if _, err := base.ValidateConnection(this.singletonDB, this.connectionConfig, this.migrationContext); err != nil { if _, err := base.ValidateConnection(this.singletonDB, this.connectionConfig, this.migrationContext, this.name); err != nil {
return err return err
} }
this.migrationContext.ApplierMySQLVersion = version this.migrationContext.ApplierMySQLVersion = version

View File

@ -29,12 +29,14 @@ type Inspector struct {
db *gosql.DB db *gosql.DB
informationSchemaDb *gosql.DB informationSchemaDb *gosql.DB
migrationContext *base.MigrationContext migrationContext *base.MigrationContext
name string
} }
func NewInspector(migrationContext *base.MigrationContext) *Inspector { func NewInspector(migrationContext *base.MigrationContext) *Inspector {
return &Inspector{ return &Inspector{
connectionConfig: migrationContext.InspectorConnectionConfig, connectionConfig: migrationContext.InspectorConnectionConfig,
migrationContext: migrationContext, migrationContext: migrationContext,
name: "inspector",
} }
} }
@ -198,7 +200,7 @@ func (this *Inspector) validateConnection() error {
return fmt.Errorf("MySQL replication length limited to 32 characters. See https://dev.mysql.com/doc/refman/5.7/en/assigning-passwords.html") return fmt.Errorf("MySQL replication length limited to 32 characters. See https://dev.mysql.com/doc/refman/5.7/en/assigning-passwords.html")
} }
version, err := base.ValidateConnection(this.db, this.connectionConfig, this.migrationContext) version, err := base.ValidateConnection(this.db, this.connectionConfig, this.migrationContext, this.name)
this.migrationContext.InspectorMySQLVersion = version this.migrationContext.InspectorMySQLVersion = version
return err return err
} }

View File

@ -42,6 +42,7 @@ type EventsStreamer struct {
listenersMutex *sync.Mutex listenersMutex *sync.Mutex
eventsChannel chan *binlog.BinlogEntry eventsChannel chan *binlog.BinlogEntry
binlogReader *binlog.GoMySQLReader binlogReader *binlog.GoMySQLReader
name string
} }
func NewEventsStreamer(migrationContext *base.MigrationContext) *EventsStreamer { func NewEventsStreamer(migrationContext *base.MigrationContext) *EventsStreamer {
@ -51,6 +52,7 @@ func NewEventsStreamer(migrationContext *base.MigrationContext) *EventsStreamer
listeners: [](*BinlogEventListener){}, listeners: [](*BinlogEventListener){},
listenersMutex: &sync.Mutex{}, listenersMutex: &sync.Mutex{},
eventsChannel: make(chan *binlog.BinlogEntry, EventsChannelBufferSize), eventsChannel: make(chan *binlog.BinlogEntry, EventsChannelBufferSize),
name: "streamer",
} }
} }
@ -106,7 +108,7 @@ func (this *EventsStreamer) InitDBConnections() (err error) {
if this.db, _, err = mysql.GetDB(this.migrationContext.Uuid, EventsStreamerUri); err != nil { if this.db, _, err = mysql.GetDB(this.migrationContext.Uuid, EventsStreamerUri); err != nil {
return err return err
} }
if _, err := base.ValidateConnection(this.db, this.connectionConfig, this.migrationContext); err != nil { if _, err := base.ValidateConnection(this.db, this.connectionConfig, this.migrationContext, this.name); err != nil {
return err return err
} }
if err := this.readCurrentBinlogCoordinates(); err != nil { if err := this.readCurrentBinlogCoordinates(); err != nil {