feat: bcs-user-manager 支持 redis 哨兵和集群模式

bcs-common/common/redisclient/cluster.go

@@ -0,0 +1,62 @@
+package redisclient
+
+import (
+	"context"
+	"time"
+
+	"github.com/go-redis/redis/v8"
+)
+
+type ClusterClient struct {
+	cli *redis.ClusterClient
+}
+
+func NewClusterClient(config Config) (*ClusterClient, error) {
+	cli := redis.NewClusterClient(&redis.ClusterOptions{
+		Addrs:        config.Addrs,
+		Password:     config.Password,
+		DialTimeout:  config.DialTimeout * time.Second,
+		ReadTimeout:  config.ReadTimeout * time.Second,
+		WriteTimeout: config.WriteTimeout * time.Second,
+		PoolSize:     config.PoolSize,
+		MinIdleConns: config.MinIdleConns,
+		IdleTimeout:  config.IdleTimeout * time.Second,
+	})
+	return &ClusterClient{cli: cli}, nil
+}
+
+func (c *ClusterClient) GetCli() redis.UniversalClient {
+	return c.cli
+}
+
+func (c *ClusterClient) Ping(ctx context.Context) (string, error) {
+	return c.cli.Ping(ctx).Result()
+}
+
+func (c *ClusterClient) Get(ctx context.Context, key string) (string, error) {
+	return c.cli.Get(ctx, key).Result()
+}
+
+func (c *ClusterClient) Exists(ctx context.Context, key ...string) (int64, error) {
+	return c.cli.Exists(ctx, key...).Result()
+}
+
+func (c *ClusterClient) Set(ctx context.Context, key string, value interface{}, duration time.Duration) (string, error) {
+	return c.cli.Set(ctx, key, value, duration).Result()
+}
+
+func (c *ClusterClient) SetNX(ctx context.Context, key string, value interface{}, expiration time.Duration) (bool, error) {
+	return c.cli.SetNX(ctx, key, value, expiration).Result()
+}
+
+func (c *ClusterClient) SetEX(ctx context.Context, key string, value interface{}, expiration time.Duration) (string, error) {
+	return c.cli.SetEX(ctx, key, value, expiration).Result()
+}
+
+func (c *ClusterClient) Del(ctx context.Context, key string) (int64, error) {
+	return c.cli.Del(ctx, key).Result()
+}
+
+func (c *ClusterClient) Expire(ctx context.Context, key string, duration time.Duration) (bool, error) {
+	return c.cli.Expire(ctx, key, duration).Result()
+}

bcs-common/common/redisclient/cluster_test.go

@@ -0,0 +1,58 @@
+package redisclient
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+// setupClusterClient function for initializing Redis ClusterClient
+func setupClusterClient(t *testing.T) *ClusterClient {
+	config := Config{
+		Mode:  ClusterMode,
+		Addrs: []string{"127.0.0.1:7021", "127.0.0.1:7022", "127.0.0.1:7023"},
+	}
+	client, err := NewClusterClient(config)
+	assert.NoError(t, err)
+	assert.NotNil(t, client)
+	return client
+}
+
+// TestClusterPing tests ClusterClient connectivity
+func TestClusterPing(t *testing.T) {
+	client := setupClusterClient(t)
+	result, err := client.GetCli().Ping(context.TODO()).Result()
+	assert.NoError(t, err)
+	assert.Equal(t, "PONG", result)
+}
+
+// TestClusterClient tests ClusterClient basic functionality
+func TestClusterClient(t *testing.T) {
+	client := setupClusterClient(t)
+	ctx := context.Background()
+
+	// Test Set operation
+	_, err := client.Set(ctx, "key1", "value1", 10*time.Second)
+	assert.NoError(t, err)
+
+	// Test Get operation
+	val, err := client.Get(ctx, "key1")
+	assert.NoError(t, err)
+	assert.Equal(t, "value1", val)
+
+	// Test Exists operation
+	exists, err := client.Exists(ctx, "key1")
+	assert.NoError(t, err)
+	assert.Equal(t, int64(1), exists)
+
+	// Test Del operation
+	_, err = client.Del(ctx, "key1")
+	assert.NoError(t, err)
+
+	// Test if the key has been deleted
+	exists, err = client.Exists(ctx, "key1")
+	assert.NoError(t, err)
+	assert.Equal(t, int64(0), exists)
+}

