Merge pull request #203 from XiaoMi/master

merge dev

backend/direct_connection.go

@@ -453,7 +453,7 @@ func (dc *DirectConnection) SetAutoCommit(v uint8) error {
 func (dc *DirectConnection) SetCharset(charset string, collation mysql.CollationID) ( /*changed*/ bool, error) {
 	charset = strings.Trim(charset, "\"'`")
 
-	if collation == 0 {
+	if collation == 0 || collation > 247 {
 		collation = mysql.CollationNames[mysql.Charsets[charset]]
 	}
 

backend/direct_connection_test.go

@@ -15,6 +15,9 @@ package backend
 
 import (
 	"bytes"
+	"github.com/XiaoMi/Gaea/mysql"
+	"github.com/XiaoMi/Gaea/util/mocks/pipeTest"
+	"github.com/stretchr/testify/require"
 	"testing"
 )
 
@@ -37,3 +40,118 @@ func TestAppendSetVariable2(t *testing.T) {
 	appendSetVariableToDefault(&buf, "sql_mode")
 	t.Log(buf.String())
 }
+
+var (
+	// 准备数据库的回应资料
+	mysqlInitHandShakeResponse = []uint8{
+		// 资料长度
+		93, 0, 0,
+		// 自增序列号码
+		0,
+		// 以下 93 笔数据
+		// 数据库的版本号 version
+		10, 53, 46, 53, 46,
+		53, 45, 49, 48, 46,
+		53, 46, 49, 50, 45,
+		77, 97, 114, 105, 97,
+		68, 66, 45, 108, 111,
+		103,
+		// 数据库的版本结尾
+		0,
+		// 连线编号 connection id
+		16, 0, 0, 0,
+		// Salt
+		81, 64, 43, 85, 76, 90, 97, 91,
+		// filter
+		0,
+		// 取得功能标志 capability
+		254, 247,
+		// 數據庫編碼 charset
+		33, // 可以用 SHOW CHARACTER SET LIKE 'utf8'; 查询
+		// 服务器状态，在 Gaea/mysql/constants.go 的 Server information
+		2, 0,
+		// 延伸的功能标志 capability
+		255, 129,
+		// Auth 资料和保留值
+		21, 0, 0, 0, 0, 0, 0, 15, 0, 0, 0,
+		// 延伸的 Salt
+		34, 53, 36, 85,
+		93, 86, 117, 105, 49,
+		87, 65, 125,
+		// 其他未用到的资料
+		0, 109, 121, 115, 113, 108, 95, 110, 97, 116,
+		105, 118, 101, 95, 112, 97, 115, 115, 119, 111,
+		114, 100, 0,
+	}
+)
+
+// TestDirectConnWithoutDB 为测试数据库的后端连线流程，以下测试不使用 MariaDB 的服务器，只是单纯的单元测试
+func TestDirectConnWithoutDB(t *testing.T) {
+	// 开始正式测试
+	t.Run("测试数据库后端连线的初始交握", func(t *testing.T) {
+		// 开始模拟
+		mockGaea, mockMariaDB := pipeTest.NewDcServerClient(t, pipeTest.TestReplyFunc) // 产生 Gaea 和 mockServer 模拟物件
+		mockGaea.SendOrReceive(mysqlInitHandShakeResponse)                             // 模拟数据库开始交握
+
+		// 產生 Mysql dc 直連物件 (用以下内容取代 reply() 函数 !)
+		var dc DirectConnection
+		var mysqlConn = mysql.NewConn(mockMariaDB.GetConnRead())
+		dc.conn = mysqlConn
+		err := dc.readInitialHandshake()
+		require.Equal(t, err, nil)
+
+		// 等待和确认资料已经写入 pipe 并单方向重置模拟物件
+		err = mockGaea.WaitAndReset(mockMariaDB)
+		require.Equal(t, err, nil)
+
+		// 开始计算
+
+		/* 功能标志 capability 的计算
+		先把所有的功能标志 capability 的数据收集起来，包含延伸部份
+		数值分别为 254, 247, 255, 129
+		并反向排列
+		数值分别为 129, 255, 247, 254
+		全部 十进制 转成 二进制
+		254 的二进制为 1111 1110
+		247 的二进制为 1111 0111
+		255 的二进制为 1111 1111
+		129 的二进制为 1000 0001
+		把全部二进制的数值合并
+		二进制数值分别为 1000 0001 1111 1111 1111 0111 1111 1110 (转成十进制数值为 2181036030)
+		再用文档 https://mariadb.com/kb/en/connection/ 进行对照
+		比如，功能标志 capability 的第一个值为 0，意思为 CLIENT_MYSQL 值为 0，代表是由服务器发出的讯息 */
+
+		/* 连线编号 connection id 的计算
+		先把所有的连线编号 connection id 的数据收集起来，包含延伸部份
+		数值分别为 16, 0, 0, 0
+		并反向排列
+		数值分别为 0, 0, 0, 16
+		全部 十进制 转成 二进制
+		  0 的二进制为 0000 0000
+		 16 的二进制为 0001 0000
+		把全部二进制的数值合并
+		二进制数值分别为 0000 0000 0001 0000 (转成十进制数值为 16) */
+
+		// 先把所有 Salt 的数据收集起来，包含延伸部份
+		// 数值分别为 81,64,43,85,76,90,97,91,34,53,36,85,93,86,117,105,49,87,65,125
+
+		/* 服务器状态 status 的计算
+		先把所有的服务器状态 的数据收集起来，包含延伸部份
+		数值分别为 2, 0
+		并反向排列
+		数值分别为 0, 2
+		全部 十进制 转成 二进制
+		2 的二进制为 0000 0010
+		0 的二进制为 0000 0000
+		把全部二进制的数值合并
+		二进制数值分别为 0000 0000 0000 0010 (转成十进制数值为 2)
+		再用代码 Gaea/mysql/constants.go 里的 Server information 进行对照
+		功能标志 capability 的第一个值为 0，意思为 CLIENT_MYSQL 值为 0，代表是由服务器发出的讯息 */
+
+		// 计算后的检查
+		require.Equal(t, dc.capability, uint32(2181036030))                                                                   // 检查功能标志 capability
+		require.Equal(t, dc.conn.ConnectionID, uint32(16))                                                                    // 检查连线编号 connection id
+		require.Equal(t, dc.salt, []uint8{81, 64, 43, 85, 76, 90, 97, 91, 34, 53, 36, 85, 93, 86, 117, 105, 49, 87, 65, 125}) // 检查 Salt
+		require.Equal(t, dc.status, mysql.ServerStatusAutocommit)                                                             // 检查服务器状态
+	})
+}

cc/service/service.go

@@ -23,6 +23,11 @@ import (
 	"github.com/XiaoMi/Gaea/models"
 )
 
+const (
+	PREPARE_RETRY_TIMES = 3
+	COMMIT_RETRY_TIMES  = 1
+)
+
 func getCoordinatorRoot(cluster string) string {
 	if cluster != "" {
 		return "/" + cluster
@@ -32,15 +37,15 @@ func getCoordinatorRoot(cluster string) string {
 
 // ListNamespace return names of all namespace
 func ListNamespace(cfg *models.CCConfig, cluster string) ([]string, error) {
-	client := models.NewClient(models.ConfigEtcd, cfg.CoordinatorAddr, cfg.UserName, cfg.Password, getCoordinatorRoot(cluster))
+	client := models.NewClient(cfg.CoordinatorType, cfg.CoordinatorAddr, cfg.UserName, cfg.Password, getCoordinatorRoot(cluster))
 	mConn := models.NewStore(client)
 	defer mConn.Close()
 	return mConn.ListNamespace()
 }
 
 // QueryNamespace return information of namespace specified by names
 func QueryNamespace(names []string, cfg *models.CCConfig, cluster string) (data []*models.Namespace, err error) {
-	client := models.NewClient(models.ConfigEtcd, cfg.CoordinatorAddr, cfg.UserName, cfg.Password, getCoordinatorRoot(cluster))
+	client := models.NewClient(cfg.CoordinatorType, cfg.CoordinatorAddr, cfg.UserName, cfg.Password, getCoordinatorRoot(cluster))
 	mConn := models.NewStore(client)
 	defer mConn.Close()
 	for _, v := range names {
@@ -71,7 +76,7 @@ func ModifyNamespace(namespace *models.Namespace, cfg *models.CCConfig, cluster
 	}
 
 	// sink namespace
-	client := models.NewClient(models.ConfigEtcd, cfg.CoordinatorAddr, cfg.UserName, cfg.Password, getCoordinatorRoot(cluster))
+	client := models.NewClient(cfg.CoordinatorType, cfg.CoordinatorAddr, cfg.UserName, cfg.Password, getCoordinatorRoot(cluster))
 	storeConn := models.NewStore(client)
 	defer storeConn.Close()
 
@@ -87,28 +92,48 @@ func ModifyNamespace(namespace *models.Namespace, cfg *models.CCConfig, cluster
 		return err
 	}
 
+	wg := sync.WaitGroup{}
 	// prepare phase
 	for _, v := range proxies {
-		err := proxy.PrepareConfig(v.IP+":"+v.AdminPort, namespace.Name, cfg)
-		if err != nil {
-			return err
-		}
+		wg.Add(1)
+		go func(v *models.ProxyMonitorMetric) {
+			for i := 0; i < PREPARE_RETRY_TIMES; i++ {
+				if err = proxy.PrepareConfig(v.IP+":"+v.AdminPort, namespace.Name, cfg); err == nil {
+					break
+				}
+				log.Warn("namespace %s, proxy prepare retry %d", namespace.Name, i)
+			}
+			if err != nil {
+				return
+			}
+			wg.Done()
+		}(v)
 	}
+	wg.Wait()
 
 	// commit phase
 	for _, v := range proxies {
-		err := proxy.CommitConfig(v.IP+":"+v.AdminPort, namespace.Name, cfg)
-		if err != nil {
-			return err
-		}
+		wg.Add(1)
+		go func(v *models.ProxyMonitorMetric) {
+			for i := 0; i < COMMIT_RETRY_TIMES; i++ {
+				if err := proxy.CommitConfig(v.IP+":"+v.AdminPort, namespace.Name, cfg); err == nil {
+					break
+				}
+				log.Warn("namespace %s, proxy prepare retry %d", namespace.Name, i)
+			}
+			if err != nil {
+				return
+			}
+			wg.Done()
+		}(v)
 	}
 
 	return nil
 }
 
 // DelNamespace delete namespace
 func DelNamespace(name string, cfg *models.CCConfig, cluster string) error {
-	client := models.NewClient(models.ConfigEtcd, cfg.CoordinatorAddr, cfg.UserName, cfg.Password, getCoordinatorRoot(cluster))
+	client := models.NewClient(cfg.CoordinatorType, cfg.CoordinatorAddr, cfg.UserName, cfg.Password, getCoordinatorRoot(cluster))
 	mConn := models.NewStore(client)
 	defer mConn.Close()
 
@@ -139,7 +164,7 @@ func SQLFingerprint(name string, cfg *models.CCConfig, cluster string) (slowSQLs
 	slowSQLs = make(map[string]string, 16)
 	errSQLs = make(map[string]string, 16)
 	// list proxy
-	client := models.NewClient(models.ConfigEtcd, cfg.CoordinatorAddr, cfg.UserName, cfg.Password, getCoordinatorRoot(cluster))
+	client := models.NewClient(cfg.CoordinatorType, cfg.CoordinatorAddr, cfg.UserName, cfg.Password, getCoordinatorRoot(cluster))
 	mConn := models.NewStore(client)
 	defer mConn.Close()
 	proxies, err := mConn.ListProxyMonitorMetrics()
@@ -183,7 +208,7 @@ func SQLFingerprint(name string, cfg *models.CCConfig, cluster string) (slowSQLs
 // ProxyConfigFingerprint return fingerprints of all proxy
 func ProxyConfigFingerprint(cfg *models.CCConfig, cluster string) (r map[string]string, err error) {
 	// list proxy
-	client := models.NewClient(models.ConfigEtcd, cfg.CoordinatorAddr, cfg.UserName, cfg.Password, getCoordinatorRoot(cluster))
+	client := models.NewClient(cfg.CoordinatorType, cfg.CoordinatorAddr, cfg.UserName, cfg.Password, getCoordinatorRoot(cluster))
 	mConn := models.NewStore(client)
 	defer mConn.Close()
 	proxies, err := mConn.ListProxyMonitorMetrics()

etc/gaea.ini

@@ -1,4 +1,5 @@
-; config type, etcd/file, you can test gaea with file type, you shoud use etcd in production
+; config type, etcd/file/etcdv3, you can test gaea with file type, you shoud use etcd/etcdv3 in production
+; 请指定设定方式为 file 或 etcd 或 etcdv3
 config_type=etcd
 ;file config path, 具体配置放到file_config_path的namespace目录下，该下级目录为固定目录
 file_config_path=./etc/file

etc/gaea_cc.ini

@@ -13,7 +13,8 @@ log_path =./logs
 log_filename=gaea_cc
 log_output=file
 
-;coordinator目前支持etcd，coodinator config
+;coordinator 目前支持 etcd 和 etcdv3，coodinator config
+coordinator_type=etcd
 coordinator_addr=http://127.0.0.1:2379
 username=root
 password=root

models/README.md

@@ -0,0 +1,17 @@
+# 中间件 Gaea Model 目录内容
+
+
+
+## 1 功能列表
+
+Model 目录包含以下功能
+
+- 处理设定值的读写逻辑
+
+
+
+## 2 相关文档
+
+以下为相关文档
+
+- [指定设定值的读写方式 (file etcd etcdv3)](./connection.md)

models/cc.go

@@ -28,6 +28,7 @@ type CCConfig struct {
 	ProxyUserName string `ini:"proxy_username"`
 	ProxyPassword string `ini:"proxy_password"`
 	// etcd 相关配置
+	CoordinatorType string `ini:"coordinator_type"`
 	CoordinatorAddr string `ini:"coordinator_addr"`
 	CoordinatorRoot string `ini:"coordinator_root"`
 	UserName        string `ini:"username"`
@@ -56,6 +57,9 @@ func ParseCCConfig(cfgFile string) (*CCConfig, error) {
 	if ccConfig.DefaultCluster == "" && ccConfig.CoordinatorRoot != "" {
 		ccConfig.DefaultCluster = strings.TrimPrefix(ccConfig.CoordinatorRoot, "/")
 	}
+	if ccConfig.CoordinatorType == "" {
+		ccConfig.CoordinatorType = ConfigEtcd
+	}
 	return ccConfig, err
 }