Add Feature: Support Redis Cluster Mode

.gitignore

@@ -4,4 +4,5 @@ out/
 *.swo
 *.tar.gz
 docs/_site
+*idea*
 .idea

config/alarm.json

@@ -5,6 +5,7 @@
         "listen": "0.0.0.0:9912"
     },
     "redis": {
+        "cluster": false,
         "addr": "%%REDIS%%",
         "maxIdle": 5,
         "highQueues": [

config/judge.json

@@ -20,6 +20,7 @@
         "minInterval": 300,
         "queuePattern": "event:p%v",
         "redis": {
+            "cluster": false,
             "dsn": "%%REDIS%%",
             "maxIdle": 5,
             "connTimeout": 5000,

modules/alarm/cron/combiner.go

@@ -18,7 +18,6 @@ import (
 	"encoding/json"
 	"fmt"
 	log "github.com/Sirupsen/logrus"
-	"github.com/garyburd/redigo/redis"
 	"github.com/open-falcon/falcon-plus/modules/alarm/api"
 	"github.com/open-falcon/falcon-plus/modules/alarm/g"
 	"github.com/open-falcon/falcon-plus/modules/alarm/redi"
@@ -197,15 +196,10 @@ func popAllSmsDto() []*SmsDto {
 	ret := []*SmsDto{}
 	queue := g.Config().Redis.UserSmsQueue
 
-	rc := g.RedisConnPool.Get()
-	defer rc.Close()
-
 	for {
-		reply, err := redis.String(rc.Do("RPOP", queue))
+		reply, err := g.RedisString(g.RedisDo("RPOP", queue))
 		if err != nil {
-			if err != redis.ErrNil {
-				log.Error("get SmsDto fail", err)
-			}
+			log.Error("rpop get SmsDto fail", err)
 			break
 		}
 
@@ -230,15 +224,10 @@ func popAllMailDto() []*MailDto {
 	ret := []*MailDto{}
 	queue := g.Config().Redis.UserMailQueue
 
-	rc := g.RedisConnPool.Get()
-	defer rc.Close()
-
 	for {
-		reply, err := redis.String(rc.Do("RPOP", queue))
+		reply, err := g.RedisString(g.RedisDo("RPOP", queue))
 		if err != nil {
-			if err != redis.ErrNil {
-				log.Error("get MailDto fail", err)
-			}
+			log.Error("rpop get MailDto fail", err)
 			break
 		}
 
@@ -263,15 +252,10 @@ func popAllImDto() []*ImDto {
 	ret := []*ImDto{}
 	queue := g.Config().Redis.UserIMQueue
 
-	rc := g.RedisConnPool.Get()
-	defer rc.Close()
-
 	for {
-		reply, err := redis.String(rc.Do("RPOP", queue))
+		reply, err := g.RedisString(g.RedisDo("RPOP", queue))
 		if err != nil {
-			if err != redis.ErrNil {
-				log.Error("get ImDto fail", err)
-			}
+			log.Error("get ImDto fail", err)
 			break
 		}
 

modules/alarm/cron/event_consumer.go

@@ -93,9 +93,6 @@ func ParseUserSms(event *cmodel.Event, action *api.Action) {
 
 	queue := g.Config().Redis.UserSmsQueue
 
-	rc := g.RedisConnPool.Get()
-	defer rc.Close()
-
 	for _, user := range userMap {
 		dto := SmsDto{
 			Priority: priority,
@@ -110,7 +107,7 @@ func ParseUserSms(event *cmodel.Event, action *api.Action) {
 			continue
 		}
 
-		_, err = rc.Do("LPUSH", queue, string(bs))
+		_, err = g.RedisDo("LPUSH", queue, string(bs))
 		if err != nil {
 			log.Error("LPUSH redis", queue, "fail:", err, "dto:", string(bs))
 		}
@@ -128,9 +125,6 @@ func ParseUserMail(event *cmodel.Event, action *api.Action) {
 
 	queue := g.Config().Redis.UserMailQueue
 
-	rc := g.RedisConnPool.Get()
-	defer rc.Close()
-
 	for _, user := range userMap {
 		dto := MailDto{
 			Priority: priority,
@@ -146,7 +140,7 @@ func ParseUserMail(event *cmodel.Event, action *api.Action) {
 			continue
 		}
 
-		_, err = rc.Do("LPUSH", queue, string(bs))
+		_, err = g.RedisDo("LPUSH", queue, string(bs))
 		if err != nil {
 			log.Error("LPUSH redis", queue, "fail:", err, "dto:", string(bs))
 		}
@@ -163,9 +157,6 @@ func ParseUserIm(event *cmodel.Event, action *api.Action) {
 
 	queue := g.Config().Redis.UserIMQueue
 
-	rc := g.RedisConnPool.Get()
-	defer rc.Close()
-
 	for _, user := range userMap {
 		dto := ImDto{
 			Priority: priority,
@@ -180,7 +171,7 @@ func ParseUserIm(event *cmodel.Event, action *api.Action) {
 			continue
 		}
 
-		_, err = rc.Do("LPUSH", queue, string(bs))
+		_, err = g.RedisDo("LPUSH", queue, string(bs))
 		if err != nil {
 			log.Error("LPUSH redis", queue, "fail:", err, "dto:", string(bs))
 		}

modules/alarm/cron/event_reader.go

@@ -19,76 +19,40 @@ import (
 	"time"
 
 	log "github.com/Sirupsen/logrus"
-	"github.com/garyburd/redigo/redis"
 	cmodel "github.com/open-falcon/falcon-plus/common/model"
 	"github.com/open-falcon/falcon-plus/modules/alarm/g"
 	eventmodel "github.com/open-falcon/falcon-plus/modules/alarm/model/event"
 )
 
-func ReadHighEvent() {
-	queues := g.Config().Redis.HighQueues
-	if len(queues) == 0 {
-		return
-	}
+func SinglePopEvent(high bool, params ...interface{}) {
+
+	log.Infof("singlePopEvent bool:%t, %v", high, params)
 
 	for {
-		event, err := popEvent(queues)
+		reply, err := g.RedisStrings(g.RedisDo("BRPOP", params...))
 		if err != nil {
-			time.Sleep(time.Second)
-			continue
+			log.Errorf("brpop alarm event from redis fail: %v, %v", err, params)
+			return
 		}
-		consume(event, true)
-	}
-}
 
-func ReadLowEvent() {
-	queues := g.Config().Redis.LowQueues
-	if len(queues) == 0 {
-		return
-	}
-
-	for {
-		event, err := popEvent(queues)
+		var event cmodel.Event
+		err = json.Unmarshal([]byte(reply[1]), &event)
 		if err != nil {
-			time.Sleep(time.Second)
-			continue
+			log.Errorf("parse alarojum event fail: %v, %+v", err, reply)
+			return
 		}
-		consume(event, false)
-	}
-}
-
-func popEvent(queues []string) (*cmodel.Event, error) {
-
-	count := len(queues)
 
-	params := make([]interface{}, count+1)
-	for i := 0; i < count; i++ {
-		params[i] = queues[i]
-	}
-	// set timeout 0
-	params[count] = 0
+		log.Debugf("pop event: %s", event.String())
 
-	rc := g.RedisConnPool.Get()
-	defer rc.Close()
+		//insert event into database
+		eventmodel.InsertEvent(&event)
+		// events no longer saved in memory
 
-	reply, err := redis.Strings(rc.Do("BRPOP", params...))
-	if err != nil {
-		log.Errorf("get alarm event from redis fail: %v", err)
-		return nil, err
-	}
+		if err != nil {
+			time.Sleep(time.Second)
+			continue
+		}
 
-	var event cmodel.Event
-	err = json.Unmarshal([]byte(reply[1]), &event)
-	if err != nil {
-		log.Errorf("parse alarm event fail: %v", err)
-		return nil, err
+		consume(&event, high)
 	}
-
-	log.Debugf("pop event: %s", event.String())
-
-	//insert event into database
-	eventmodel.InsertEvent(&event)
-	// events no longer saved in memory
-
-	return &event, nil
 }

modules/alarm/g/cfg.go

@@ -28,6 +28,7 @@ type HttpConfig struct {
 }
 
 type RedisConfig struct {
+	Cluster       bool     `json:"cluster"`
 	Addr          string   `json:"addr"`
 	MaxIdle       int      `json:"maxIdle"`
 	HighQueues    []string `json:"highQueues"`

modules/alarm/g/redis.go

@@ -15,34 +15,136 @@
 package g
 
 import (
-	"github.com/garyburd/redigo/redis"
+	redisCluster "github.com/chasex/redis-go-cluster"
+	redisgo "github.com/garyburd/redigo/redis"
 	"log"
 	"time"
 )
 
-var RedisConnPool *redis.Pool
+var RedisConnPool *redisgo.Pool
+var RedisCluster *redisCluster.Cluster
 
 func InitRedisConnPool() {
 	redisConfig := Config().Redis
-
-	RedisConnPool = &redis.Pool{
-		MaxIdle:     redisConfig.MaxIdle,
-		IdleTimeout: 240 * time.Second,
-		Dial: func() (redis.Conn, error) {
-			c, err := redis.Dial("tcp", redisConfig.Addr)
-			if err != nil {
-				return nil, err
-			}
-			return c, err
-		},
-		TestOnBorrow: PingRedis,
+	if redisConfig.Cluster {
+		var err error
+		RedisCluster, err = redisCluster.NewCluster(
+			&redisCluster.Options{
+				StartNodes:   []string{redisConfig.Addr},
+				ConnTimeout:  500 * time.Millisecond,
+				ReadTimeout:  500 * time.Millisecond,
+				WriteTimeout: 500 * time.Millisecond,
+				KeepAlive:    16,
+				AliveTime:    60 * time.Second,
+			})
+		if err != nil {
+			log.Println("[ERROR] redis cluster init fail", err)
+		}
+	} else {
+		RedisConnPool = &redisgo.Pool{
+			MaxIdle:     redisConfig.MaxIdle,
+			IdleTimeout: 240 * time.Second,
+			Dial: func() (redisgo.Conn, error) {
+				c, err := redisgo.Dial("tcp", redisConfig.Addr)
+				if err != nil {
+					log.Println("[ERROR] redis pool init fail", err)
+					return nil, err
+				}
+				return c, err
+			},
+			TestOnBorrow: PingRedis,
+		}
 	}
 }
 
-func PingRedis(c redis.Conn, t time.Time) error {
+func PingRedis(c redisgo.Conn, t time.Time) error {
 	_, err := c.Do("ping")
 	if err != nil {
 		log.Println("[ERROR] ping redis fail", err)
 	}
 	return err
 }
+
+func RedisString(reply interface{}, err error) (string, error) {
+	redisConfig := Config().Redis
+	if redisConfig.Cluster {
+		reply, err := redisCluster.String(reply, err)
+		if err == redisCluster.ErrNil {
+			err = nil
+		}
+		if err != nil {
+			log.Println("[ERROR] RedisString", err, reply)
+		}
+		return reply, err
+	} else {
+		rc := RedisConnPool.Get()
+		defer rc.Close()
+		reply, err := redisgo.String(reply, err)
+		if err == redisgo.ErrNil {
+			err = nil
+		}
+		if err != nil {
+			log.Println("[ERROR] RedisString", err, reply)
+		}
+		return reply, err
+	}
+}
+
+func RedisStrings(reply interface{}, err error) ([]string, error) {
+	redisConfig := Config().Redis
+	if redisConfig.Cluster {
+		reply, err := redisCluster.Strings(reply, err)
+		if err == redisCluster.ErrNil {
+			err = nil
+		}
+		if err != nil {
+			log.Println("[ERROR] RedisString", err, reply)
+		}
+		return reply, err
+	} else {
+		rc := RedisConnPool.Get()
+		defer rc.Close()
+		reply, err := redisgo.Strings(reply, err)
+		if err == redisgo.ErrNil {
+			err = nil
+		}
+		if err != nil {
+			log.Println("[ERROR] RedisString", err, reply)
+		}
+		return reply, err
+	}
+}
+
+func RedisDo(commandName string, args ...interface{}) (interface{}, error) {
+	redisConfig := Config().Redis
+	if redisConfig.Cluster {
+		reply, err := RedisCluster.Do(commandName, args...)
+		if err == redisCluster.ErrNil {
+			err = nil
+		}
+		if err != nil {
+			log.Println("[ERROR] RedisDo", err, commandName, args)
+		}
+		return reply, err
+	} else {
+		rc := RedisConnPool.Get()
+		defer rc.Close()
+		reply, err := rc.Do(commandName, args...)
+		if err == redisgo.ErrNil {
+			err = nil
+		}
+		if err != nil {
+			log.Println("[ERROR] RedisDo", err, commandName, args)
+		}
+		return reply, err
+	}
+}
+
+func RedisClose() {
+	redisConfig := Config().Redis
+	if redisConfig.Cluster {
+		RedisCluster.Close()
+	} else {
+		RedisConnPool.Close()
+	}
+}