*: add cleanup handle #569

[summary]
add cleanup admin to support clean up the old data after shifted.
syntax:
radon cleanup;
[test case]
src/ctl/v1/shard_test.go
src/proxy/admin_cleanup_test.go
src/proxy/proxy_test.go
src/proxy/radon_test.go
src/vendor/github.com/xelabs/go-mysqlstack/sqlparser/radon_test.go
[patch codecov]
src/ctl/v1/shard.go 90.9%
src/proxy/admin_cleanup.go 100%
src/proxy/proxy.go 85.6%
src/proxy/radon.go 77.8%
src/vendor/github.com/xelabs/go-mysqlstack/sqlparser/ast.go 85.9%

Author: zhyass

docs/api.md

@@ -342,7 +342,6 @@ This api is used to migrate the data from one backend to another.
 Path:    /v1/shard/migrate
 Method:  POST
 Request: {
-			"to-flavor":                 "Destination db flavor, like mysql/mariadb/radondb, default mysql"
 			"from":                      "Source MySQL backend(host:port)",                                   [required]
 			"from-user":                 "MySQL user, must have replication privilege",                       [required]
 			"from-password":             "MySQL user password",                                               [required]

src/ctl/v1/shard.go

@@ -9,15 +9,16 @@
 package v1
 
 import (
+	"fmt"
 	"net/http"
 	"regexp"
 	"strconv"
 	"strings"
 
+	"plugins/shiftmanager"
 	"proxy"
 
 	"github.com/ant0ine/go-json-rest/rest"
-	"github.com/radondb/shift/shift"
 	shiftlog "github.com/radondb/shift/xlog"
 	"github.com/xelabs/go-mysqlstack/xlog"
 )
@@ -360,8 +361,6 @@ func globalsHandler(log *xlog.Log, proxy *proxy.Proxy, w rest.ResponseWriter, r
 }
 
 type migrateParams struct {
-	ToFlavor string `json:"to-flavor"`
-
 	From         string `json:"from"`
 	FromUser     string `json:"from-user"`
 	FromPassword string `json:"from-password"`
@@ -388,7 +387,8 @@ type migrateParams struct {
 // Returns:
 // 1. Status:200
 // 2. Status:204
-// 3. Status:500
+// 3. Status:403
+// 4. Status:500
 func ShardMigrateHandler(log *xlog.Log, proxy *proxy.Proxy) rest.HandlerFunc {
 	f := func(w rest.ResponseWriter, r *rest.Request) {
 		shardMigrateHandler(proxy, w, r)
@@ -397,9 +397,9 @@ func ShardMigrateHandler(log *xlog.Log, proxy *proxy.Proxy) rest.HandlerFunc {
 }
 
 func shardMigrateHandler(proxy *proxy.Proxy, w rest.ResponseWriter, r *rest.Request) {
+	scatter := proxy.Scatter()
 	log := shiftlog.NewStdLog(shiftlog.Level(shiftlog.INFO))
 	p := &migrateParams{
-		ToFlavor:               shift.ToMySQLFlavor,
 		RadonURL:               "http://" + proxy.Config().Proxy.PeerAddress,
 		Rebalance:              false,
 		Cleanup:                false,
@@ -423,18 +423,30 @@ func shardMigrateHandler(proxy *proxy.Proxy, w rest.ResponseWriter, r *rest.Requ
 	}
 
 	// check args.
-	if len(p.From) == 0 || len(p.FromUser) == 0 || len(p.FromDatabase) == 0 || len(p.FromTable) == 0 ||
-		len(p.To) == 0 || len(p.ToUser) == 0 || len(p.ToDatabase) == 0 || len(p.ToTable) == 0 {
+	if len(p.FromUser) == 0 || len(p.FromDatabase) == 0 || len(p.FromTable) == 0 ||
+		len(p.ToUser) == 0 || len(p.ToDatabase) == 0 || len(p.ToTable) == 0 {
 		log.Error("api.v1.shard.migrate[%+v].error:some param is empty", p)
 		rest.Error(w, "some args are empty", http.StatusNoContent)
 		return
 	}
-	log.Warning(`
-           IMPORTANT: Please check that the shift run completes successfully.
-           At the end of a successful shift run prints "shift.completed.OK!".`)
 
-	cfg := &shift.Config{
-		ToFlavor:               p.ToFlavor,
+	// Check the backend name.
+	var fromBackend, toBackend string
+	backends := scatter.BackendConfigsClone()
+	for _, backend := range backends {
+		if backend.Address == p.From {
+			fromBackend = backend.Name
+		} else if backend.Address == p.To {
+			toBackend = backend.Name
+		}
+	}
+	if fromBackend == "" || toBackend == "" {
+		log.Error("api.v1.shard.migrate.fromBackend[%s].or.toBackend[%s].is.NULL", fromBackend, toBackend)
+		rest.Error(w, "api.v1.shard.migrate.backend.NULL", http.StatusInternalServerError)
+		return
+	}
+
+	cfg := &shiftmanager.ShiftInfo{
 		From:                   p.From,
 		FromUser:               p.FromUser,
 		FromPassword:           p.FromPassword,
@@ -447,23 +459,26 @@ func shardMigrateHandler(proxy *proxy.Proxy, w rest.ResponseWriter, r *rest.Requ
 		ToTable:                p.ToTable,
 		Rebalance:              p.Rebalance,
 		Cleanup:                p.Cleanup,
-		MySQLDump:              p.MySQLDump,
+		MysqlDump:              p.MySQLDump,
 		Threads:                p.Threads,
-		Behinds:                p.Behinds,
+		PosBehinds:             p.Behinds,
 		RadonURL:               p.RadonURL,
 		Checksum:               p.Checksum,
 		WaitTimeBeforeChecksum: p.WaitTimeBeforeChecksum,
 	}
-	log.Info("shift.cfg:%+v", cfg)
 
-	shift := shift.NewShift(log, cfg)
-	if err := shift.Start(); err != nil {
+	shiftMgr := proxy.Plugins().PlugShiftMgr()
+	shift, _ := shiftMgr.NewShiftInstance(cfg, shiftmanager.ShiftTypeRebalance)
+
+	key := fmt.Sprintf("`%s`.`%s`_%s", p.ToDatabase, p.ToTable, toBackend)
+	err = shiftMgr.StartShiftInstance(key, shift, shiftmanager.ShiftTypeRebalance)
+	if err != nil {
 		log.Error("shift.start.error:%+v", err)
 		rest.Error(w, err.Error(), http.StatusInternalServerError)
 		return
 	}
 
-	err = shift.WaitFinish()
+	err = shiftMgr.WaitInstanceFinish(key)
 	if err != nil {
 		log.Error("shift.wait.finish.error:%+v", err)
 		rest.Error(w, err.Error(), http.StatusInternalServerError)

src/ctl/v1/shard_test.go

@@ -1402,13 +1402,12 @@ func TestCtlV1ShardMigrateErr(t *testing.T) {
 		}
 	}
 	p := &migrateParams{
-		ToFlavor:               "mysql",
-		From:                   "127.0.0.1" + from,
+		From:                   from,
 		FromUser:               fromUsr,
 		FromPassword:           fromPasswd,
 		FromDatabase:           "test",
 		FromTable:              "a",
-		To:                     "127.0.0.1" + to,
+		To:                     to,
 		ToUser:                 toUsr,
 		ToPassword:             toPasswd,
 		ToDatabase:             "test",
@@ -1475,6 +1474,14 @@ func TestCtlV1ShardMigrateErr(t *testing.T) {
 		recorded.CodeIs(204)
 	}
 
+	// check backend null.
+	{
+		p.ToTable = "a"
+		p.To = "192.168.0.1:3306"
+		recorded := test.RunRequest(t, handler, test.MakeSimpleRequest("POST", "http://localhost/v1/shard/migrate", p))
+		recorded.CodeIs(500)
+	}
+
 	// Set readonly.
 	{
 		proxy.SetReadOnly(true)

src/plugins/shiftmanager/handler.go

@@ -60,6 +60,7 @@ type ShiftInfo struct {
 	ToDatabase string
 	ToTable    string
 
+	Rebalance              bool
 	Cleanup                bool
 	Checksum               bool
 	MysqlDump              string

src/plugins/shiftmanager/shiftmanager.go

@@ -88,6 +88,7 @@ func (shiftMgr *ShiftManager) NewShiftInstance(shiftInfo *ShiftInfo, typ ShiftTy
 		ToPassword:             shiftInfo.ToPassword,
 		ToDatabase:             shiftInfo.ToDatabase,
 		ToTable:                shiftInfo.ToTable,
+		Rebalance:              shiftInfo.Rebalance,
 		Cleanup:                shiftInfo.Cleanup,
 		MySQLDump:              shiftInfo.MysqlDump,
 		Threads:                shiftInfo.Threads,

src/proxy/admin_cleanup.go

@@ -0,0 +1,124 @@
+package proxy
+
+import (
+	"fmt"
+	"strings"
+
+	"backend"
+	"plugins/shiftmanager"
+	"router"
+
+	"github.com/xelabs/go-mysqlstack/sqlparser/depends/sqltypes"
+	"github.com/xelabs/go-mysqlstack/xlog"
+)
+
+// Cleanup ...
+type Cleanup struct {
+	log     *xlog.Log
+	scatter *backend.Scatter
+	router  *router.Router
+	spanner *Spanner
+}
+
+// NewCleanup -- creates new Cleanup handler.
+func NewCleanup(log *xlog.Log, scatter *backend.Scatter, router *router.Router, spanner *Spanner) *Cleanup {
+	return &Cleanup{
+		log:     log,
+		scatter: scatter,
+		router:  router,
+		spanner: spanner,
+	}
+}
+
+// Cleanup used to find and cleanup the old data.
+func (c *Cleanup) Cleanup() (*sqltypes.Result, error) {
+	backends := c.scatter.AllBackends()
+	for _, backend := range backends {
+		dbQuery := "show databases"
+		qr, err := c.spanner.ExecuteOnThisBackend(backend, dbQuery)
+		if err != nil {
+			return nil, err
+		}
+
+		for _, r := range qr.Rows {
+			db := string(r[0].Raw())
+			err := c.cleanupHandler(db, backend)
+			if err != nil {
+				return nil, err
+			}
+		}
+	}
+
+	return &sqltypes.Result{}, nil
+}
+
+func (c *Cleanup) cleanupHandler(database string, backend string) error {
+	log := c.log
+	var err error
+	if isSysDB := c.router.IsSystemDB(database); isSysDB {
+		return nil
+	}
+
+	// 1.Check if the database is in router. If not, drop database.
+	if err = c.router.CheckDatabase(database); err != nil {
+		sql := fmt.Sprintf("drop database if exists `%s`", database)
+		if _, err = c.spanner.ExecuteOnThisBackend(backend, sql); err != nil {
+			log.Error("cleanup.drop.database[%s].on.backend[%s].error:%v", database, backend, err)
+			return err
+		}
+		log.Warning("cleanup.database[%s].on.backend[%s].has.been.cleaned", database, backend)
+		return nil
+	}
+
+	// 2. Find the table with suffix '_cleanup', check whether exist in radon. If not, drop table.
+	sql := fmt.Sprintf("select table_name from information_schema.tables where table_schema = '%s' and table_name like '%%_cleanup'", database)
+	qr, err := c.spanner.ExecuteOnThisBackend(backend, sql)
+	if err != nil {
+		return err
+	}
+
+	for _, r := range qr.Rows {
+		tb := string(r[0].Raw())
+		// Because partition table suffix is a 4-digit number,
+		// here can only be a global table or a single table.
+		// Therefore, we can directly check in the router.
+		if isExist, _ := c.router.CheckTable(database, tb); !isExist {
+			sql := fmt.Sprintf("drop table if exists `%s`.`%s`", database, tb)
+			_, err = c.spanner.ExecuteOnThisBackend(backend, sql)
+			if err != nil {
+				log.Error("cleanup.drop.table[%s.%s].on.backend[%s].error:%v", database, tb, backend, err)
+				return err
+			}
+			log.Warning("cleanup.table[%s.%s].on.backend[%s].has.been.cleaned", database, tb, backend)
+		}
+	}
+
+	// 3. Find the table with suffix '_migrate', check whether migrating or exist in radon. If not, drop table.
+	sql = fmt.Sprintf("select table_name from information_schema.tables where table_schema = '%s' and table_name like '%%_migrate'", database)
+	qr, err = c.spanner.ExecuteOnThisBackend(backend, sql)
+	if err != nil {
+		return err
+	}
+
+	for _, r := range qr.Rows {
+		tb := string(r[0].Raw())
+		if isExist, _ := c.router.CheckTable(database, tb); !isExist {
+			// Check if the table is migrating.
+			key := fmt.Sprintf("`%s`.`%s`_%s", database, strings.TrimSuffix(tb, "_migrate"), backend)
+			shiftMgr := c.spanner.plugins.PlugShiftMgr()
+			status := shiftMgr.GetStatus(key)
+			if status == shiftmanager.ShiftStatusMigrating {
+				continue
+			}
+
+			sql := fmt.Sprintf("drop table if exists `%s`.`%s`", database, tb)
+			_, err = c.spanner.ExecuteOnThisBackend(backend, sql)
+			if err != nil {
+				log.Error("cleanup.drop.table[%s.%s].on.backend[%s].error:%v", database, tb, backend, err)
+				return err
+			}
+			log.Warning("cleanup.table[%s.%s].on.backend[%s].has.been.cleaned", database, tb, backend)
+		}
+	}
+	return nil
+}