bugfix: no set tables on start migrator

.gitignore

@@ -15,6 +15,8 @@ operator/operator
 
 *.tar.gz
 
+# docker/migrator/pkg/*.pyc
+
 # swagger lib
 swagger/css
 swagger/fonts

docker/migrator/pkg/config.py

@@ -17,7 +17,7 @@ def __init__(self, db, host, port, user, password, tables=None):
         self.tables = tables
 
     def getDataDir(self):
-        return ROOT + "/" + self.db
+        return ROOT + self.db
 
     def getDumpedMeta(self):
         return self.getDataDir() + "/metadata"

docker/migrator/pkg/migrator.py

@@ -44,7 +44,7 @@ def dump(self):
     # load local data to tidb
     def load(self):
         if not path.isfile(self.src.getDumpedMeta()):
-            self.dump(self.notice)
+            self.dump()
 
         rest.sync_stat(self.notice, 'Loading')
         cmds = self.dest.toLoader()
@@ -56,9 +56,6 @@ def load(self):
             logs.critical("load error")
 
     def sync(self):
-        if not path.isfile(self.src.getCheckpoint()):
-            self.load(self.notice)
-
         self.src.genSyncConfigFile(self.dest)
         if not path.isfile(self.src.getMeta()):
             self.src.genSyncMetaFile()
@@ -151,4 +148,4 @@ def main(argv):
 if __name__ == "__main__":
     main(sys.argv[1:])
 
-# migrator.py --database xinyang1 --src-host 10.213.125.4 --src-port 13306 --src-user root --src-password EJq4dspojdY3FmVF?TYVBkEMB --dest-host 10.213.44.128 --dest-port 14051 --dest-user xinyang1 --dest-password xinyang1 --operator sync --notice http://10.213.44.128:12808/tidb/api/v1/tidbs/006-xinyang1 --tables t1,t2
+# migrator.py --database xinyang1 --src-host 10.213.125.85 --src-port 13306 --src-user root --src-password EJq4dspojdY3FmVF?TYVBkEMB --dest-host 10.213.44.128 --dest-port 14988 --dest-user xinyang1 --dest-password xinyang1 --operator sync --notice http://10.213.44.128:12808/tidb/api/v1/tidbs/006-xinyang1 --tables t1,t2

kubernetes/migrator-job.yaml

@@ -37,6 +37,7 @@ spec:
               --dest-user {{du}} \
               --dest-password {{dp}} \
               --operator {{op}} \
+              --tables "{{tables}}" \
               --notice "{{api}}"
         env: 
         - name: TZ

operator/controllers/db.go

@@ -243,14 +243,17 @@ func (dc *TidbController) Migrate() {
 	}
 	m := &Migrator{
 		Include: true,
+		Notify:  true,
 	}
 	if err := json.Unmarshal(b, m); err != nil {
 		dc.CustomAbort(400, fmt.Sprintf("Parse body error: %v", err))
 	}
 	db, err := operator.GetDb(cell)
 	errHandler(dc.Controller, err, "get db "+cell)
