Merge pull request #21 from larsnovikov/refactoring_and_pg_carcass

Refactoring and pg carcass

Author: larsnovikov

files/sql/pg_create_log_func.sql

@@ -0,0 +1,41 @@
+CREATE OR REPLACE FUNCTION horgh.if_modified_func() RETURNS TRIGGER AS $body$
+DECLARE
+    v_old_data TEXT;
+    v_new_data TEXT;
+BEGIN
+    IF (TG_OP = 'UPDATE') THEN
+        v_old_data := ROW(OLD.*);
+        v_new_data := ROW(NEW.*);
+        INSERT INTO horgh.logged_actions (schema_name,table_name,user_name,action,original_data,new_data,query)
+        VALUES (TG_TABLE_SCHEMA::TEXT,TG_TABLE_NAME::TEXT,session_user::TEXT,substring(TG_OP,1,1),v_old_data,v_new_data, current_query());
+        RETURN NEW;
+    ELSIF (TG_OP = 'DELETE') THEN
+        v_old_data := ROW(OLD.*);
+        INSERT INTO horgh.logged_actions (schema_name,table_name,user_name,action,original_data,query)
+        VALUES (TG_TABLE_SCHEMA::TEXT,TG_TABLE_NAME::TEXT,session_user::TEXT,substring(TG_OP,1,1),v_old_data, current_query());
+        RETURN OLD;
+    ELSIF (TG_OP = 'INSERT') THEN
+        v_new_data := ROW(NEW.*);
+        INSERT INTO horgh.logged_actions (schema_name,table_name,user_name,action,new_data,query)
+        VALUES (TG_TABLE_SCHEMA::TEXT,TG_TABLE_NAME::TEXT,session_user::TEXT,substring(TG_OP,1,1),v_new_data, current_query());
+        RETURN NEW;
+    ELSE
+        RAISE WARNING '[HORGH.IF_MODIFIED_FUNC] - Other action occurred: %, at %',TG_OP,now();
+        RETURN NULL;
+    END IF;
+
+EXCEPTION
+    WHEN data_exception THEN
+        RAISE WARNING '[HORGH.IF_MODIFIED_FUNC] - UDF ERROR [DATA EXCEPTION] - SQLSTATE: %, SQLERRM: %',SQLSTATE,SQLERRM;
+        RETURN NULL;
+    WHEN unique_violation THEN
+        RAISE WARNING '[HORGH.IF_MODIFIED_FUNC] - UDF ERROR [UNIQUE] - SQLSTATE: %, SQLERRM: %',SQLSTATE,SQLERRM;
+        RETURN NULL;
+    WHEN OTHERS THEN
+        RAISE WARNING '[HORGH.IF_MODIFIED_FUNC] - UDF ERROR [OTHER] - SQLSTATE: %, SQLERRM: %',SQLSTATE,SQLERRM;
+        RETURN NULL;
+END;
+$body$
+LANGUAGE plpgsql
+SECURITY DEFINER
+SET search_path = pg_catalog, horgh;

files/sql/pg_create_log_table.sql

@@ -0,0 +1,21 @@
+CREATE schema horgh;
+REVOKE CREATE ON schema horgh FROM public;
+
+CREATE TABLE IF NOT EXISTS horgh.%s (
+    id SERIAL,
+    schema_name text NOT NULL,
+    TABLE_NAME text NOT NULL,
+    user_name text,
+    action_tstamp TIMESTAMP WITH TIME zone NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    action TEXT NOT NULL CHECK (action IN ('I','D','U')),
+    original_data text,
+    new_data text,
+    query text
+) WITH (fillfactor=100);
+
+REVOKE ALL ON horgh.%s FROM public;
+
+GRANT SELECT ON horgh.%s TO public;
+
+CREATE INDEX %s_action_idx
+ON horgh.logged_actions(action);

files/sql/pg_create_log_trigger.sql

@@ -0,0 +1,3 @@
+CREATE TRIGGER %s_horgh
+AFTER INSERT OR UPDATE OR DELETE ON horgh.%s
+FOR EACH ROW EXECUTE PROCEDURE horgh.if_modified_func();

