Merge pull request #32 from felixhao/cluster

Version 1.4.0

.travis.yml

@@ -8,10 +8,13 @@ go_import_path: overlord
 services:
   - memcached
   - redis
+  - docker
 
 install:
   - go get -u github.com/alecthomas/gometalinter
   - gometalinter --install
+  - docker pull grokzen/redis-cluster:4.0.9
+  - docker run -e "IP=0.0.0.0" -d -p 7000-7007:7000-7007 grokzen/redis-cluster:4.0.9
 
 script:
   - gometalinter --vendor --disable-all --enable=gosimple --enable=golint --enable=vet --enable=ineffassign --enable=unconvert \
@@ -26,3 +29,7 @@ script:
 
 after_success:
   - bash <(curl -s https://codecov.io/bash)
+
+after_script:
+- docker stop redis-cluster
+- docker rm redis-clust

CHANGELOG.md

@@ -1,5 +1,8 @@
 # Overlord
 
+## Version 1.4.0
+1. add redis cluster support.
+
 ## Version 1.3.2
 1. change round-chan to race-chan. 
 

README.md

@@ -52,7 +52,7 @@ Congratulations! You've just run the overlord proxy.
 ## Features
 
 - [x] support memcache protocol
-- [x] support redis protocol
+- [x] support redis&cluster protocol
 - [x] connection pool for reduce number to backend caching servers
 - [x] keepalive & failover
 - [x] hash tag: specify the part of the key used for hashing

cmd/proxy/main.go

@@ -17,7 +17,7 @@ import (
 
 const (
 	// VERSION version
-	VERSION = "1.1.0"
+	VERSION = "1.4.0"
 )
 
 var (

cmd/proxy/proxy-cluster-example.toml

@@ -7,7 +7,7 @@ hash_method = "fnv1a_64"
 hash_distribution = "ketama"
 # A two character string that specifies the part of the key used for hashing. Eg "{}".
 hash_tag = ""
-# cache type: memcache | memcache_binary |redis 
+# cache type: memcache | memcache_binary | redis | redis_cluster
 cache_type = "memcache"
 # proxy listen proto: tcp | unix
 listen_proto = "tcp"
@@ -26,10 +26,10 @@ node_connections = 2
 # The number of consecutive failures on a server that would lead to it being temporarily ejected when auto_eject is set to true. Defaults to 3.
 ping_fail_limit = 3
 # A boolean value that controls if server should be ejected temporarily when it fails consecutively ping_fail_limit times.
-ping_auto_eject = false
+ping_auto_eject = true
 # A list of server address, port and weight (name:port:weight or ip:port:weight) for this server pool. Also you can use alias name like: ip:port:weight alias.
 servers = [
-    "127.0.0.1:11211:1",
+    "127.0.0.1:11211:1 mc1",
 ]
 
 [[clusters]]
@@ -41,7 +41,7 @@ hash_method = "fnv1a_64"
 hash_distribution = "ketama"
 # A two character string that specifies the part of the key used for hashing. Eg "{}".
 hash_tag = ""
-# cache type: memcache | memcache_binary |redis 
+# cache type: memcache | memcache_binary | redis | redis_cluster
 cache_type = "redis"
 # proxy listen proto: tcp | unix
 listen_proto = "tcp"
@@ -63,5 +63,40 @@ ping_fail_limit = 3
 ping_auto_eject = false
 # A list of server address, port and weight (name:port:weight or ip:port:weight) for this server pool. Also you can use alias name like: ip:port:weight alias.
 servers = [
-    "127.0.0.1:6379:1",
-]
+    "127.0.0.1:6379:1 redis1",
+]
+
+[[clusters]]
+# This be used to specify the name of cache cluster.
+name = "test-redis-cluster"
+# The name of the hash function. Possible values are: sha1.
+hash_method = "fnv1a_64"
+# The key distribution mode. Possible values are: ketama.
+hash_distribution = "ketama"
+# A two character string that specifies the part of the key used for hashing. Eg "{}".
+hash_tag = "{}"
+# cache type: memcache | memcache_binary | redis | redis_cluster
+cache_type = "redis_cluster"
+# proxy listen proto: tcp | unix
+listen_proto = "tcp"
+# proxy listen addr: tcp addr | unix sock path
+listen_addr = "0.0.0.0:27000"
+# Authenticate to the Redis server on connect.
+redis_auth = ""
+# The dial timeout value in msec that we wait for to establish a connection to the server. By default, we wait indefinitely.
+dial_timeout = 1000
+# The read timeout value in msec that we wait for to receive a response from a server. By default, we wait indefinitely.
+read_timeout = 1000
+# The write timeout value in msec that we wait for to write a response to a server. By default, we wait indefinitely.
+write_timeout = 1000
+# The number of connections that can be opened to each server. By default, we open at most 1 server connection.
+node_connections = 2
+# The number of consecutive failures on a server that would lead to it being temporarily ejected when auto_eject is set to true. Defaults to 3.
+ping_fail_limit = 3
+# A boolean value that controls if server should be ejected temporarily when it fails consecutively ping_fail_limit times.
+ping_auto_eject = false
+# A list of server address, port (name:port or ip:port) for this server pool when cache type is redis_cluster.
+servers = [
+    "127.0.0.1:7000",
+    "127.0.0.1:7001",
+]

lib/backoff/backoff.go

@@ -7,8 +7,8 @@ import (
 
 // defaultConfig uses values specified for backoff in common.
 var defaultConfig = Config{
-	MaxDelay:  20 * time.Second,
-	BaseDelay: 1 * time.Second,
+	MaxDelay:  2 * time.Second,
+	BaseDelay: 500 * time.Millisecond,
 	Factor:    1.6,
 	Jitter:    0.2,
 }

lib/hashkit/crc16.go

@@ -0,0 +1,48 @@
+package hashkit
+
+// CRC16 implementation according to CCITT standards.
+// Copyright 2001-2010 Georges Menie (www.menie.org)
+// Copyright 2013 The Go Authors. All rights reserved.
+// http://redis.io/topics/cluster-spec#appendix-a-crc16-reference-implementation-in-ansi-c
+var crc16tab = [256]uint16{
+	0x0000, 0x1021, 0x2042, 0x3063, 0x4084, 0x50a5, 0x60c6, 0x70e7,
+	0x8108, 0x9129, 0xa14a, 0xb16b, 0xc18c, 0xd1ad, 0xe1ce, 0xf1ef,
+	0x1231, 0x0210, 0x3273, 0x2252, 0x52b5, 0x4294, 0x72f7, 0x62d6,
+	0x9339, 0x8318, 0xb37b, 0xa35a, 0xd3bd, 0xc39c, 0xf3ff, 0xe3de,
+	0x2462, 0x3443, 0x0420, 0x1401, 0x64e6, 0x74c7, 0x44a4, 0x5485,
+	0xa56a, 0xb54b, 0x8528, 0x9509, 0xe5ee, 0xf5cf, 0xc5ac, 0xd58d,
+	0x3653, 0x2672, 0x1611, 0x0630, 0x76d7, 0x66f6, 0x5695, 0x46b4,
+	0xb75b, 0xa77a, 0x9719, 0x8738, 0xf7df, 0xe7fe, 0xd79d, 0xc7bc,
+	0x48c4, 0x58e5, 0x6886, 0x78a7, 0x0840, 0x1861, 0x2802, 0x3823,
+	0xc9cc, 0xd9ed, 0xe98e, 0xf9af, 0x8948, 0x9969, 0xa90a, 0xb92b,
+	0x5af5, 0x4ad4, 0x7ab7, 0x6a96, 0x1a71, 0x0a50, 0x3a33, 0x2a12,
+	0xdbfd, 0xcbdc, 0xfbbf, 0xeb9e, 0x9b79, 0x8b58, 0xbb3b, 0xab1a,
+	0x6ca6, 0x7c87, 0x4ce4, 0x5cc5, 0x2c22, 0x3c03, 0x0c60, 0x1c41,
+	0xedae, 0xfd8f, 0xcdec, 0xddcd, 0xad2a, 0xbd0b, 0x8d68, 0x9d49,
+	0x7e97, 0x6eb6, 0x5ed5, 0x4ef4, 0x3e13, 0x2e32, 0x1e51, 0x0e70,
+	0xff9f, 0xefbe, 0xdfdd, 0xcffc, 0xbf1b, 0xaf3a, 0x9f59, 0x8f78,
+	0x9188, 0x81a9, 0xb1ca, 0xa1eb, 0xd10c, 0xc12d, 0xf14e, 0xe16f,
+	0x1080, 0x00a1, 0x30c2, 0x20e3, 0x5004, 0x4025, 0x7046, 0x6067,
+	0x83b9, 0x9398, 0xa3fb, 0xb3da, 0xc33d, 0xd31c, 0xe37f, 0xf35e,
+	0x02b1, 0x1290, 0x22f3, 0x32d2, 0x4235, 0x5214, 0x6277, 0x7256,
+	0xb5ea, 0xa5cb, 0x95a8, 0x8589, 0xf56e, 0xe54f, 0xd52c, 0xc50d,
+	0x34e2, 0x24c3, 0x14a0, 0x0481, 0x7466, 0x6447, 0x5424, 0x4405,
+	0xa7db, 0xb7fa, 0x8799, 0x97b8, 0xe75f, 0xf77e, 0xc71d, 0xd73c,
+	0x26d3, 0x36f2, 0x0691, 0x16b0, 0x6657, 0x7676, 0x4615, 0x5634,
+	0xd94c, 0xc96d, 0xf90e, 0xe92f, 0x99c8, 0x89e9, 0xb98a, 0xa9ab,
+	0x5844, 0x4865, 0x7806, 0x6827, 0x18c0, 0x08e1, 0x3882, 0x28a3,
+	0xcb7d, 0xdb5c, 0xeb3f, 0xfb1e, 0x8bf9, 0x9bd8, 0xabbb, 0xbb9a,
+	0x4a75, 0x5a54, 0x6a37, 0x7a16, 0x0af1, 0x1ad0, 0x2ab3, 0x3a92,
+	0xfd2e, 0xed0f, 0xdd6c, 0xcd4d, 0xbdaa, 0xad8b, 0x9de8, 0x8dc9,
+	0x7c26, 0x6c07, 0x5c64, 0x4c45, 0x3ca2, 0x2c83, 0x1ce0, 0x0cc1,
+	0xef1f, 0xff3e, 0xcf5d, 0xdf7c, 0xaf9b, 0xbfba, 0x8fd9, 0x9ff8,
+	0x6e17, 0x7e36, 0x4e55, 0x5e74, 0x2e93, 0x3eb2, 0x0ed1, 0x1ef0,
+}
+
+// Crc16 caculate key crc to divide slot of redis
+func Crc16(key []byte) (crc uint16) {
+	for i := 0; i < len(key); i++ {
+		crc = (crc << 8) ^ crc16tab[(byte(crc>>8)^key[i])&0x00ff]
+	}
+	return
+}

lib/hashkit/crc16_test.go

@@ -0,0 +1,12 @@
+package hashkit
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestCrcCheckOk(t *testing.T) {
+	assert.Equal(t, uint16(0x31C3), Crc16([]byte("123456789")))
+	assert.Equal(t, uint16(21847), Crc16([]byte{83, 153, 134, 118, 229, 214, 244, 75, 140, 37, 215, 215}))
+}

proto/batch.go

@@ -156,8 +156,8 @@ func (m *MsgBatch) Done(cluster, addr string) {
 	}
 }
 
-// BatchDoneWithError will set done with error and report prom ErrIncr.
-func (m *MsgBatch) BatchDoneWithError(cluster, addr string, err error) {
+// DoneWithError will set done with error and report prom ErrIncr.
+func (m *MsgBatch) DoneWithError(cluster, addr string, err error) {
 	for _, msg := range m.Msgs() {
 		msg.WithError(err)
 		if log.V(1) {

proto/memcache/binary/node_conn.go

@@ -15,8 +15,8 @@ import (
 )
 
 const (
-	handlerOpening = int32(0)
-	handlerClosed  = int32(1)
+	opened = int32(0)
+	closed = int32(1)
 )
 
 type nodeConn struct {
@@ -27,7 +27,7 @@ type nodeConn struct {
 	bw   *bufio.Writer
 	br   *bufio.Reader
 
-	closed int32
+	state int32
 }
 
 // NewNodeConn returns node conn.
@@ -44,6 +44,10 @@ func NewNodeConn(cluster, addr string, dialTimeout, readTimeout, writeTimeout ti
 }
 
 func (n *nodeConn) WriteBatch(mb *proto.MsgBatch) (err error) {
+	if n.Closed() {
+		err = errors.Wrap(ErrClosed, "MC Writer write")
+		return
+	}
 	var (
 		m   *proto.Message
 		idx int
@@ -177,12 +181,12 @@ func (n *nodeConn) fillMCRequest(mcr *MCRequest, data []byte) (size int, err err
 }
 
 func (n *nodeConn) Close() error {
-	if atomic.CompareAndSwapInt32(&n.closed, handlerOpening, handlerClosed) {
+	if atomic.CompareAndSwapInt32(&n.state, opened, closed) {
 		return n.conn.Close()
 	}
 	return nil
 }
 
 func (n *nodeConn) Closed() bool {
-	return atomic.LoadInt32(&n.closed) == handlerClosed
+	return atomic.LoadInt32(&n.state) == closed
 }

proto/memcache/binary/node_conn_test.go

@@ -119,6 +119,7 @@ func TestNodeConnWriteClosed(t *testing.T) {
 	assert.Error(t, err)
 	_causeEqual(t, ErrClosed, err)
 	assert.NoError(t, nc.Close())
+	_causeEqual(t, ErrClosed, nc.WriteBatch(nil))
 }
 
 type mockReq struct {

proto/memcache/binary/pinger.go

@@ -41,10 +41,11 @@ var (
 )
 
 type mcPinger struct {
-	conn   *libnet.Conn
-	bw     *bufio.Writer
-	br     *bufio.Reader
-	closed int32
+	conn *libnet.Conn
+	bw   *bufio.Writer
+	br   *bufio.Reader
+
+	state int32
 }
 
 // NewPinger new pinger.
@@ -57,7 +58,7 @@ func NewPinger(nc *libnet.Conn) proto.Pinger {
 }
 
 func (m *mcPinger) Ping() (err error) {
-	if atomic.LoadInt32(&m.closed) == handlerClosed {
+	if atomic.LoadInt32(&m.state) == closed {
 		err = ErrPingerPong
 		return
 	}
@@ -79,7 +80,7 @@ func (m *mcPinger) Ping() (err error) {
 }
 
 func (m *mcPinger) Close() error {
-	if atomic.CompareAndSwapInt32(&m.closed, handlerOpening, handlerClosed) {
+	if atomic.CompareAndSwapInt32(&m.state, opened, closed) {
 		return m.conn.Close()
 	}
 	return nil

proto/memcache/node_conn.go

@@ -2,19 +2,21 @@ package memcache
 
 import (
 	"bytes"
+	"strconv"
 	"sync/atomic"
 	"time"
 
 	"overlord/lib/bufio"
+	"overlord/lib/log"
 	libnet "overlord/lib/net"
 	"overlord/proto"
 
 	"github.com/pkg/errors"
 )
 
 const (
-	handlerOpening = int32(0)
-	handlerClosed  = int32(1)
+	opened = int32(0)
+	closed = int32(1)
 )
 
 type nodeConn struct {
@@ -25,7 +27,7 @@ type nodeConn struct {
 	bw   *bufio.Writer
 	br   *bufio.Reader
 
-	closed int32
+	state int32
 }
 
 // NewNodeConn returns node conn.
@@ -42,6 +44,10 @@ func NewNodeConn(cluster, addr string, dialTimeout, readTimeout, writeTimeout ti
 }
 
 func (n *nodeConn) WriteBatch(mb *proto.MsgBatch) (err error) {
+	if n.Closed() {
+		err = errors.Wrap(ErrClosed, "MC Writer conn closed")
+		return
+	}
 	var (
 		m   *proto.Message
 		idx int
@@ -161,8 +167,15 @@ func (n *nodeConn) fillMCRequest(mcr *MCRequest, data []byte) (size int, err err
 	if bytes.Equal(bs, endBytes) {
 		return
 	}
+
+	if bytes.Equal(bs, errorBytes) {
+		log.Errorf("parse bad with request %v mc response with data string %s and bytes %v and bs is %v", *mcr, strconv.Quote(string(data)), data, bs)
+		return
+	}
+
 	length, err := findLength(bs, mcr.rTp == RequestTypeGets || mcr.rTp == RequestTypeGats)
 	if err != nil {
+		log.Errorf("parse bad with request %v mc response with data string %s and bytes %v and bs is %v", *mcr, strconv.Quote(string(data)), data, bs)
 		err = errors.Wrap(err, "MC Handler while parse length")
 		return
 	}
@@ -176,13 +189,13 @@ func (n *nodeConn) fillMCRequest(mcr *MCRequest, data []byte) (size int, err err
 }
 
 func (n *nodeConn) Close() error {
-	if atomic.CompareAndSwapInt32(&n.closed, handlerOpening, handlerClosed) {
+	if atomic.CompareAndSwapInt32(&n.state, opened, closed) {
 		err := n.conn.Close()
 		return err
 	}
 	return nil
 }
 
 func (n *nodeConn) Closed() bool {
-	return atomic.LoadInt32(&n.closed) == handlerClosed
+	return atomic.LoadInt32(&n.state) == closed
 }

proto/memcache/node_conn_test.go

@@ -152,6 +152,7 @@ func TestNodeConnWriteClosed(t *testing.T) {
 	assert.Error(t, err)
 	_causeEqual(t, ErrClosed, err)
 	assert.NoError(t, nc.Close())
+	_causeEqual(t, ErrClosed, nc.WriteBatch(nil))
 }
 
 type mockReq struct {