-
-	api := fmt.Sprintf(statAPI, beego.BConfig.Listen.HTTPAddr, beego.BConfig.Listen.HTTPPort, cell)
+	api := ""
+	if m.Notify {
+		api = fmt.Sprintf(statAPI, beego.BConfig.Listen.HTTPAddr, beego.BConfig.Listen.HTTPPort, cell)
+	}
 	errHandler(
 		dc.Controller,
 		db.Migrate(m.Mysql, api, m.Sync, m.Include, m.Tables),
@@ -261,9 +264,10 @@ func (dc *TidbController) Migrate() {
 // Migrator a migrated target
 type Migrator struct {
 	mysqlutil.Mysql `json:",inline"`
-	Include         bool     `json:"include,omitempty"`
+	Include         bool     `json:"include"`
 	Tables          []string `json:"tables,omitempty"`
 	Sync            bool     `json:"sync,omitempty"`
+	Notify          bool     `json:"notify"`
 }
 
 func errHandler(c beego.Controller, err error, msg string) {

operator/db_test.go

@@ -15,7 +15,7 @@ func TestDb_Migrate(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if err = db.stopMigrator(); err != nil {
+	if err = db.StopMigrator(); err != nil {
 		t.Error(err)
 	}
 	time.Sleep(6 * time.Second)
@@ -26,7 +26,7 @@ func TestDb_Migrate(t *testing.T) {
 		User:     "root",
 		Password: "EJq4dspojdY3FmVF?TYVBkEMB",
 	}
-	if err = db.Migrate(src, "", true); err != nil {
+	if err = db.Migrate(src, "", true, true, nil); err != nil {
 		t.Fatal(err)
 	}
 	time.Sleep(10 * time.Second)

operator/migrator.go

@@ -138,6 +138,7 @@ func (db *Db) startMigrator(my *tsql.Migration) (err error) {
 		"{{dh}}", my.Dest.IP, "{{dP}}", fmt.Sprintf("%v", my.Dest.Port),
 		"{{du}}", my.Dest.User, "{{dp}}", my.Dest.Password,
 		"{{op}}", sync,
+		"{{tables}}", strings.Join(my.Tables, ","),
 		"{{api}}", my.NotifyAPI)
 	s := r.Replace(mysqlMigrateYaml)
 	var j []byte

operator/migrator_test.go

@@ -0,0 +1,12 @@
+package operator
+
+import "testing"
+
+func TestDb_StopMigrator(t *testing.T) {
+	db := NewDb()
+	db.Metadata.Name = "006-xinyang1"
+	err := db.StopMigrator()
+	if err != nil {
+		t.Fatal(err)
+	}
+}

operator/template.go

@@ -322,6 +322,7 @@ spec:
               --dest-user {{du}} \
               --dest-password {{dp}} \
               --operator {{op}} \
+              --tables "{{tables}}" \
               --notice "{{api}}"
         env: 
         - name: TZ

pkg/util/k8sutil/job.go

@@ -59,10 +59,14 @@ func CreateAndWaitJob(job *v1.Job, timeout time.Duration) (*v1.Job, error) {
 // DeleteJob delete a job by name
 func DeleteJob(name string) error {
 	err := kubecli.BatchV1().Jobs(Namespace).Delete(name, &metav1.DeleteOptions{})
-	if !apierrors.IsNotFound(err) {
+	if err != nil && !apierrors.IsNotFound(err) {
 		return err
 	}
-	return DeletePodsByLabel(map[string]string{"job-name": name})
+	err = DeletePodsByLabel(map[string]string{"job-name": name})
+	if err != nil {
+		return err
+	}
+	return nil
 }
 
 // GetJob get a job by name

pkg/util/mysqlutil/migrate.go

@@ -66,6 +66,7 @@ func (m *Migration) Check() error {
 				}
 			}
 		}
+		m.Tables = tables
 		cmd += (" " + strings.Join(tables, " "))
 	}
 	o, err := execShell(cmd)

test/endpoints_test.go

@@ -126,18 +126,28 @@ func Test_DeleteDB(t *testing.T) {
 	}
 }
 
+/*
+CREATE TABLE t1 (id INT, age INT, PRIMARY KEY(id)) ENGINE=InnoDB;
+CREATE TABLE t2 (id INT, name VARCHAR(256), PRIMARY KEY(id)) ENGINE=InnoDB;
+CREATE TABLE t_error (
+  c timestamp(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3)
+) ENGINE=InnoDB DEFAULT CHARSET=latin1;
+INSERT INTO t1 VALUES (1, 1), (2, 2), (3, 3);
+INSERT INTO t2 VALUES (1, "a"), (2, "b"), (3, "c");
+*/
 func Test_Migrate(t *testing.T) {
 	m := controllers.Migrator{
 		Mysql: mysqlutil.Mysql{
 			Database: "xinyang1",
-			IP:       "10.213.125.4",
+			IP:       "10.213.125.85",
 			Port:     13306,
 			User:     "xinyang1",
 			Password: "xinyang1",
 		},
 		Include: true,
 		Tables:  []string{"t1", "t2"},
 		Sync:    false,
+		Notify:  false,
 	}
 	body, _ := json.Marshal(m)
 	fmt.Printf("%s", body)