src/Gopkg.lock

@@ -1,8 +1,9 @@
-package clickhouse
+package slave
 
 import (
 	"fmt"
 	"horgh-replicator/src/connectors"
+	"horgh-replicator/src/connectors/clickhouse"
 	"horgh-replicator/src/constants"
 	"horgh-replicator/src/helpers"
 	"strings"
@@ -149,6 +150,6 @@ func (model *Model) Exec(params helpers.Query) bool {
 }
 
 func (model *Model) Connection() helpers.Storage {
-	helpers.ConnPool.Slave = GetConnection(helpers.ConnPool.Slave, constants.DBSlave).(helpers.Storage)
+	helpers.ConnPool.Slave = clickhouse.GetConnection(helpers.ConnPool.Slave, constants.DBSlave).(helpers.Storage)
 	return helpers.ConnPool.Slave
 }

src/connectors/clickhouse/model.go renamed to src/connectors/clickhouse/slave/model.go

@@ -0,0 +1,145 @@
+package master
+
+import (
+	"bufio"
+	"fmt"
+	"github.com/siddontang/go-log/log"
+	"github.com/siddontang/go-mysql/mysql"
+	"horgh-replicator/src/constants"
+	"horgh-replicator/src/helpers"
+	"horgh-replicator/src/models/slave"
+	"horgh-replicator/src/tools/exit"
+	toolsHelper "horgh-replicator/src/tools/helpers"
+	"os/exec"
+	"regexp"
+	"strconv"
+	"strings"
+)
+
+const (
+	createDump      = "mysqldump --extended-insert=FALSE --no-create-info --master-data=1 --port=%s -u%s -p%s -h %s %s %s"
+	insertRegexp    = `VALUES \([A-Za-z0-9,\s,\S]+\)`
+	positionRegexp  = `MASTER_LOG_FILE=\'([a-zA-Z\-\.0-9]+)\', MASTER_LOG_POS=([0-9]+)`
+	parseStringSize = 99999999
+)
+
+func buildModel(tableName string) {
+	toolsHelper.Table = tableName
+	if canHandle() == true {
+		toolsHelper.ParseStrings = make(chan string, parseStringSize)
+		go parseLine(toolsHelper.ParseStrings)
+
+		readDump()
+
+		toolsHelper.Wait(func() bool {
+			return slave.GetSlaveByName(toolsHelper.Table).GetChannelLen() == 0 && len(toolsHelper.ParseStrings) == 0
+		})
+	}
+}
+
+func canHandle() bool {
+	savedPos := GetSavedPos(toolsHelper.Table)
+	if savedPos.Name == "" && savedPos.Pos == 0 {
+		return true
+	}
+
+	exit.Fatal(constants.ErrorSlaveBuilt, toolsHelper.Table, toolsHelper.Table)
+	return false
+}
+
+func readDump() {
+	log.Infof(constants.MessageStartReadDump, toolsHelper.Table)
+	cred := helpers.GetCredentials(constants.DBMaster).(helpers.CredentialsDB)
+
+	dumpCmd := fmt.Sprintf(createDump, strconv.Itoa(cred.Port), cred.User, cred.Pass, cred.Host, cred.DBname, toolsHelper.Table)
+	cmdArgs := strings.Fields(dumpCmd)
+	cmd := exec.Command(cmdArgs[0], cmdArgs[1:len(cmdArgs)]...)
+	// create a pipe for the output of the script
+	cmdReader, err := cmd.StdoutPipe()
+	if err != nil {
+		exit.Fatal(constants.ErrorDumpRead, err)
+	}
+
+	scanner := bufio.NewScanner(cmdReader)
+
+	go func() {
+		for scanner.Scan() {
+			toolsHelper.ParseStrings <- scanner.Text()
+		}
+	}()
+
+	err = cmd.Start()
+	if err != nil {
+		exit.Fatal(constants.ErrorDumpRead, err)
+	}
+
+	log.Infof(constants.MessageDumpRead, toolsHelper.Table)
+}
+
+func parseLine(c chan string) {
+	for {
+		line := <-c
+
+		// try to parse like insert
+		if parseInsert(line) == true {
+			continue
+		}
+
+		// try to parse like position setter
+		if parsePosition(line) == true {
+			continue
+		}
+	}
+}
+
+func parseInsert(line string) bool {
+	re := regexp.MustCompile(insertRegexp)
+	match := re.FindStringSubmatch(line)
+	if len(match) > 0 {
+		// TODO fix me
+		r := strings.NewReplacer("VALUES", "",
+			"'", "",
+			"(", "",
+			")", "")
+
+		params := strings.Split(strings.TrimSpace(r.Replace(match[0])), ",")
+
+		slave.GetSlaveByName(toolsHelper.Table).ClearParams()
+
+		interfaceParams := make([]interface{}, len(params))
+		for i := range params {
+			interfaceParams[i] = params[i]
+		}
+		err := ParseRow(slave.GetSlaveByName(toolsHelper.Table), interfaceParams)
+		if err != nil {
+			exit.Fatal(constants.ErrorParseLine, line, err)
+		}
+
+		header, _ := toolsHelper.GetHeader()
+
+		slave.GetSlaveByName(toolsHelper.Table).Insert(&header)
+
+		return true
+	}
+
+	return false
+}
+
+func parsePosition(line string) bool {
+	re := regexp.MustCompile(positionRegexp)
+	match := re.FindStringSubmatch(line)
+
+	if len(match) > 0 {
+		pos, _ := strconv.Atoi(match[2])
+		toolsHelper.Position = mysql.Position{
+			Name: match[1],
+			Pos:  uint32(pos),
+		}
+
+		toolsHelper.SetPosition()
+
+		return true
+	}
+
+	return false
+}

src/connectors/config.go renamed to src/connectors/main.go

@@ -1,4 +1,4 @@
-package parser
+package master
 
 import (
 	"fmt"
@@ -25,13 +25,7 @@ var AllowHandling = true
 var curCanal *canal.Canal
 
 func (h *binlogHandler) canOperate(logTableName string) bool {
-	for _, tableName := range helpers.GetTables() {
-		if tableName == logTableName {
-			return true
-		}
-	}
-
-	return false
+	return helpers.GetTable() == logTableName
 }
 
 func (h *binlogHandler) prepareCanal() {
@@ -110,7 +104,7 @@ func (h *binlogHandler) OnRow(e *canal.RowsEvent) error {
 
 	for i := n; i < len(e.Rows); i += k {
 		if h.ParseBinLog(currentSlave, e, i) != nil {
-			exit.Fatal(constants.ErrorBinlogParsing)
+			exit.Fatal(constants.ErrorLogParsing)
 		}
 
 		if e.Action == canal.UpdateAction {
@@ -163,7 +157,7 @@ func (h *binlogHandler) String() string {
 	return "binlogHandler"
 }
 
-func BinlogListener() {
+func Listen() {
 	c, err := getDefaultCanal()
 	if err == nil {
 		position, err := c.GetMasterPos()

src/connectors/mysql/master/build_slave.go

@@ -0,0 +1,19 @@
+package master
+
+import (
+	"horgh-replicator/src/models/slave"
+	"horgh-replicator/src/tools/exit"
+)
+
+type Model struct {
+}
+
+func (model *Model) Listen() {
+	exit.BeforeExitPool = append(exit.BeforeExitPool, stop)
+	exit.BeforeExitPool = append(exit.BeforeExitPool, slave.Stop)
+	Listen()
+}
+
+func (model *Model) BuildSlave(table string) {
+	buildModel(table)
+}

src/parser/binlog.go renamed to src/connectors/mysql/master/listener.go

@@ -1,4 +1,4 @@
-package parser
+package master
 
 import (
 	"github.com/siddontang/go-log/log"
@@ -127,7 +127,7 @@ func prepareType(fieldName string, fieldType string, value interface{}, params m
 	}
 }
 
-func Stop() bool {
+func stop() bool {
 	// stop handle binlog
 	log.Infof(constants.MessageStopHandlingBinlog)
 	AllowHandling = false