bcs-common/common/redisclient/config.go

@@ -0,0 +1,20 @@
+package redisclient
+
+import "time"
+
+// Config 初始化 Redis 客户端需要使用的
+type Config struct {
+	Addrs      []string  // 节点列表
+	MasterName string    // 哨兵模式下的主节点名
+	Password   string    // 密码
+	DB         int       // 单节点模式下的数据库
+	Mode       RedisMode // single, sentinel, cluster
+
+	// Options configs
+	DialTimeout  time.Duration
+	ReadTimeout  time.Duration
+	WriteTimeout time.Duration
+	PoolSize     int
+	MinIdleConns int
+	IdleTimeout  time.Duration
+}

bcs-common/common/redisclient/sentinel.go

@@ -0,0 +1,68 @@
+package redisclient
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"github.com/go-redis/redis/v8"
+)
+
+type SentinelClient struct {
+	cli *redis.Client
+}
+
+func NewSentinelClient(config Config) (*SentinelClient, error) {
+	if config.Mode != SentinelMode {
+		return nil, errors.New("redis mode not supported")
+	}
+	cli := redis.NewFailoverClient(&redis.FailoverOptions{
+		MasterName:    config.MasterName,
+		SentinelAddrs: config.Addrs,
+		Password:      config.Password,
+		DB:            config.DB,
+		DialTimeout:   config.DialTimeout,
+		ReadTimeout:   config.ReadTimeout,
+		WriteTimeout:  config.WriteTimeout,
+		PoolSize:      config.PoolSize,
+		MinIdleConns:  config.MinIdleConns,
+		IdleTimeout:   config.IdleTimeout,
+	})
+	return &SentinelClient{cli: cli}, nil
+}
+
+func (c *SentinelClient) GetCli() redis.UniversalClient {
+	return c.cli
+}
+
+func (c *SentinelClient) Ping(ctx context.Context) (string, error) {
+	return c.cli.Ping(ctx).Result()
+}
+
+func (c *SentinelClient) Set(ctx context.Context, key string, value interface{}, duration time.Duration) (string, error) {
+	return c.cli.Set(ctx, key, value, duration).Result()
+}
+
+func (c *SentinelClient) SetNX(ctx context.Context, key string, value interface{}, expiration time.Duration) (bool, error) {
+	return c.cli.SetNX(ctx, key, value, expiration).Result()
+}
+
+func (c *SentinelClient) Get(ctx context.Context, key string) (string, error) {
+	return c.cli.Get(ctx, key).Result()
+}
+
+func (c *SentinelClient) Exists(ctx context.Context, key ...string) (int64, error) {
+	return c.cli.Exists(ctx, key...).Result()
+}
+
+func (c *SentinelClient) SetEX(ctx context.Context, key string, value interface{}, expiration time.Duration) (string, error) {
+	return c.cli.SetEX(ctx, key, value, expiration).Result()
+}
+
+func (c *SentinelClient) Del(ctx context.Context, key string) (int64, error) {
+	return c.cli.Del(ctx, key).Result()
+}
+
+func (c *SentinelClient) Expire(ctx context.Context, key string, duration time.Duration) (bool, error) {
+	return c.cli.Expire(ctx, key, duration).Result()
+}

bcs-common/common/redisclient/sentinel_test.go

