enable redis as store

Author: CMGS

cluster/calcium/calcium.go

@@ -16,6 +16,7 @@ import (
 	"github.com/projecteru2/core/source/gitlab"
 	"github.com/projecteru2/core/store"
 	"github.com/projecteru2/core/store/etcdv3"
+	"github.com/projecteru2/core/store/redis"
 	"github.com/projecteru2/core/types"
 )
 
@@ -32,10 +33,21 @@ type Calcium struct {
 // New returns a new cluster config
 func New(config types.Config, embeddedStorage bool) (*Calcium, error) {
 	logger := log.WithField("Calcium", "New").WithField("config", config)
+
 	// set store
-	store, err := etcdv3.New(config, embeddedStorage)
-	if err != nil {
-		return nil, logger.Err(errors.WithStack(err))
+	var store store.Store
+	var err error
+	switch config.Store {
+	case types.Etcd:
+		store, err = etcdv3.New(config, embeddedStorage)
+		if err != nil {
+			return nil, logger.Err(errors.WithStack(err))
+		}
+	case types.Redis:
+		store, err = redis.New(config, embeddedStorage)
+		if err != nil {
+			return nil, logger.Err(errors.WithStack(err))
+		}
 	}
 
 	// set scheduler

core.yaml.sample

@@ -18,6 +18,8 @@ grpc:
     service_discovery_interval: 5s  # WatchServiceStatus push interval
     service_heartbeat_interval: 5s  # RegisterService heartbeat
 
+store: etcd
+
 etcd:
     machines:
         - "http://127.0.0.1:2379"

go.sum

@@ -1,11 +1,19 @@
 package redis
 
 import (
+	"context"
 	"strings"
 
 	"github.com/projecteru2/core/strategy"
 )
 
+// extracts node name from key
+// /nodestatus/nodename -> nodename
+func extractNodename(s string) string {
+	ps := strings.Split(s, "/")
+	return ps[len(ps)-1]
+}
+
 func parseStatusKey(key string) (string, string, string, string) {
 	parts := strings.Split(key, "/")
 	l := len(parts)
@@ -19,3 +27,30 @@ func setCount(nodesCount map[string]int, strategyInfos []strategy.Info) {
 		}
 	}
 }
+
+// getByKeyPattern gets key-value pairs that key matches pattern
+func (r *Rediaron) getByKeyPattern(ctx context.Context, pattern string, limit int64) (map[string]string, error) {
+	var (
+		cursor uint64
+		result []string
+		err    error
+		count  int64
+		keys   = []string{}
+	)
+	for {
+		result, cursor, err = r.cli.Scan(ctx, cursor, pattern, 0).Result()
+		if err != nil {
+			return nil, err
+		}
+
+		keys = append(keys, result...)
+		count += int64(len(result))
+		if cursor == 0 || (limit > 0 && count >= limit) {
+			break
+		}
+	}
+	if limit > 0 && int64(len(keys)) >= limit {
+		keys = keys[:limit]
+	}
+	return r.GetMulti(ctx, keys)
+}

store/redis/helper.go

@@ -361,10 +361,3 @@ func (r *Rediaron) NodeStatusStream(ctx context.Context) chan *types.NodeStatus
 	}()
 	return ch
 }
-
-// extracts node name from key
-// /nodestatus/nodename -> nodename
-func extractNodename(s string) string {
-	ps := strings.Split(s, "/")
-	return ps[len(ps)-1]
-}

store/redis/node.go

