backend refactor

Author: jethe

backend/backend_conn.go

@@ -53,6 +53,10 @@ type Conn struct {
 	pkgErr        error
 }
 
+func (c *Conn) GetTCPConnect() net.Conn {
+	return c.conn
+}
+
 func (c *Conn) Connect(addr string, user string, password string, db string) error {
 	c.addr = addr
 	c.user = user
@@ -133,6 +137,14 @@ func (c *Conn) Close() error {
 	return nil
 }
 
+func (c *Conn) Read() ([]byte, error) {
+	return c.readPacket()
+}
+
+func (c *Conn) Write(data []byte) error {
+	return c.writePacket(data)
+}
+
 func (c *Conn) readPacket() ([]byte, error) {
 	d, err := c.pkg.ReadPacket()
 	c.pkgErr = err

backend/db.go

@@ -247,6 +247,7 @@ func (db *DB) PopConn() (*Conn, error) {
 	if cacheConns == nil || idleConns == nil {
 		return nil, errors.ErrDatabaseClose
 	}
+
 	co = db.GetConnFromCache(cacheConns)
 	if co == nil {
 		co, err = db.GetConnFromIdle(cacheConns, idleConns)

mysql/charset.go

@@ -549,7 +549,7 @@ var CollationNames = map[string]CollationId{
 }
 
 const (
-	DEFAULT_CHARSET                    = "utf8"
-	DEFAULT_COLLATION_ID   CollationId = 33
-	DEFAULT_COLLATION_NAME string      = "utf8_general_ci"
+	DEFAULT_CHARSET                    = "utf8mb4"
+	DEFAULT_COLLATION_ID   CollationId = 224
+	DEFAULT_COLLATION_NAME string      = "utf8mb4_unicode_ci"
 )

mysql/const.go

@@ -18,7 +18,7 @@ const (
 	MinProtocolVersion byte   = 10
 	MaxPayloadLen      int    = 1<<24 - 1
 	TimeFormat         string = "2006-01-02 15:04:05"
-	ServerVersion      string = "5.6.20-kingshard"
+	ServerVersion      string = "0.1-anywhere-mysql-load-balancer"
 )
 
 const (

proxy/server/conn.go

@@ -17,7 +17,6 @@ package server
 import (
 	"bytes"
 	"encoding/binary"
-	"fmt"
 	"net"
 	"runtime"
 	"sync"
@@ -57,10 +56,6 @@ type ClientConn struct {
 
 	lastInsertId int64
 	affectedRows int64
-
-	// stmtId uint32
-
-	// stmts map[uint32]*Stmt //prepare相关,client端到proxy的stmt
 }
 
 var DEFAULT_CAPABILITY uint32 = mysql.CLIENT_LONG_PASSWORD | mysql.CLIENT_LONG_FLAG |
@@ -69,28 +64,6 @@ var DEFAULT_CAPABILITY uint32 = mysql.CLIENT_LONG_PASSWORD | mysql.CLIENT_LONG_F
 
 var baseConnId uint32 = 10000
 
-func (c *ClientConn) IsAllowConnect() bool {
-	clientHost, _, err := net.SplitHostPort(c.c.RemoteAddr().String())
-	if err != nil {
-		fmt.Println(err)
-	}
-	clientIP := net.ParseIP(clientHost)
-
-	ipVec := c.proxy.allowips[c.proxy.allowipsIndex]
-	if ipVecLen := len(ipVec); ipVecLen == 0 {
-		return true
-	}
-	for _, ip := range ipVec {
-		if ip.Equal(clientIP) {
-			return true
-		}
-	}
-
-	golog.Error("server", "IsAllowConnect", "error", mysql.ER_ACCESS_DENIED_ERROR,
-		"ip address", c.c.RemoteAddr().String(), " access denied by kindshard.")
-	return false
-}
-
 func (c *ClientConn) Handshake() error {
 	if err := c.writeInitialHandshake(); err != nil {
 		golog.Error("server", "Handshake", err.Error(),
@@ -248,14 +221,49 @@ func (c *ClientConn) readHandshakeResponse() error {
 
 	}
 
-	golog.Error("handshake ", "response", "db ", 0, db)
 	if err := c.useDB(db); err != nil {
 		return err
 	}
 
 	return nil
 }
 
+func (c *ClientConn) DoStreamRoute() (err error) {
+	srcConn := c.c.(*net.TCPConn)
+	golog.Info("ClientConn", "Run", "address Info client end", 0, srcConn.RemoteAddr().String(), srcConn.LocalAddr().String())
+
+	//find default backend
+	node := c.proxy.GetNode("node1")
+	backendConn, err := c.getBackendConn(node, true)
+	dstConn := backendConn.Conn.GetTCPConnect().(*net.TCPConn)
+
+	golog.Info("ClientConn", "Run", "address Info backend ", 0, dstConn.RemoteAddr().String(), dstConn.LocalAddr().String())
+	if err != nil {
+		golog.Error("ClientConn", "Run getBackendConn", err.Error(), 0)
+	}
+
+	var buf bytes.Buffer
+	n, err := buf.ReadFrom(srcConn)
+
+	if err != nil {
+		golog.Error("ClientConn", "Run", "route send bytes error", 0, n)
+	}
+
+	golog.Info("ClientConn", "Run", "proxy send server bytes", 0, n)
+
+	n, err = buf.WriteTo(dstConn)
+
+	if err != nil {
+		golog.Error("ClientConn", "Run", "route receive bytes error", 0, n)
+	}
+
+	golog.Info("ClientConn", "Run", "proxy receive server bytes", 0, n)
+	buf.Reset()
+	buf.ReadFrom(dstConn)
+	buf.WriteTo(srcConn)
+	return
+}
+
 func (c *ClientConn) Run() {
 	defer func() {
 		r := recover()
@@ -268,24 +276,34 @@ func (c *ClientConn) Run() {
 				err.Error(), 0,
 				"stack", string(buf))
 		}
-
 		c.Close()
 	}()
 
 	for {
-		data, err := c.readPacket()
+		// if this client connection have not set the route for specific ID
+		// TODO find route
 
-		if err != nil {
-			return
-		}
+		// now just do default route
+		err := c.DoStreamRoute()
 
-		if err := c.dispatch(data); err != nil {
-			c.proxy.counter.IncrErrLogTotal()
-			golog.Error("server", "Run",
-				err.Error(), c.connectionId,
-			)
-			c.writeError(err)
+		if err != nil {
+			golog.Error("ClientConn", "Run", "route btyes error", c.connectionId, err.Error())
 		}
+		//
+		// golog.Info("ClientConn", "Run", "route btyes", 0, n)
+
+		// data, err := c.readPacket()
+		// if err != nil {
+		// 	return
+		// }
+		//
+		// if err := c.dispatch(data); err != nil {
+		// 	c.proxy.counter.IncrErrLogTotal()
+		// 	golog.Error("server", "Run",
+		// 		err.Error(), c.connectionId,
+		// 	)
+		// 	c.writeError(err)
+		// }
 
 		if c.closed {
 			return
@@ -297,50 +315,48 @@ func (c *ClientConn) Run() {
 
 func (c *ClientConn) dispatch(data []byte) error {
 	c.proxy.counter.IncrClientQPS()
-	cmd := data[0]
-	data = data[1:]
-
-	switch cmd {
-	// case mysql.COM_QUIT:
-	// 	c.Close()
-	// 	return nil
-	case mysql.COM_QUERY:
-		golog.Warn("ClientConn", "dispatch", "query", 0, hack.String(data))
-		node := c.proxy.GetNode("node1")
-		co, err := c.getBackendConn(node, true)
-		res, err := co.Execute(hack.String(data))
-		if err != nil {
-			return err
-		}
-		c.writeResultset(res.Status, res.Resultset)
-	// 	return c.handleQuery(hack.String(data))
-	// case mysql.COM_PING:
-	// 	return c.writeOK(nil)
-	// case mysql.COM_INIT_DB:
-	// 	if err := c.useDB(hack.String(data)); err != nil {
+	//cmd := data[0]
+	//data = data[1:]
+	if len(hack.String(data)) == 0 {
+		golog.Warn("ClientConn", "dispatch", "skip empty query", 0)
+	}
+	// switch cmd {
+	// case mysql.COM_QUERY:
+	// 	golog.Info("ClientConn", "dispatch", "query", 0, hack.String(data))
+	// 	node := c.proxy.GetNode("node1")
+	// 	co, err := c.getBackendConn(node, true)
+	// 	res, err := co.Execute(hack.String(data))
+	// 	if err != nil {
 	// 		return err
-	// 	} else {
-	// 		return c.writeOK(nil)
 	// 	}
-	// case mysql.COM_FIELD_LIST:
-	// 	return c.handleFieldList(data)
-	// case mysql.COM_STMT_PREPARE:
-	// 	return c.handleStmtPrepare(hack.String(data))
-	// case mysql.COM_STMT_EXECUTE:
-	// 	return c.handleStmtExecute(data)
-	// case mysql.COM_STMT_CLOSE:
-	// 	return c.handleStmtClose(data)
-	// case mysql.COM_STMT_SEND_LONG_DATA:
-	// 	return c.handleStmtSendLongData(data)
-	// case mysql.COM_STMT_RESET:
-	// 	return c.handleStmtReset(data)
-	// case mysql.COM_SET_OPTION:
-	// 	return c.writeEOF(0)
-	default:
-		msg := fmt.Sprintf("command %d not supported now, data is %s", cmd, string(data))
-		golog.Error("ClientConn", "dispatch", msg, 0)
-		return mysql.NewError(mysql.ER_UNKNOWN_ERROR, msg)
+	// 	c.writeResultset(res.Status, res.Resultset)
+	// default:
+	// 	msg := fmt.Sprintf("command %d not supported now, data is %s", cmd, string(data))
+	// 	golog.Error("ClientConn", "dispatch", msg, 0)
+	// 	return mysql.NewError(mysql.ER_UNKNOWN_ERROR, msg)
+	// }
+
+	golog.Info("ClientConn", "dispatch", "query", 0, hack.String(data))
+	node := c.proxy.GetNode("node1")
+	backendConn, err := c.getBackendConn(node, true)
+	backendConn.UseDB(c.db)
+
+	//	res, err := co.Execute(hack.String(data))
+	err = backendConn.Write(data)
+	if err != nil {
+		return err
 	}
+	result, err := backendConn.Read()
+	if err != nil {
+		return err
+	}
+
+	c.writePacket(result)
+	// if res.Resultset != nil {
+	// 	c.writeResultset(res.Status, res.Resultset)
+	// } else {
+	// 	c.writeOK(res)
+	// }
 
 	return nil
 }
@@ -362,7 +378,7 @@ func (c *ClientConn) useDB(db string) error {
 	// if err = co.UseDB(db); err != nil {
 	// 	return err
 	// }
-	// c.db = db
+	c.db = db
 	return nil
 }