From 86c28571ad3f8f5b85df516378b8c1e58911074d Mon Sep 17 00:00:00 2001 From: Heathcliff Date: Thu, 12 Dec 2024 12:07:44 +0100 Subject: [PATCH] Remove deprecated redis package Fixes: #75 Signed-off-by: Heathcliff --- examples/example-config.yaml | 2 - pkg/lock-manager/config.go | 4 - pkg/lock-manager/manager.go | 6 - pkg/lock-manager/manager_test.go | 27 --- .../storage/redis/loadbalancer.go | 129 ------------- .../storage/redis/loadbalancer_test.go | 144 -------------- pkg/lock-manager/storage/redis/storage.go | 176 ------------------ tests/storage/redis_test.go | 113 ----------- 8 files changed, 601 deletions(-) delete mode 100644 pkg/lock-manager/storage/redis/loadbalancer.go delete mode 100644 pkg/lock-manager/storage/redis/loadbalancer_test.go delete mode 100644 pkg/lock-manager/storage/redis/storage.go delete mode 100644 tests/storage/redis_test.go diff --git a/examples/example-config.yaml b/examples/example-config.yaml index 8f01a7f8..3983c057 100644 --- a/examples/example-config.yaml +++ b/examples/example-config.yaml @@ -37,8 +37,6 @@ storage: # etcd: Use etcd database # kubernetes: Use kubernetes leases # - # redis: DEPRECATED, use valkey instead. This option will be removed in a future release in 2025 - # # Default: memory # type: memory diff --git a/pkg/lock-manager/config.go b/pkg/lock-manager/config.go index b0316ac9..55611245 100644 --- a/pkg/lock-manager/config.go +++ b/pkg/lock-manager/config.go @@ -4,9 +4,6 @@ import ( "github.com/heathcliff26/fleetlock/pkg/lock-manager/errors" "github.com/heathcliff26/fleetlock/pkg/lock-manager/storage/etcd" "github.com/heathcliff26/fleetlock/pkg/lock-manager/storage/kubernetes" - - //nolint:staticcheck - "github.com/heathcliff26/fleetlock/pkg/lock-manager/storage/redis" "github.com/heathcliff26/fleetlock/pkg/lock-manager/storage/sql" "github.com/heathcliff26/fleetlock/pkg/lock-manager/storage/valkey" ) @@ -17,7 +14,6 @@ type StorageConfig struct { Postgres sql.PostgresConfig `yaml:"postgres,omitempty"` MySQL sql.MySQLConfig `yaml:"mysql,omitempty"` Valkey valkey.ValkeyConfig `yaml:"valkey,omitempty"` - Redis redis.RedisConfig `yaml:"redis,omitempty"` Etcd etcd.EtcdConfig `yaml:"etcd,omitempty"` Kubernetes kubernetes.KubernetesConfig `yaml:"kubernetes,omitempty"` } diff --git a/pkg/lock-manager/manager.go b/pkg/lock-manager/manager.go index 16e9551d..b7d304e9 100644 --- a/pkg/lock-manager/manager.go +++ b/pkg/lock-manager/manager.go @@ -8,9 +8,6 @@ import ( "github.com/heathcliff26/fleetlock/pkg/lock-manager/storage/etcd" "github.com/heathcliff26/fleetlock/pkg/lock-manager/storage/kubernetes" "github.com/heathcliff26/fleetlock/pkg/lock-manager/storage/memory" - - //nolint:staticcheck - "github.com/heathcliff26/fleetlock/pkg/lock-manager/storage/redis" "github.com/heathcliff26/fleetlock/pkg/lock-manager/storage/sql" "github.com/heathcliff26/fleetlock/pkg/lock-manager/storage/valkey" "github.com/heathcliff26/fleetlock/pkg/lock-manager/types" @@ -64,9 +61,6 @@ func NewManager(groups Groups, storageCfg StorageConfig) (*LockManager, error) { storage, err = sql.NewPostgresBackend(storageCfg.Postgres) case "mysql": storage, err = sql.NewMySQLBackend(storageCfg.MySQL) - case "redis": - //nolint:staticcheck - storage, err = redis.NewRedisBackend(storageCfg.Redis) case "valkey": storage, err = valkey.NewValkeyBackend(storageCfg.Valkey) case "etcd": diff --git a/pkg/lock-manager/manager_test.go b/pkg/lock-manager/manager_test.go index 951c8fe2..99306354 100644 --- a/pkg/lock-manager/manager_test.go +++ b/pkg/lock-manager/manager_test.go @@ -7,9 +7,6 @@ import ( "github.com/alicebob/miniredis/v2" "github.com/heathcliff26/fleetlock/pkg/lock-manager/storage/etcd" "github.com/heathcliff26/fleetlock/pkg/lock-manager/storage/kubernetes" - - //nolint:staticcheck - "github.com/heathcliff26/fleetlock/pkg/lock-manager/storage/redis" "github.com/heathcliff26/fleetlock/pkg/lock-manager/storage/sql" "github.com/heathcliff26/fleetlock/pkg/lock-manager/storage/valkey" "github.com/stretchr/testify/assert" @@ -85,30 +82,6 @@ func TestNewManager(t *testing.T) { }, Error: "failed to open mysql database", }, - { - Name: "RedisBackend", - Storage: StorageConfig{ - Type: "redis", - Redis: redis.RedisConfig{ - Addr: mr.Addr(), - }, - }, - Result: result{ - groups: initGroups(NewDefaultGroups()), - storage: "*redis.RedisBackend", - }, - Error: "", - }, - { - Name: "ErrorNewRedisBackend", - Storage: StorageConfig{ - Type: "redis", - Redis: redis.RedisConfig{ - Addr: "", - }, - }, - Error: "no alive address in InitAddress", - }, { Name: "ValkeyBackend", Storage: StorageConfig{ diff --git a/pkg/lock-manager/storage/redis/loadbalancer.go b/pkg/lock-manager/storage/redis/loadbalancer.go deleted file mode 100644 index 32654d3b..00000000 --- a/pkg/lock-manager/storage/redis/loadbalancer.go +++ /dev/null @@ -1,129 +0,0 @@ -// Deprecated: The redis package has been replaced by valkey. It will be removed in 2025. -// See: https://github.com/heathcliff26/fleetlock/issues/67 -// -// This package is frozen and no new functionality will be added. -package redis - -import ( - "context" - "crypto/tls" - "log/slog" - "net" - "slices" - "strings" - "sync" - "time" - - "github.com/valkey-io/valkey-go" -) - -const loadbalancerHealtchCheckPeriod = time.Second * 10 - -type loadbalancer struct { - // List of addresses for redis endpoints - addrs []string - // Options for connecting to redis - options valkey.ClientOption - // Context to cancle health check - ctx context.Context - cancel context.CancelFunc - - client valkey.Client - selected int - rwlock sync.RWMutex -} - -// Deprecated: Use valkey.NewValkeyLoadbalancer instead -// See: https://github.com/heathcliff26/fleetlock/issues/67 -// -// Create a new redis client with loadbalanced connections -func NewRedisLoadbalancer(opt valkey.ClientOption) (valkey.Client, *loadbalancer, error) { - opt.ForceSingleClient = true - - ctx, cancel := context.WithCancel(context.Background()) - lb := &loadbalancer{ - addrs: opt.InitAddress, - options: opt, - ctx: ctx, - cancel: cancel, - } - - opt.DialFn = lb.DialFn - - client, err := valkey.NewClient(opt) - if err != nil { - return nil, nil, err - } - - lb.client = client - lb.PeriodicHealthCheck() - - return client, lb, nil -} - -// Determine the first healthy master node -func (lb *loadbalancer) HealthCheck() { - for i, addr := range lb.addrs { - opt := lb.options - opt.InitAddress = []string{addr} - - client, err := valkey.NewClient(opt) - if err != nil { - slog.Debug("Endpoint down", slog.String("addr", addr), "err", err) - continue - } - defer client.Close() - - cmdInfo := client.B().Info().Build() - res, err := client.Do(context.Background(), cmdInfo).ToString() - if err != nil { - slog.Error("Failed to get endpoint info", slog.String("addr", addr), "err", err) - continue - } - s := strings.Split(res, "\r\n") - if slices.Contains(s, "role:master") || slices.Contains(s, "role:active-replica") { - lb.rwlock.Lock() - defer lb.rwlock.Unlock() - - if lb.selected != i { - lb.selected = i - slog.Info("Failed over to new database", slog.String("addr", addr)) - // valkey keeps a connection. Try to ping it to ensure it gets terminated and the next try will be a new connection - _ = lb.client.Do(context.Background(), client.B().Ping().Build()) - } - break - } - } -} - -// Starts go-routine that periodically runs a healthcheck in the background -func (lb *loadbalancer) PeriodicHealthCheck() { - go lb.periodicHealthCheck() -} - -func (lb *loadbalancer) periodicHealthCheck() { - for { - lb.HealthCheck() - - select { - case <-lb.ctx.Done(): - return - case <-time.After(loadbalancerHealtchCheckPeriod): - } - } -} - -func (lb *loadbalancer) DialFn(_ string, dialer *net.Dialer, cfg *tls.Config) (conn net.Conn, err error) { - lb.rwlock.RLock() - defer lb.rwlock.RUnlock() - dst := lb.addrs[lb.selected] - - if cfg != nil { - return tls.DialWithDialer(dialer, "tcp", dst, cfg) - } - return dialer.Dial("tcp", dst) -} - -func (lb *loadbalancer) Close() { - lb.cancel() -} diff --git a/pkg/lock-manager/storage/redis/loadbalancer_test.go b/pkg/lock-manager/storage/redis/loadbalancer_test.go deleted file mode 100644 index 5ec5d1a1..00000000 --- a/pkg/lock-manager/storage/redis/loadbalancer_test.go +++ /dev/null @@ -1,144 +0,0 @@ -// Deprecated: The redis package has been replaced by valkey. It will be removed in 2025. -// See: https://github.com/heathcliff26/fleetlock/issues/67 -// -// This package is frozen and no new functionality will be added. -package redis - -import ( - "context" - "testing" - "time" - - "github.com/alicebob/miniredis/v2" - "github.com/heathcliff26/fleetlock/tests/utils" - "github.com/stretchr/testify/assert" - "github.com/valkey-io/valkey-go" -) - -func TestLoadbalancer(t *testing.T) { - t.Run("Basic", func(t *testing.T) { - mr1 := miniredis.RunT(t) - mr2 := miniredis.RunT(t) - - opt := valkey.ClientOption{ - InitAddress: []string{mr1.Addr(), mr2.Addr()}, - DisableCache: true, - } - - assert := assert.New(t) - - client, lb, err := NewRedisLoadbalancer(opt) - if !assert.NoError(err, "Should not return an error") || !assert.NotNil(client, "Should return a client") || !assert.NotNil(lb, "Should return a loadbalancer") { - t.FailNow() - } - t.Cleanup(func() { - lb.Close() - client.Close() - }) - - res, err := client.Do(context.Background(), client.B().Ping().Build()).ToString() - - assert.Nil(err, "Can reach client") - assert.Equal("PONG", res, "Can reach client") - }) - t.Run("Failover", func(t *testing.T) { - if !utils.HasContainerRuntimer() { - t.Skip("Missing Container Runtime") - } - - err := utils.ExecCRI("run", "--name", "fleetlock-redis-loadbalancer-failover-1", "-d", "--net", "host", "docker.io/eqalpha/keydb:latest", "--port", "6381", "--active-replica", "yes", "--replicaof", "localhost", "6382") - if err != nil { - t.Fatalf("Failed to start test db: %v\n", err) - } - t.Cleanup(func() { - _ = utils.ExecCRI("stop", "fleetlock-redis-loadbalancer-failover-1") - _ = utils.ExecCRI("rm", "fleetlock-redis-loadbalancer-failover-1") - }) - err = utils.ExecCRI("run", "--name", "fleetlock-redis-loadbalancer-failover-2", "-d", "--net", "host", "docker.io/eqalpha/keydb:latest", "--port", "6382", "--active-replica", "yes", "--replicaof", "localhost", "6381") - if err != nil { - t.Fatalf("Failed to start test db: %v\n", err) - } - t.Cleanup(func() { - _ = utils.ExecCRI("stop", "fleetlock-redis-loadbalancer-failover-2") - _ = utils.ExecCRI("rm", "fleetlock-redis-loadbalancer-failover-2") - }) - - assert := assert.New(t) - - opt := valkey.ClientOption{ - InitAddress: []string{"localhost:6381", "localhost:6382"}, - DisableCache: true, - } - client, lb, err := NewRedisLoadbalancer(opt) - if !assert.NoError(err, "Should not return an error") || !assert.NotNil(client, "Should return a client") || !assert.NotNil(lb, "Should return a loadbalancer") { - t.FailNow() - } - t.Cleanup(func() { - lb.Close() - client.Close() - }) - - assert.Equal(0, lb.selected, "Should have currently the first client selected") - - err = utils.ExecCRI("stop", "fleetlock-redis-loadbalancer-failover-1") - if err != nil { - t.Fatalf("Failed to stop keydb instance: %v\n", err) - } - - lb.HealthCheck() - if !assert.Equal(1, lb.selected, "Should have failed over") { - t.FailNow() - } - - res, err := client.Do(context.Background(), client.B().Ping().Build()).ToString() - - assert.NoError(err, "Should have failed over") - assert.Equal("PONG", res, "Should have failed over") - }) - t.Run("DeadlockCheck", func(t *testing.T) { - mr1 := miniredis.RunT(t) - mr2 := miniredis.RunT(t) - - opt := valkey.ClientOption{ - InitAddress: []string{mr1.Addr(), mr2.Addr()}, - DisableCache: true, - } - - assert := assert.New(t) - - client, lb, err := NewRedisLoadbalancer(opt) - if !assert.NoError(err, "Should not return an error") || !assert.NotNil(client, "Should return a client") || !assert.NotNil(lb, "Should return a loadbalancer") { - t.FailNow() - } - t.Cleanup(func() { - lb.Close() - client.Close() - }) - - // Ensure no failover happens automatically - lb.cancel() - - assert.Equal(0, lb.selected, "Should have currently the first client selected") - - mr1.Close() - - done := make(chan struct{}, 1) - - go func() { - lb.HealthCheck() - done <- struct{}{} - close(done) - }() - go func() { - _, err = client.Do(context.Background(), client.B().Ping().Build()).ToString() - - assert.Error(err, "Call should fail") - }() - - select { - case <-done: - case <-time.After(10 * time.Second): - t.Fatal("Timed out waiting for the failover to finished") - } - }) -} diff --git a/pkg/lock-manager/storage/redis/storage.go b/pkg/lock-manager/storage/redis/storage.go deleted file mode 100644 index 8a6b972b..00000000 --- a/pkg/lock-manager/storage/redis/storage.go +++ /dev/null @@ -1,176 +0,0 @@ -// Deprecated: The redis package has been replaced by valkey. It will be removed in 2025. -// See: https://github.com/heathcliff26/fleetlock/issues/67 -// -// This package is frozen and no new functionality will be added. -package redis - -import ( - "context" - "crypto/tls" - "fmt" - "log/slog" - "time" - - "github.com/heathcliff26/fleetlock/pkg/lock-manager/types" - "github.com/valkey-io/valkey-go" -) - -const keyformat = "group:%s,id:%s" - -type RedisBackend struct { - client valkey.Client - lb *loadbalancer -} - -type RedisConfig struct { - Addr string `yaml:"address,omitempty"` - Addrs []string `yaml:"addresses,omitempty"` - Username string `yaml:"username,omitempty"` - Password string `yaml:"password,omitempty"` - DB int `yaml:"db,omitempty"` - TLS bool `yaml:"tls,omitempty"` - Sentinel RedisSentinelConfig `yaml:"sentinel,omitempty"` -} - -type RedisSentinelConfig struct { - Enabled bool `yaml:"enabled,omitempty"` - MasterName string `yaml:"master,omitempty"` - Addresses []string `yaml:"addresses,omitempty"` - Username string `yaml:"username,omitempty"` - Password string `yaml:"password,omitempty"` -} - -// Deprecated: Use valkey.NewValkeyBackend instead -// See: https://github.com/heathcliff26/fleetlock/issues/67 -func NewRedisBackend(cfg RedisConfig) (*RedisBackend, error) { - var client valkey.Client - var lb *loadbalancer - var tlsConfig *tls.Config - - slog.Warn("DEPRECATED: The redis backend type has been renamed to valkey. The redis option will be removed in 2025. See: https://github.com/heathcliff26/fleetlock/issues/67") - - if cfg.TLS { - tlsConfig = &tls.Config{} - } - - if cfg.Addr != "" && len(cfg.Addrs) == 0 { - cfg.Addrs = []string{cfg.Addr} - } - - opt := valkey.ClientOption{ - InitAddress: cfg.Addrs, - Username: cfg.Username, - Password: cfg.Password, - SelectDB: cfg.DB, - TLSConfig: tlsConfig, - - DisableCache: true, - } - - var err error - switch { - case cfg.Sentinel.Enabled: - opt.Sentinel = valkey.SentinelOption{ - MasterSet: cfg.Sentinel.MasterName, - Username: cfg.Sentinel.Username, - Password: cfg.Sentinel.Password, - } - opt.InitAddress = cfg.Sentinel.Addresses - - client, err = valkey.NewClient(opt) - case len(cfg.Addrs) > 1: - client, lb, err = NewRedisLoadbalancer(opt) - default: - client, err = valkey.NewClient(opt) - } - if err != nil { - return nil, fmt.Errorf("failed to connect to redis server: %v", err) - } - - return &RedisBackend{ - client: client, - lb: lb, - }, nil -} - -// Reserve a lock for the given group. -// Returns true if the lock is successfully reserved, even if the lock is already held by the specific id -func (r *RedisBackend) Reserve(group string, id string) error { - key := fmt.Sprintf(keyformat, group, id) - ctx := context.Background() - - cmdSetNX := r.client.B().Setnx().Key(key).Value(time.Now().String()).Build() - cmdSAdd := r.client.B().Sadd().Key(group).Member(key).Build() - - ok, err := r.client.Do(ctx, cmdSetNX).AsBool() - if err != nil { - return fmt.Errorf("failed to create key: %w", err) - } - - if ok { - err := r.client.Do(ctx, cmdSAdd).Error() - if err != nil { - return fmt.Errorf("failed to add key to group list: %w", err) - } - } - - return nil -} - -// Returns the current number of locks for the given group -func (r *RedisBackend) GetLocks(group string) (int, error) { - cmdSCard := r.client.B().Scard().Key(group).Build() - result, err := r.client.Do(context.Background(), cmdSCard).AsInt64() - if err != nil { - return 0, fmt.Errorf("failed to get locks from database: %w", err) - } - return int(result), nil -} - -// Release the lock currently held by the id. -// Does not fail when no lock is held. -func (r *RedisBackend) Release(group string, id string) error { - key := fmt.Sprintf(keyformat, group, id) - ctx := context.Background() - - cmdDel := r.client.B().Del().Key(key).Build() - cmdSRem := r.client.B().Srem().Key(group).Member(key).Build() - - err := r.client.Do(ctx, cmdDel).Error() - if err != nil { - return fmt.Errorf("failed to delete key in database: %w", err) - } - - err = r.client.Do(ctx, cmdSRem).Error() - if err != nil { - return fmt.Errorf("failed to remove key from group: %w", err) - } - return nil -} - -// Return all locks older than x -func (r *RedisBackend) GetStaleLocks(ts time.Duration) ([]types.Lock, error) { - panic("not implemented") // TODO: Implement -} - -// Check if a given id already has a lock for this group -func (r *RedisBackend) HasLock(group string, id string) (bool, error) { - key := fmt.Sprintf(keyformat, group, id) - ctx := context.Background() - - cmdExists := r.client.B().Exists().Key(key).Build() - count, err := r.client.Do(ctx, cmdExists).AsInt64() - if err != nil { - return false, fmt.Errorf("failed to count keys in group: %w", err) - } - return count == 1, nil -} - -// Calls all necessary finalization if necessary -func (r *RedisBackend) Close() error { - if r.lb != nil { - r.lb.Close() - } - r.client.Close() - return nil -} diff --git a/tests/storage/redis_test.go b/tests/storage/redis_test.go deleted file mode 100644 index 351f62c1..00000000 --- a/tests/storage/redis_test.go +++ /dev/null @@ -1,113 +0,0 @@ -package storage - -import ( - "net" - "testing" - "time" - - "github.com/alicebob/miniredis/v2" - //nolint:staticcheck - "github.com/heathcliff26/fleetlock/pkg/lock-manager/storage/redis" - "github.com/heathcliff26/fleetlock/tests/utils" -) - -func TestRedisBackend(t *testing.T) { - mr := miniredis.RunT(t) - - cfg := redis.RedisConfig{ - Addr: mr.Addr(), - } - - //nolint:staticcheck - storage, err := redis.NewRedisBackend(cfg) - if err != nil { - t.Fatalf("Failed to create storage backend: %v", err) - } - - RunLockManagerTestsuiteWithStorage(t, storage) -} - -func TestRedisLoadbalancerBackend(t *testing.T) { - mr1 := miniredis.RunT(t) - mr2 := miniredis.RunT(t) - - cfg := redis.RedisConfig{ - Addrs: []string{mr1.Addr(), mr2.Addr()}, - } - - //nolint:staticcheck - storage, err := redis.NewRedisBackend(cfg) - if err != nil { - t.Fatalf("Failed to create storage backend: %v", err) - } - - RunLockManagerTestsuiteWithStorage(t, storage) -} - -func TestRedisSentinelBackend(t *testing.T) { - if !utils.HasContainerRuntimer() { - t.Skip("Missing Container Runtime") - } - - err := utils.ExecCRI("run", "--name", "fleetlock-redis-sentinel-db", "-d", "--net", "host", - "docker.io/valkey/valkey:latest", - "--port", "6379", - ) - if err != nil { - t.Fatalf("Failed to start test db: %v\n", err) - } - t.Cleanup(func() { - _ = utils.ExecCRI("stop", "fleetlock-redis-sentinel-db") - _ = utils.ExecCRI("rm", "fleetlock-redis-sentinel-db") - }) - - err = utils.ExecCRI("run", "--name", "fleetlock-redis-sentinel-sentinel", "-d", "--net", "host", - "-v", "./testdata/valkey-sentinel.conf:/config/sentinel.conf", "--userns=keep-id", - "docker.io/valkey/valkey:latest", - "/config/sentinel.conf", "--sentinel", - ) - if err != nil { - t.Fatalf("Failed to start test sentinel: %v\n", err) - } - t.Cleanup(func() { - _ = utils.ExecCRI("stop", "fleetlock-redis-sentinel-sentinel") - _ = utils.ExecCRI("rm", "fleetlock-redis-sentinel-sentinel") - }) - - for i := 0; i < 10; { - conn, err := net.Dial("tcp", "localhost:26379") - if err == nil { - conn.Close() - break - } - <-time.After(time.Second) - i++ - } - - cfg := redis.RedisConfig{ - Sentinel: redis.RedisSentinelConfig{ - Enabled: true, - MasterName: "valkey-sentinel-backend", - Addresses: []string{"localhost:26379"}, - }, - } - - //nolint:staticcheck - storage, err := redis.NewRedisBackend(cfg) - if err != nil { - cmd := utils.GetCommand("logs", "fleetlock-redis-sentinel-sentinel") - out, _ := cmd.Output() - t.Log("logs from sentinel:\n" + string(out)) - cmd = utils.GetCommand("logs", "fleetlock-redis-sentinel-db") - out, _ = cmd.Output() - t.Log("logs from db:\n" + string(out)) - - cmd = utils.GetCommand("ps", "-a") - out, _ = cmd.Output() - t.Log("Output of ps -a:\n" + string(out)) - - t.Fatalf("Failed to create storage backend: %v", err) - } - - RunLockManagerTestsuiteWithStorage(t, storage) -}