@@ -65,41 +65,19 @@ type Rediaron struct {
 // New creates a new Rediaron instance from config
 // Only redis address and db is used
 // db is used to separate data, by default db 0 will be used
-func New(config types.Config) (*Rediaron, error) {
+func New(config types.Config, embeddedStorage bool) (*Rediaron, error) {
 	cli := redis.NewClient(&redis.Options{
 		Addr: config.Redis.Addr,
 		DB:   config.Redis.DB,
 	})
 
-	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
-	defer cancel()
-
-	_, err := cli.Ping(ctx).Result()
-	if err != nil {
-		return nil, err
-	}
-
 	return &Rediaron{
 		cli:    cli,
 		config: config,
 		db:     config.Redis.DB,
 	}, nil
 }
 
-// // keysWatchedWithRetry provides a method to retry a transaction
-// func (r *Rediaron) keysWatchedWithRetry(ctx context.Context, keys []string, retry int, txn func(*redis.Tx) error) error {
-// 	for i := 0; i < retry; i++ {
-// 		err := r.cli.Watch(ctx, txn, keys...)
-// 		if err == redis.TxFailedErr {
-// 			// value of watched keys changed, retry
-// 			continue
-// 		}
-// 		// no error or other errors, return
-// 		return err
-// 	}
-// 	return ErrMaxRetryExceeded
-// }
-
 // KNotifyMessage is received when using KNotify
 type KNotifyMessage struct {
 	Key    string
@@ -134,33 +112,6 @@ func (r *Rediaron) KNotify(ctx context.Context, pattern string) chan *KNotifyMes
 	return ch
 }
 
-// getByKeyPattern gets key-value pairs that key matches pattern
-func (r *Rediaron) getByKeyPattern(ctx context.Context, pattern string, limit int64) (map[string]string, error) {
-	var (
-		cursor uint64
-		result []string
-		err    error
-		count  int64
-		keys   = []string{}
-	)
-	for {
-		result, cursor, err = r.cli.Scan(ctx, cursor, pattern, 0).Result()
-		if err != nil {
-			return nil, err
-		}
-
-		keys = append(keys, result...)
-		count += int64(len(result))
-		if cursor == 0 || (limit > 0 && count >= limit) {
-			break
-		}
-	}
-	if limit > 0 && int64(len(keys)) >= limit {
-		keys = keys[:limit]
-	}
-	return r.GetMulti(ctx, keys)
-}
-
 // GetOne is a wrapper
 func (r *Rediaron) GetOne(ctx context.Context, key string) (string, error) {
 	return r.cli.Get(ctx, key).Result()

store/redis/rediaron.go

@@ -26,43 +26,6 @@ func (s *RediaronTestSuite) TearDownTest() {
 	s.rediaron.cli.FlushAll(context.Background())
 }
 
-// func (s *RediaronTestSuite) TestKeysWatchedWithRetry() {
-// 	ctx := context.Background()
-//
-// 	update := func(key string) error {
-// 		txf := func(tx *redis.Tx) error {
-// 			n, err := tx.Get(ctx, key).Int()
-// 			if err != nil && err != redis.Nil {
-// 				return err
-// 			}
-//
-// 			n++
-//
-// 			_, err = tx.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
-// 				pipe.Set(ctx, key, n, 0)
-// 				return nil
-// 			})
-// 			return err
-// 		}
-// 		return s.rediaron.keysWatchedWithRetry(ctx, []string{"test"}, 200, txf)
-// 	}
-//
-// 	var wg sync.WaitGroup
-// 	for i := 0; i < 100; i++ {
-// 		wg.Add(1)
-// 		go func() {
-// 			defer wg.Done()
-//
-// 			s.NoError(update("test"))
-// 		}()
-// 	}
-// 	wg.Wait()
-//
-// 	n, err := s.rediaron.cli.Get(context.Background(), "test").Int()
-// 	s.NoError(err)
-// 	s.Equal(100, n)
-// }
-
 func (s *RediaronTestSuite) TestKeyNotify() {
 	ctx, cancel := context.WithCancel(context.Background())
 	ch := s.rediaron.KNotify(ctx, "a*")

store/redis/rediaron_test.go

@@ -4,6 +4,13 @@ import (
 	"time"
 )
 
+const (
+	// Etcd .
+	Etcd = "etcd"
+	// Redis .
+	Redis = "redis"
+)
+
 // Config holds eru-core config
 type Config struct {
 	LogLevel       string        `yaml:"log_level" required:"true" default:"INFO"`
@@ -14,8 +21,10 @@ type Config struct {
 	Profile        string        `yaml:"profile"`                                       // profile ip:port
 	CertPath       string        `yaml:"cert_path"`                                     // docker cert files path
 	MaxConcurrency int64         `yaml:"max_concurrency" default:"20"`                  // concurrently call single runtime in the same time
-	Auth           AuthConfig    `yaml:"auth"`                                          // grpc auth
-	GRPCConfig     GRPCConfig    `yaml:"grpc"`                                          // grpc config
+	Store          string        `yaml:"store" default:"etcd"`                          // store type
+
+	Auth       AuthConfig `yaml:"auth"` // grpc auth
+	GRPCConfig GRPCConfig `yaml:"grpc"` // grpc config
 
 	WALFile        string        `yaml:"wal_file" required:"true" default:"core.wal"`   // WAL file path
 	WALOpenTimeout time.Duration `yaml:"wal_open_timeout" required:"true" default:"8s"` // timeout for opening a WAL file