@@ -0,0 +1,61 @@
+package redisclient
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+// setupClusterClient function for initializing Redis SentinelClient
+func setupSentinel(t *testing.T) *SentinelClient {
+	config := Config{
+		Mode:       SentinelMode,
+		Addrs:      []string{"127.0.0.1:5001"}, // Sentinel addresses
+		MasterName: "mymaster",                 // Master name
+		DB:         0,
+		Password:   "",
+	}
+	client, err := NewSentinelClient(config)
+	assert.NoError(t, err)
+	assert.NotNil(t, client)
+	return client
+}
+
+// TestSentinelClientPing tests SentinelClient connectivity
+func TestSentinelClientPing(t *testing.T) {
+	client := setupSentinel(t)
+	result, err := client.GetCli().Ping(context.TODO()).Result()
+	assert.NoError(t, err)
+	assert.Equal(t, "PONG", result)
+}
+
+// TestSentinelClient tests SentinelClient basic functionality
+func TestSentinelClient(t *testing.T) {
+	client := setupSentinel(t)
+	ctx := context.Background()
+
+	// Test Set operation
+	_, err := client.Set(ctx, "key1", "value1", 10*time.Second)
+	assert.NoError(t, err)
+
+	// Test Get operation
+	val, err := client.Get(ctx, "key1")
+	assert.NoError(t, err)
+	assert.Equal(t, "value1", val)
+
+	// Test Exists operation
+	exists, err := client.Exists(ctx, "key1")
+	assert.NoError(t, err)
+	assert.Equal(t, int64(1), exists)
+
+	// Test Del operation
+	_, err = client.Del(ctx, "key1")
+	assert.NoError(t, err)
+
+	// Test if the key has been deleted
+	exists, err = client.Exists(ctx, "key1")
+	assert.NoError(t, err)
+	assert.Equal(t, int64(0), exists)
+}

bcs-common/common/redisclient/single.go

@@ -0,0 +1,81 @@
+package redisclient
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"github.com/go-redis/redis/v8"
+)
+
+type SingleClient struct {
+	cli *redis.Client
+}
+
+// NewSingleClient init SingleClient from config
+func NewSingleClient(config Config) (*SingleClient, error) {
+	if config.Mode != SingleMode {
+		return nil, errors.New("redis mode not supported")
+	}
+	if len(config.Addrs) == 0 {
+		return nil, errors.New("address is empty")
+	}
+	cli := redis.NewClient(&redis.Options{
+		Addr:         config.Addrs[0],
+		Password:     config.Password,
+		DB:           config.DB,
+		DialTimeout:  config.DialTimeout,
+		ReadTimeout:  config.ReadTimeout,
+		WriteTimeout: config.WriteTimeout,
+		PoolSize:     config.PoolSize,
+		MinIdleConns: config.MinIdleConns,
+		IdleTimeout:  config.IdleTimeout,
+	})
+	return &SingleClient{cli: cli}, nil
+}
+
+// NewSingleClientFromDSN init SingleClient by dsn
+func NewSingleClientFromDSN(dsn string) (*SingleClient, error) {
+	options, err := redis.ParseURL(dsn)
+	if err != nil {
+		return nil, err
+	}
+	cli := redis.NewClient(options)
+	return &SingleClient{cli: cli}, nil
+}
+
+func (c *SingleClient) GetCli() redis.UniversalClient {
+	return c.cli
+}
+
+func (c *SingleClient) Ping(ctx context.Context) (string, error) {
+	return c.cli.Ping(ctx).Result()
+}
+
+func (c *SingleClient) Get(ctx context.Context, key string) (string, error) {
+	return c.cli.Get(ctx, key).Result()
+}
+
+func (c *SingleClient) Set(ctx context.Context, key string, value interface{}, duration time.Duration) (string, error) {
+	return c.cli.Set(ctx, key, value, duration).Result()
+}
+
+func (c *SingleClient) SetNX(ctx context.Context, key string, value interface{}, expiration time.Duration) (bool, error) {
+	return c.cli.SetNX(ctx, key, value, expiration).Result()
+}
+
+func (c *SingleClient) SetEX(ctx context.Context, key string, value interface{}, expiration time.Duration) (string, error) {
+	return c.cli.SetEX(ctx, key, value, expiration).Result()
+}
+
+func (c *SingleClient) Exists(ctx context.Context, key ...string) (int64, error) {
+	return c.cli.Exists(ctx, key...).Result()
+}
+
+func (c *SingleClient) Del(ctx context.Context, key string) (int64, error) {
+	return c.cli.Del(ctx, key).Result()
+}
+
+func (c *SingleClient) Expire(ctx context.Context, key string, duration time.Duration) (bool, error) {
+	return c.cli.Expire(ctx, key, duration).Result()
+}