Skip to content

Commit

Permalink
patch: Enhance raft's invalid logic
Browse files Browse the repository at this point in the history
  • Loading branch information
@TCeason @BohuTANG
TCeason authored and BohuTANG committed Jul 23, 2019
1 parent 90590e5 commit b162334
Show file tree
Hide file tree
Showing 9 changed files with 210 additions and 60 deletions.
43 changes: 37 additions & 6 deletions src/mysql/api.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -114,27 +114,58 @@ func (m *Mysql) GTIDGreaterThan(gtid *model.GTID) (bool, model.GTID, error) {
return cmp > 0, this, nil
}

func (m *Mysql) GetLocalGTID(gtid string) (string, error) {
log := m.log
if gtid == "" {
return "", nil
}

uuid, err := m.GetUUID()
if err != nil {
log.Error("mysql.GetLocalGTID.error[%v]", err)
return "", err
}

s_gtid := strings.Split(gtid, ",")
for _, gtid := range s_gtid {
if strings.Contains(gtid, uuid) {
return gtid, nil
}
}

return "", nil
}

// CheckGTID use to compare the followerGTID and candidateGTID
func (m *Mysql) CheckGTID(followerGTID *model.GTID, candidateGTID *model.GTID) bool {
log := m.log
fRetrivedGTID := followerGTID.Retrieved_GTID_Set
cRetrivedGTID := candidateGTID.Retrieved_GTID_Set
fExecutedGTID := followerGTID.Executed_GTID_Set
fGTID, err := m.GetLocalGTID(fExecutedGTID)
if err != nil {
log.Error("mysql.CheckGTID.error[%v]", err)
}

cExecutedGTID := candidateGTID.Executed_GTID_Set
cGTID, err := m.GetLocalGTID(cExecutedGTID)
if err != nil {
log.Error("mysql.CheckGTID.error[%v]", err)
}

// follower never generate events, should vote, but if some one execute reset master, this may be error
// if a normal restart the follower retrived_gtid_set will be "" can't setState(INVALID)
if fRetrivedGTID == "" {
if fGTID == "" {
return false
}

// candidate has none RetrivedGTID, may be none retrived_gtid_set
// this means the candidate or new leader has not written, shouldnt vote
if cRetrivedGTID == "" {
if cGTID == "" {
return false
}

// gtid_sub is not none, means the follower gtid is bigger than candidate gtid
// if viewdiff<=0 it must be localcommitted
gtid_sub, err := m.GetGtidSubtract(fRetrivedGTID, cRetrivedGTID)
// if viewdiff<=0 and gtid_sub is not null it must be localcommitted
gtid_sub, err := m.GetGtidSubtract(fGTID, cGTID)
if err != nil {
log.Error("mysql.CheckGTID.error[%v]", err)
return false
Expand Down
117 changes: 90 additions & 27 deletions src/mysql/api_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -178,6 +178,27 @@ func TestWaitUntilAfterGTID(t *testing.T) {
assert.Nil(t, err)
}

func TestGetLocalGTID(t *testing.T) {
db, mock, err := sqlmock.New()
assert.Nil(t, err)
defer db.Close()

//log
log := xlog.NewStdLog(xlog.Level(xlog.PANIC))
conf := config.DefaultMysqlConfig()
mysql := NewMysql(conf, log)
mysql.db = db

query := "SELECT @@SERVER_UUID"
columns := []string{"@@SERVER_UUID"}
mockRows := sqlmock.NewRows(columns).AddRow("84030605-66aa-11e6-9465-52540e7fd51c")
mock.ExpectQuery(query).WillReturnRows(mockRows)

want := "84030605-66aa-11e6-9465-52540e7fd51c:1-160"
got, err := mysql.GetLocalGTID("84030605-66aa-11e6-9465-52540e7fd51c:1-160, 84030605-66bb-11e6-9465-52540e7fd51c:1-160")
assert.Equal(t, want, got)
}

func TestCheckGTID(t *testing.T) {
db, mock, err := sqlmock.New()
assert.Nil(t, err)
Expand All @@ -190,13 +211,18 @@ func TestCheckGTID(t *testing.T) {
mysql := NewMysql(conf, log)
mysql.db = db

// local is a normal follower, leader Executed_Gtid_Set is ""
// local is a normal follower, leader Executed_GTID_Set is ""
{
query := "SELECT @@SERVER_UUID"
columns := []string{"@@SERVER_UUID"}
mockRows := sqlmock.NewRows(columns).AddRow("84030605-66aa-11e6-9465-52540e7fd51c")
mock.ExpectQuery(query).WillReturnRows(mockRows)

GTID1 = model.GTID{
Retrieved_GTID_Set: "84030605-66aa-11e6-9465-52540e7fd51c:154-160",
Executed_GTID_Set: "84030605-66aa-11e6-9465-52540e7fd51c:154-160",
}
GTID2 = model.GTID{
Retrieved_GTID_Set: "",
Executed_GTID_Set: "",
}

want := false
Expand All @@ -205,13 +231,13 @@ func TestCheckGTID(t *testing.T) {
assert.Equal(t, want, got)
}

// local is a normal follower Retrieved_GTID_Set is "", leader Executed_Gtid_Set is ""
// local is a normal follower Executed_GTID_Set is "", leader Executed_GTID_Set is ""
{
GTID1 = model.GTID{
Retrieved_GTID_Set: "",
Executed_GTID_Set: "",
}
GTID2 = model.GTID{
Retrieved_GTID_Set: "",
Executed_GTID_Set: "",
}

want := false
Expand All @@ -220,13 +246,19 @@ func TestCheckGTID(t *testing.T) {
assert.Equal(t, want, got)
}

// local is a normal follower Retrieved_GTID_Set is "", leader do some dml
// local is a normal follower Executed_GTID_Set is "", leader do some dml
{
GTID1 = model.GTID{
Retrieved_GTID_Set: "",
Executed_GTID_Set: "",
}

query := "SELECT @@SERVER_UUID"
columns := []string{"@@SERVER_UUID"}
mockRows := sqlmock.NewRows(columns).AddRow("84030605-66aa-11e6-9465-52540e7fd51c")
mock.ExpectQuery(query).WillReturnRows(mockRows)

GTID2 = model.GTID{
Retrieved_GTID_Set: "84030605-66aa-11e6-9465-52540e7fd51c:1-160",
Executed_GTID_Set: "84030605-66aa-11e6-9465-52540e7fd51c:1-160",
}

want := false
Expand All @@ -237,37 +269,57 @@ func TestCheckGTID(t *testing.T) {

// local is a leader bug sprain, remote has leader but has none write
{
query := "SELECT @@SERVER_UUID"
columns := []string{"@@SERVER_UUID"}
mockRows := sqlmock.NewRows(columns).AddRow("84030605-66aa-11e6-9465-52540e7fd51c")
mock.ExpectQuery(query).WillReturnRows(mockRows)

GTID1 = model.GTID{
Retrieved_GTID_Set: "84030605-66aa-11e6-9465-52540e7fd51c:1-160",
Executed_GTID_Set: "84030605-66aa-11e6-9465-52540e7fd51c:1-160",
}

query = "SELECT @@SERVER_UUID"
columns = []string{"@@SERVER_UUID"}
mockRows = sqlmock.NewRows(columns).AddRow("84030605-66aa-11e6-9465-52540e7fd51c")
mock.ExpectQuery(query).WillReturnRows(mockRows)

GTID2 = model.GTID{
Retrieved_GTID_Set: "84030605-66aa-11e6-9465-52540e7fd51c:1-160",
Executed_GTID_Set: "84030605-66aa-11e6-9465-52540e7fd51c:1-160",
}

query := "SELECT GTID_SUBTRACT\\('84030605-66aa-11e6-9465-52540e7fd51c:1-160','84030605-66aa-11e6-9465-52540e7fd51c:1-160'\\) as gtid_sub"
log.Warning("%v", query)
columns := []string{"gtid_sub"}
mockRows := sqlmock.NewRows(columns).AddRow("")
query = "SELECT GTID_SUBTRACT\\('84030605-66aa-11e6-9465-52540e7fd51c:1-160','84030605-66aa-11e6-9465-52540e7fd51c:1-160'\\) as gtid_sub"
columns = []string{"gtid_sub"}
mockRows = sqlmock.NewRows(columns).AddRow("")
mock.ExpectQuery(query).WillReturnRows(mockRows)

want := false
got := mysql.CheckGTID(&GTID1, &GTID2)

assert.Equal(t, want, got)
}

// local is a leader bug sprain, remote has leader has writen
{
query := "SELECT @@SERVER_UUID"
columns := []string{"@@SERVER_UUID"}
mockRows := sqlmock.NewRows(columns).AddRow("84030605-66aa-11e6-9465-52540e7fd51c")
mock.ExpectQuery(query).WillReturnRows(mockRows)

GTID1 = model.GTID{
Retrieved_GTID_Set: "84030605-66aa-11e6-9465-52540e7fd51c:1-160",
Executed_GTID_Set: "84030605-66aa-11e6-9465-52540e7fd51c:1-160",
}

query = "SELECT @@SERVER_UUID"
columns = []string{"@@SERVER_UUID"}
mockRows = sqlmock.NewRows(columns).AddRow("84030605-66aa-11e6-9465-52540e7fd51c")
mock.ExpectQuery(query).WillReturnRows(mockRows)

GTID2 = model.GTID{
Retrieved_GTID_Set: "84030605-66aa-11e6-9465-52540e7fd51c:1-160, 84030605-77bb-11e6-9465-52540e7fd51c:1-10",
Executed_GTID_Set: "84030605-66aa-11e6-9465-52540e7fd51c:1-160, 84030605-77bb-11e6-9465-52540e7fd51c:1-10",
}

query := "SELECT GTID_SUBTRACT\\('84030605-66aa-11e6-9465-52540e7fd51c:1-160','84030605-66aa-11e6-9465-52540e7fd51c:1-160, 84030605-77bb-11e6-9465-52540e7fd51c:1-10'\\) as gtid_sub"
columns := []string{"gtid_sub"}
mockRows := sqlmock.NewRows(columns).AddRow("")
query = "SELECT GTID_SUBTRACT\\('84030605-66aa-11e6-9465-52540e7fd51c:1-160','84030605-66aa-11e6-9465-52540e7fd51c:1-160'\\) as gtid_sub"
columns = []string{"gtid_sub"}
mockRows = sqlmock.NewRows(columns).AddRow("")
mock.ExpectQuery(query).WillReturnRows(mockRows)

want := false
Expand All @@ -278,16 +330,27 @@ func TestCheckGTID(t *testing.T) {

// local is a leader bug sprain and localcommitted, remote has leader has writen
{
query := "SELECT @@SERVER_UUID"
columns := []string{"@@SERVER_UUID"}
mockRows := sqlmock.NewRows(columns).AddRow("84030605-66aa-11e6-9465-52540e7fd51c")
mock.ExpectQuery(query).WillReturnRows(mockRows)

GTID1 = model.GTID{
Retrieved_GTID_Set: "84030605-66aa-11e6-9465-52540e7fd51c:1-161",
Executed_GTID_Set: "84030605-66aa-11e6-9465-52540e7fd51c:1-161",
}

query = "SELECT @@SERVER_UUID"
columns = []string{"@@SERVER_UUID"}
mockRows = sqlmock.NewRows(columns).AddRow("84030605-66aa-11e6-9465-52540e7fd51c")
mock.ExpectQuery(query).WillReturnRows(mockRows)

GTID2 = model.GTID{
Retrieved_GTID_Set: "84030605-66aa-11e6-9465-52540e7fd51c:1-160, 84030605-77bb-11e6-9465-52540e7fd51c:1-10",
Executed_GTID_Set: "84030605-66aa-11e6-9465-52540e7fd51c:1-160, 84030605-77bb-11e6-9465-52540e7fd51c:1-10",
}

query := "SELECT GTID_SUBTRACT\\('84030605-66aa-11e6-9465-52540e7fd51c:1-161','84030605-66aa-11e6-9465-52540e7fd51c:1-160, 84030605-77bb-11e6-9465-52540e7fd51c:1-10'\\) as gtid_sub"
columns := []string{"gtid_sub"}
mockRows := sqlmock.NewRows(columns).AddRow("84030605-66aa-11e6-9465-52540e7fd51c:161")
query = "SELECT GTID_SUBTRACT\\('84030605-66aa-11e6-9465-52540e7fd51c:1-161','84030605-66aa-11e6-9465-52540e7fd51c:1-160'\\) as gtid_sub"
columns = []string{"gtid_sub"}
mockRows = sqlmock.NewRows(columns).AddRow("84030605-66aa-11e6-9465-52540e7fd51c:161")
mock.ExpectQuery(query).WillReturnRows(mockRows)

want := true
Expand Down Expand Up @@ -521,7 +584,7 @@ func TestGetGTID(t *testing.T) {

want := model.GTID{Master_Log_File: "mysql-bin.000001",
Read_Master_Log_Pos: 147,
Retrieved_GTID_Set: "84030605-66aa-11e6-9465-52540e7fd51c:154-160",
Retrieved_GTID_Set: "",
Executed_GTID_Set: "84030605-66aa-11e6-9465-52540e7fd51c:154-160",
Slave_IO_Running: true,
Slave_SQL_Running: true,
Expand Down
37 changes: 20 additions & 17 deletions src/mysql/mock.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@ type MockGTID struct {
ChangeToMasterFn func(*sql.DB) error
WaitUntilAfterGTIDFn func(*sql.DB, string) error
GetGtidSubtractFn func(*sql.DB, string, string) (string, error)
GetUUIDFn func(*sql.DB) (string, error)
CheckGTIDFn func(*model.GTID, *model.GTID) bool
SetGlobalSysVarFn func(*sql.DB, string) error
ResetMasterFn func(*sql.DB) error
Expand Down Expand Up @@ -70,6 +71,16 @@ func (mogtid *MockGTID) GetSlaveGTID(db *sql.DB) (*model.GTID, error) {
return mogtid.GetSlaveGTIDFn(db)
}

// DefaultGetUUID mock.
func DefaultGetUUID(db *sql.DB) (string, error) {
return "84030605-66aa-11e6-9465-52540e7fd51c", nil
}

// GetUUID mock.
func (mogtid *MockGTID) GetUUID(db *sql.DB) (string, error) {
return mogtid.GetUUIDFn(db)
}

// DefaultGetMasterGTID mock.
func DefaultGetMasterGTID(db *sql.DB) (*model.GTID, error) {
gtid := &model.GTID{}
Expand Down Expand Up @@ -386,6 +397,7 @@ func defaultMockGTID() *MockGTID {
mock.ChangeToMasterFn = DefaultChangeToMaster
mock.WaitUntilAfterGTIDFn = DefaultWaitUntilAfterGTID
mock.GetGtidSubtractFn = DefaultGetGtidSubtract
mock.GetUUIDFn = DefaultGetUUID
mock.CheckGTIDFn = DefaultCheckGTID
mock.SetGlobalSysVarFn = DefaultSetGlobalSysVar
mock.ResetMasterFn = DefaultResetMaster
Expand Down Expand Up @@ -450,31 +462,21 @@ func GetMasterGTIDA(db *sql.DB) (*model.GTID, error) {
return gtid, nil
}

func GetUUIDA(db *sql.DB) (string, error) {
return "052077a5-b6f4-ee1b-61ec-d80a8b27d749", nil
}

// NewMockGTIDA mock.
func NewMockGTIDA() *MockGTID {
mock := defaultMockGTID()
mock.GetMasterGTIDFn = GetMasterGTIDA
mock.GetSlaveGTIDFn = GetSlaveGTIDA
mock.GetUUIDFn = GetUUIDA
return mock
}

// GetSlaveGTIDLC mock.
func GetSlaveGTIDLC(db *sql.DB) (*model.GTID, error) {
gtid := &model.GTID{}
gtid.Master_Log_File = ""
gtid.Read_Master_Log_Pos = 0
gtid.Slave_IO_Running = true
gtid.Slave_SQL_Running = true
gtid.Slave_IO_Running_Str = "Yes"
gtid.Slave_SQL_Running_Str = "Yes"
gtid.Seconds_Behind_Master = "1"
gtid.Last_Error = ""
gtid.Slave_SQL_Running_State = "Slave has read all relay log; waiting for the slave I/O thread to update it"
gtid.Executed_GTID_Set = `052077a5-b6f4-ee1b-61ec-d80a8b27d749:1-37,
12446bf7-3219-11e5-9434-080027079e3d:8058-963126`
gtid.Retrieved_GTID_Set = `052077a5-b6f4-ee1b-61ec-d80a8b27d749:1-36,
12446bf7-3219-11e5-9434-080027079e3d:8058-963126`
return gtid, nil
func GetUUIDLC(db *sql.DB) (string, error) {
return "052077a5-b6f4-ee1b-61ec-d80a8b27d749", nil
}

// GetMasterGTIDLC mock.
Expand Down Expand Up @@ -505,6 +507,7 @@ func NewMockGTIDLC() *MockGTID {
mock.GetSlaveGTIDFn = GetMasterGTIDLC
mock.CheckGTIDFn = CheckGTIDLC
mock.GetGtidSubtractFn = GetGtidSubtractInvalid
mock.GetUUIDFn = GetUUIDLC
return mock
}

Expand Down
21 changes: 21 additions & 0 deletions src/mysql/mysql.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -123,6 +123,27 @@ func (m *Mysql) Ping() {
m.pingEntry = *pe
}

// GetUUID used to get local uuid.
func (m *Mysql) GetUUID() (string, error) {
var err error
var db *sql.DB
var uuid string
log := m.log

if db, err = m.getDB(); err != nil {
log.Error("mysql.get.local.uuid.error[%v]", err)
return "", err
}

if uuid, err = m.mysqlHandler.GetUUID(db); err != nil {
log.Error("mysql.get.local.uuid.error[%v]", err)
return "", err
}
log.Info("mysql.get.local.uuid:[%v]", uuid)

return uuid, nil
}

// GetMasterGTID used to get master binlog info.
func (m *Mysql) GetMasterGTID() (*model.GTID, error) {
var err error
Expand Down
3 changes: 3 additions & 0 deletions src/mysql/mysql_handler.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,9 @@ type MysqlHandler interface {
// waits until slave replication reaches at least targetGTID
WaitUntilAfterGTID(*sql.DB, string) error

// get local uuid
GetUUID(db *sql.DB) (string, error)

// get gtid subtract with slavegtid and master gtid
GetGtidSubtract(*sql.DB, string, string) (string, error)

Expand Down
Loading

0 comments on commit b162334

Please sign in to comment.