Repository: pingcap/dumpling
Pull request #117 (2020-07-08 13:25:40)

fix concurrency problem
<!--
Thank you for contributing to Dumpling! Please read the [CONTRIBUTING](https://github.com/pingcap/dumpling/blob/master/CONTRIBUTING.md) document **BEFORE** filing this PR.
-->

### What problem does this PR solve? <!--add issue link with summary if exists-->
1. Now we get tables before we flush all tables, it's not safe when there are tables concurrently written.
2. Support record `show slave status` info in metadata for mysql/mariaDB.
3. Fix the problem that can't set snapshot `lock` when server type is `tidb`.

### What is changed and how it works?
1. For consistency `flush`: lock tables -> record metadata -> get tables list
    For consistency `lock`:  record metadata -> get tables list -> lock tables
2. Add metadata support in `metadata.go`.
3. Check whether consistency is `snapshot` before add `snapshot` session variable.

### Check List <!--REMOVE the items that are not applicable-->

Tests <!-- At least one of them must be included. -->

 - Unit test
 - Integration test
 - Manual test (add detailed scripts or steps below)
    Manually check metadata for tidb/mysql. Looks fine to me.

Side effects

 - Increased code complexity
 
### Release note

<!-- bugfixes or new feature need a release note, must in the form of a list, such as

- support -T/--tables-list argument

or if no need to be included in the release note, just add the following line

- No release note
-->
- Fix the bug that flush tables may not work for tables to dump

codecov[bot]: # [Codecov](https://codecov.io/gh/pingcap/dumpling/pull/117?src=pr&el=h1) Report
> Merging [#117](https://codecov.io/gh/pingcap/dumpling/pull/117?src=pr&el=desc) into [master](https://codecov.io/gh/pingcap/dumpling/commit/99dd90410fee5c4a0d009fa02e348b0004fb4255&el=desc) will **increase** coverage by `1.18%`.
> The diff coverage is `56.43%`.


```diff
@@            Coverage Diff             @@
##           master     #117      +/-   ##
==========================================
+ Coverage   50.76%   51.94%   +1.18%     
==========================================
  Files          17       17              
  Lines        1775     1829      +54     
==========================================
+ Hits          901      950      +49     
- Misses        811      813       +2     
- Partials       63       66       +3     
```


Patch:
diff --git a/v4/export/dump.go b/v4/export/dump.go
index ca55d13..eb6a910 100644
--- a/v4/export/dump.go
+++ b/v4/export/dump.go
@@ -76,12 +76,14 @@ func Dump(pCtx context.Context, conf *Config) (err error) {
 		if conf.ServerInfo.ServerType != ServerTypeTiDB {
 			return errors.New("snapshot consistency is not supported for this server")
 		}
-		hasTiKV, err := CheckTiDBWithTiKV(pool)
-		if err != nil {
-			return err
-		}
-		if hasTiKV {
-			conf.SessionParams["tidb_snapshot"] = conf.Snapshot
+		if conf.Consistency == "snapshot" {
+			hasTiKV, err := CheckTiDBWithTiKV(pool)
+			if err != nil {
+				return err
+			}
+			if hasTiKV {
+				conf.SessionParams["tidb_snapshot"] = conf.Snapshot
+			}
 		}
 	}
 
@@ -103,26 +105,23 @@ func Dump(pCtx context.Context, conf *Config) (err error) {
 		return err
 	}
 
-	databases, err := prepareDumpingDatabases(conf, pool)
-	if err != nil {
-		return err
-	}
-
-	conf.Tables, err = listAllTables(pool, databases)
-	if err != nil {
-		return err
-	}
+	m := newGlobalMetadata(conf.OutputDirPath)
+	// write metadata even if dump failed
+	defer m.writeGlobalMetaData()
 
-	if !conf.NoViews {
-		views, err := listAllViews(pool, databases)
+	// for consistency lock, we should lock tables at first to get the tables we want to lock & dump
+	// for consistency lock, record meta pos before lock tables because other tables may still be modified while locking tables
+	if conf.Consistency == "lock" {
+		m.recordStartTime(time.Now())
+		err = m.recordGlobalMetaData(pool, conf.ServerInfo.ServerType)
 		if err != nil {
+			log.Info("get global metadata failed", zap.Error(err))
+		}
+		if err = prepareTableListToDump(conf, pool); err != nil {
 			return err
 		}
-		conf.Tables.Merge(views)
 	}
 
-	filterTables(conf)
-
 	conCtrl, err := NewConsistencyController(conf, pool)
 	if err != nil {
 		return err
@@ -131,13 +130,17 @@ func Dump(pCtx context.Context, conf *Config) (err error) {
 		return err
 	}
 
-	m := newGlobalMetadata(conf.OutputDirPath)
-	// write metadata even if dump failed
-	defer m.writeGlobalMetaData()
-	m.recordStartTime(time.Now())
-	err = m.getGlobalMetaData(pool, conf.ServerInfo.ServerType)
-	if err != nil {
-		log.Info("get global metadata failed", zap.Error(err))
+	// for other consistencies, we should get table list after consistency is set up and GlobalMetaData is cached
+	// for other consistencies, record snapshot after whole tables are locked. The recorded meta info is exactly the locked snapshot.
+	if conf.Consistency != "lock" {
+		m.recordStartTime(time.Now())
+		err = m.recordGlobalMetaData(pool, conf.ServerInfo.ServerType)
+		if err != nil {
+			log.Info("get global metadata failed", zap.Error(err))
+		}
+		if err = prepareTableListToDump(conf, pool); err != nil {
+			return err
+		}
 	}
 
 	var writer Writer
@@ -197,6 +200,29 @@ func dumpDatabases(ctx context.Context, conf *Config, db *sql.DB, writer Writer)
 	return nil
 }
 
+func prepareTableListToDump(conf *Config, pool *sql.DB) error {
+	databases, err := prepareDumpingDatabases(conf, pool)
+	if err != nil {
+		return err
+	}
+
+	conf.Tables, err = listAllTables(pool, databases)
+	if err != nil {
+		return err
+	}
+
+	if !conf.NoViews {
+		views, err := listAllViews(pool, databases)
+		if err != nil {
+			return err
+		}
+		conf.Tables.Merge(views)
+	}
+
+	filterTables(conf)
+	return nil
+}
+
 func dumpSql(ctx context.Context, conf *Config, db *sql.DB, writer Writer) error {
 	tableIR, err := SelectFromSql(conf, db)
 	if err != nil {
diff --git a/v4/export/metadata.go b/v4/export/metadata.go
index b233584..36e14af 100644
--- a/v4/export/metadata.go
+++ b/v4/export/metadata.go
@@ -1,20 +1,19 @@
 package export
 
 import (
+	"bytes"
 	"database/sql"
 	"errors"
+	"fmt"
 	"path"
+	"strings"
 	"time"
 )
 
 type globalMetadata struct {
-	logFile string
-	pos     string
-	gtidSet string
+	buffer bytes.Buffer
 
-	filePath   string
-	startTime  time.Time
-	finishTime time.Time
+	filePath string
 }
 
 const (
@@ -31,43 +30,25 @@ const (
 func newGlobalMetadata(outputDir string) *globalMetadata {
 	return &globalMetadata{
 		filePath: path.Join(outputDir, metadataPath),
+		buffer:   bytes.Buffer{},
 	}
 }
 
 func (m globalMetadata) String() string {
-	str := ""
-	if m.startTime.IsZero() {
-		return str
-	}
-	str += "Started dump at: " + m.startTime.Format(metadataTimeLayout) + "\n"
-
-	str += "SHOW MASTER STATUS:\n"
-	if m.logFile != "" {
-		str += "\t\tLog: " + m.logFile + "\n"
-	}
-	if m.pos != "" {
-		str += "\t\tPos: " + m.pos + "\n"
-	}
-	if m.gtidSet != "" {
-		str += "\t\tGTID:" + m.gtidSet + "\n"
-	}
-
-	if m.finishTime.IsZero() {
-		return str
-	}
-	str += "Finished dump at: " + m.finishTime.Format(metadataTimeLayout) + "\n"
-	return str
+	return m.buffer.String()
 }
 
 func (m *globalMetadata) recordStartTime(t time.Time) {
-	m.startTime = t
+	m.buffer.WriteString("Started dump at: " + t.Format(metadataTimeLayout) + "\n")
 }
 
 func (m *globalMetadata) recordFinishTime(t time.Time) {
-	m.finishTime = t
+	m.buffer.WriteString("Finished dump at: " + t.Format(metadataTimeLayout) + "\n")
 }
 
-func (m *globalMetadata) getGlobalMetaData(db *sql.DB, serverType ServerType) error {
+func (m *globalMetadata) recordGlobalMetaData(db *sql.DB, serverType ServerType) error {
+	// get master status info
+	m.buffer.WriteString("SHOW MASTER STATUS:\n")
 	switch serverType {
 	// For MySQL:
 	// mysql> SHOW MASTER STATUS;
@@ -91,9 +72,15 @@ func (m *globalMetadata) getGlobalMetaData(db *sql.DB, serverType ServerType) er
 		if err != nil {
 			return err
 		}
-		m.logFile = str[fileFieldIndex]
-		m.pos = str[posFieldIndex]
-		m.gtidSet = str[gtidSetFieldIndex]
+		if logFile := str[fileFieldIndex]; logFile != "" {
+			m.buffer.WriteString("\tLog: " + logFile + "\n")
+		}
+		if pos := str[posFieldIndex]; pos != "" {
+			m.buffer.WriteString("\tPos: " + pos + "\n")
+		}
+		if gtidSet := str[gtidSetFieldIndex]; gtidSet != "" {
+			m.buffer.WriteString("\tGTID:" + gtidSet + "\n")
+		}
 	// For MariaDB:
 	// SHOW MASTER STATUS;
 	// +--------------------+----------+--------------+------------------+
@@ -113,16 +100,81 @@ func (m *globalMetadata) getGlobalMetaData(db *sql.DB, serverType ServerType) er
 		if err != nil {
 			return err
 		}
-		m.logFile = str[fileFieldIndex]
-		m.pos = str[posFieldIndex]
-		err = db.QueryRow("SELECT @@global.gtid_binlog_pos").Scan(&m.gtidSet)
+		if logFile := str[fileFieldIndex]; logFile != "" {
+			m.buffer.WriteString("\tLog: " + logFile + "\n")
+		}
+		if pos := str[posFieldIndex]; pos != "" {
+			m.buffer.WriteString("\tPos: " + pos + "\n")
+		}
+		var gtidSet string
+		err = db.QueryRow("SELECT @@global.gtid_binlog_pos").Scan(&gtidSet)
 		if err != nil {
 			return err
 		}
+		if gtidSet != "" {
+			m.buffer.WriteString("\tGTID:" + gtidSet + "\n")
+		}
 	default:
-		return errors.New("unsupported serverType" + serverType.String() + "for getGlobalMetaData")
+		return errors.New("unsupported serverType" + serverType.String() + "for recordGlobalMetaData")
+	}
+	m.buffer.WriteString("\n")
+	if serverType == ServerTypeTiDB {
+		return nil
+	}
+	// get follower status info
+	var (
+		isms  bool
+		query string
+	)
+	if err := simpleQuery(db, "SELECT @@default_master_connection", func(rows *sql.Rows) error {
+		isms = true
+		return nil
+	}); err != nil {
+		isms = false
 	}
-	return nil
+	if isms {
+		query = "SHOW ALL SLAVES STATUS"
+	} else {
+		query = "SHOW SLAVE STATUS"
+	}
+	return simpleQuery(db, query, func(rows *sql.Rows) error {
+		cols, err := rows.Columns()
+		if err != nil {
+			return err
+		}
+		data := make([]string, len(cols))
+		args := make([]interface{}, 0, len(cols))
+		for i := range data {
+			args = append(args, &data[i])
+		}
+		if err := rows.Scan(args...); err != nil {
+			return err
+		}
+		var connName, pos, logFile, host, gtidSet string
+		for i, col := range cols {
+			col = strings.ToLower(col)
+			switch col {
+			case "connection_name":
+				connName = data[i]
+			case "exec_master_log_pos":
+				pos = data[i]
+			case "relay_master_log_file":
+				logFile = data[i]
+			case "master_host":
+				host = data[i]
+			case "executed_gtid_set":
+				gtidSet = data[i]
+			}
+		}
+		if len(host) > 0 {
+			m.buffer.WriteString("SHOW SLAVE STATUS:\n")
+			if isms {
+				m.buffer.WriteString("\tConnection name: " + connName + "\n")
+			}
+			fmt.Fprintf(&m.buffer, "\tHost: %s\n\tLog: %s\n\tPos: %s\n\tGTID:%s\n\n", host, logFile, pos, gtidSet)
+		}
+		return nil
+	})
 }
 
 func (m *globalMetadata) writeGlobalMetaData() error {


Test patch:
diff --git a/v4/export/metadata_test.go b/v4/export/metadata_test.go
index 3c332d9..e412ed9 100644
--- a/v4/export/metadata_test.go
+++ b/v4/export/metadata_test.go
@@ -1,9 +1,11 @@
 package export
 
 import (
+	"fmt"
+	"path"
+
 	"github.com/DATA-DOG/go-sqlmock"
 	. "github.com/pingcap/check"
-	"path"
 )
 
 var _ = Suite(&testMetaDataSuite{})
@@ -21,16 +23,52 @@ func (s *testMetaDataSuite) TestMysqlMetaData(c *C) {
 	rows := sqlmock.NewRows([]string{"File", "Position", "Binlog_Do_DB", "Binlog_Ignore_DB", "Executed_Gtid_Set"}).
 		AddRow(logFile, pos, "", "", gtidSet)
 	mock.ExpectQuery("SHOW MASTER STATUS").WillReturnRows(rows)
+	mock.ExpectQuery("SELECT @@default_master_connection").WillReturnError(fmt.Errorf("mock error"))
+	mock.ExpectQuery("SHOW SLAVE STATUS").WillReturnRows(
+		sqlmock.NewRows([]string{"exec_master_log_pos", "relay_master_log_file", "master_host", "Executed_Gtid_Set", "Seconds_Behind_Master"}))
 
 	testFilePath := "/test"
 	m := newGlobalMetadata(testFilePath)
-	c.Assert(m.getGlobalMetaData(db, ServerTypeMySQL), IsNil)
+	c.Assert(m.recordGlobalMetaData(db, ServerTypeMySQL), IsNil)
 	c.Assert(m.filePath, Equals, path.Join(testFilePath, metadataPath))
 
-	c.Assert(m.logFile, Equals, logFile)
-	c.Assert(m.pos, Equals, pos)
-	c.Assert(m.gtidSet, Equals, gtidSet)
+	c.Assert(m.buffer.String(), Equals, "SHOW MASTER STATUS:\n"+
+		"\tLog: ON.000001\n"+
+		"\tPos: 7502\n"+
+		"\tGTID:6ce40be3-e359-11e9-87e0-36933cb0ca5a:1-29\n\n")
+	c.Assert(mock.ExpectationsWereMet(), IsNil)
+}
+
+func (s *testMetaDataSuite) TestMysqlWithFollowersMetaData(c *C) {
+	db, mock, err := sqlmock.New()
+	c.Assert(err, IsNil)
+	defer db.Close()
 
+	logFile := "ON.000001"
+	pos := "7502"
+	gtidSet := "6ce40be3-e359-11e9-87e0-36933cb0ca5a:1-29"
+	rows := sqlmock.NewRows([]string{"File", "Position", "Binlog_Do_DB", "Binlog_Ignore_DB", "Executed_Gtid_Set"}).
+		AddRow(logFile, pos, "", "", gtidSet)
+	followerRows := sqlmock.NewRows([]string{"exec_master_log_pos", "relay_master_log_file", "master_host", "Executed_Gtid_Set", "Seconds_Behind_Master"}).
+		AddRow("256529431", "mysql-bin.001821", "192.168.1.100", gtidSet, 0)
+	mock.ExpectQuery("SHOW MASTER STATUS").WillReturnRows(rows)
+	mock.ExpectQuery("SELECT @@default_master_connection").WillReturnError(fmt.Errorf("mock error"))
+	mock.ExpectQuery("SHOW SLAVE STATUS").WillReturnRows(followerRows)
+
+	testFilePath := "/test"
+	m := newGlobalMetadata(testFilePath)
+	c.Assert(m.recordGlobalMetaData(db, ServerTypeMySQL), IsNil)
+	c.Assert(m.filePath, Equals, path.Join(testFilePath, metadataPath))
+
+	c.Assert(m.buffer.String(), Equals, "SHOW MASTER STATUS:\n"+
+		"\tLog: ON.000001\n"+
+		"\tPos: 7502\n"+
+		"\tGTID:6ce40be3-e359-11e9-87e0-36933cb0ca5a:1-29\n\n"+
+		"SHOW SLAVE STATUS:\n"+
+		"\tHost: 192.168.1.100\n"+
+		"\tLog: mysql-bin.001821\n"+
+		"\tPos: 256529431\n"+
+		"\tGTID:6ce40be3-e359-11e9-87e0-36933cb0ca5a:1-29\n\n")
 	c.Assert(mock.ExpectationsWereMet(), IsNil)
 }
 
@@ -48,14 +86,54 @@ func (s *testMetaDataSuite) TestMariaDBMetaData(c *C) {
 	rows = sqlmock.NewRows([]string{"@@global.gtid_binlog_pos"}).
 		AddRow(gtidSet)
 	mock.ExpectQuery("SELECT @@global.gtid_binlog_pos").WillReturnRows(rows)
+	mock.ExpectQuery("SHOW SLAVE STATUS").WillReturnRows(rows)
 	testFilePath := "/test"
 	m := newGlobalMetadata(testFilePath)
-	c.Assert(m.getGlobalMetaData(db, ServerTypeMariaDB), IsNil)
+	c.Assert(m.recordGlobalMetaData(db, ServerTypeMariaDB), IsNil)
 	c.Assert(m.filePath, Equals, path.Join(testFilePath, metadataPath))
 
-	c.Assert(m.logFile, Equals, logFile)
-	c.Assert(m.pos, Equals, pos)
-	c.Assert(m.gtidSet, Equals, gtidSet)
+	c.Assert(mock.ExpectationsWereMet(), IsNil)
+}
+
+func (s *testMetaDataSuite) TestMariaDBWithFollowersMetaData(c *C) {
+	db, mock, err := sqlmock.New()
+	c.Assert(err, IsNil)
+	defer db.Close()
+
+	logFile := "ON.000001"
+	pos := "7502"
+	gtidSet := "6ce40be3-e359-11e9-87e0-36933cb0ca5a:1-29"
+	rows := sqlmock.NewRows([]string{"File", "Position", "Binlog_Do_DB", "Binlog_Ignore_DB", "Executed_Gtid_Set"}).
+		AddRow(logFile, pos, "", "", gtidSet)
+	followerRows := sqlmock.NewRows([]string{"exec_master_log_pos", "relay_master_log_file", "master_host", "Executed_Gtid_Set", "connection_name", "Seconds_Behind_Master"}).
+		AddRow("256529431", "mysql-bin.001821", "192.168.1.100", gtidSet, "connection_1", 0).
+		AddRow("256529451", "mysql-bin.001820", "192.168.1.102", gtidSet, "connection_2", 200)
+	mock.ExpectQuery("SHOW MASTER STATUS").WillReturnRows(rows)
+	mock.ExpectQuery("SELECT @@default_master_connection").
+		WillReturnRows(sqlmock.NewRows([]string{"@@default_master_connection"}).
+			AddRow("connection_1"))
+	mock.ExpectQuery("SHOW ALL SLAVES STATUS").WillReturnRows(followerRows)
+
+	testFilePath := "/test"
+	m := newGlobalMetadata(testFilePath)
+	c.Assert(m.recordGlobalMetaData(db, ServerTypeMySQL), IsNil)
+	c.Assert(m.filePath, Equals, path.Join(testFilePath, metadataPath))
 
+	c.Assert(m.buffer.String(), Equals, "SHOW MASTER STATUS:\n"+
+		"\tLog: ON.000001\n"+
+		"\tPos: 7502\n"+
+		"\tGTID:6ce40be3-e359-11e9-87e0-36933cb0ca5a:1-29\n\n"+
+		"SHOW SLAVE STATUS:\n"+
+		"\tConnection name: connection_1\n"+
+		"\tHost: 192.168.1.100\n"+
+		"\tLog: mysql-bin.001821\n"+
+		"\tPos: 256529431\n"+
+		"\tGTID:6ce40be3-e359-11e9-87e0-36933cb0ca5a:1-29\n\n"+
+		"SHOW SLAVE STATUS:\n"+
+		"\tConnection name: connection_2\n"+
+		"\tHost: 192.168.1.102\n"+
+		"\tLog: mysql-bin.001820\n"+
+		"\tPos: 256529451\n"+
+		"\tGTID:6ce40be3-e359-11e9-87e0-36933cb0ca5a:1-29\n\n")
 	c.Assert(mock.ExpectationsWereMet(), IsNil)
 }