Merge pull request #193 from menloresearch/add-redis-universal-connections

Switch to use Redis Universal connections

Author: locnguyen1986

apps/jan-api-gateway/application/app/infrastructure/cache/redis_cache_service.go

@@ -3,45 +3,46 @@ package cache
 import (
 	"context"
 	"fmt"
-	"strconv"
+	"strings"
 	"time"
 
 	"github.com/go-redsync/redsync/v4"
+	"github.com/go-redsync/redsync/v4/redis/goredis/v9"
 	"github.com/redis/go-redis/v9"
 	"menlo.ai/jan-api-gateway/app/utils/logger"
 	"menlo.ai/jan-api-gateway/config/environment_variables"
 )
 
-// RedisCacheService provides caching functionality using Redis
 type RedisCacheService struct {
-	client *redis.Client
+	client redis.UniversalClient
 	rs     *redsync.Redsync
 }
 
-// NewRedisCacheService creates a new Redis cache service
 func NewRedisCacheService() *RedisCacheService {
 	redisURL := environment_variables.EnvironmentVariables.REDIS_URL
 	if redisURL == "" {
 		panic("REDIS_URL environment variable must be set")
 	}
 
-	opts, err := redis.ParseURL(redisURL)
+	opts, err := buildUniversalOptions(redisURL)
 	if err != nil {
 		panic(fmt.Sprintf("failed to parse Redis URL: %v", err))
 	}
 
-	if environment_variables.EnvironmentVariables.REDIS_PASSWORD != "" {
-		opts.Password = environment_variables.EnvironmentVariables.REDIS_PASSWORD
+	if pwd := environment_variables.EnvironmentVariables.REDIS_PASSWORD; pwd != "" {
+		opts.Password = pwd
 	}
-	if environment_variables.EnvironmentVariables.REDIS_DB != "" {
-		if db, err := strconv.Atoi(environment_variables.EnvironmentVariables.REDIS_DB); err == nil {
-			opts.DB = db
-		} else {
-			panic(fmt.Sprintf("invalid REDIS_DB value: %v", err))
-		}
+
+	if dbVal := environment_variables.EnvironmentVariables.REDIS_DB; dbVal != 0 {
+		opts.DB = dbVal
+	}
+
+	if len(opts.Addrs) > 1 && opts.DB != 0 {
+		logger.GetLogger().Warn("Ignoring non-zero REDIS_DB when using Redis Cluster configuration")
+		opts.DB = 0
 	}
 
-	client := redis.NewClient(opts)
+	client := redis.NewUniversalClient(opts)
 
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
@@ -52,17 +53,83 @@ func NewRedisCacheService() *RedisCacheService {
 
 	logger.GetLogger().Info("Successfully connected to Redis")
 
+	rs := redsync.New(goredis.NewPool(client))
+
 	return &RedisCacheService{
 		client: client,
+		rs:     rs,
 	}
 }
 
-// Set stores a string value in Redis with an expiration time
+func buildUniversalOptions(raw string) (*redis.UniversalOptions, error) {
+	parts := strings.Split(raw, ",")
+	opts := &redis.UniversalOptions{}
+
+	for _, part := range parts {
+		part = strings.TrimSpace(part)
+		if part == "" {
+			continue
+		}
+
+		if strings.Contains(part, "://") {
+			parsed, err := redis.ParseURL(part)
+			if err != nil {
+				return nil, err
+			}
+
+			opts.Addrs = append(opts.Addrs, parsed.Addr)
+
+			if opts.Username == "" {
+				opts.Username = parsed.Username
+			}
+
+			if opts.Password == "" {
+				opts.Password = parsed.Password
+			}
+
+			if opts.DB == 0 {
+				opts.DB = parsed.DB
+			}
+
+			if opts.TLSConfig == nil {
+				opts.TLSConfig = parsed.TLSConfig
+			}
+
+			if opts.ReadTimeout == 0 {
+				opts.ReadTimeout = parsed.ReadTimeout
+			}
+
+			if opts.WriteTimeout == 0 {
+				opts.WriteTimeout = parsed.WriteTimeout
+			}
+
+			if opts.DialTimeout == 0 {
+				opts.DialTimeout = parsed.DialTimeout
+			}
+
+			if opts.PoolSize == 0 {
+				opts.PoolSize = parsed.PoolSize
+			}
+
+			if opts.MinIdleConns == 0 {
+				opts.MinIdleConns = parsed.MinIdleConns
+			}
+		} else {
+			opts.Addrs = append(opts.Addrs, part)
+		}
+	}
+
+	if len(opts.Addrs) == 0 {
+		return nil, fmt.Errorf("no Redis addresses provided")
+	}
+
+	return opts, nil
+}
+
 func (r *RedisCacheService) Set(ctx context.Context, key string, value string, expiration time.Duration) error {
 	return r.client.Set(ctx, key, value, expiration).Err()
 }
 
-// Get retrieves a string value from Redis
 func (r *RedisCacheService) Get(ctx context.Context, key string) (string, error) {
 	val, err := r.client.Get(ctx, key).Result()
 	if err != nil {
@@ -75,40 +142,32 @@ func (r *RedisCacheService) Get(ctx context.Context, key string) (string, error)
 	return val, nil
 }
 
-// GetWithFallback retrieves a string value from Redis, or executes fallback function if not found
 func (r *RedisCacheService) GetWithFallback(ctx context.Context, key string, fallback func() (string, error), expiration time.Duration) (string, error) {
-	// Try to get from cache first
 	result, err := r.Get(ctx, key)
 	if err == nil {
-		return result, nil // Found in cache
+		return result, nil
 	}
 
-	// Cache miss, execute fallback
 	result, err = fallback()
 	if err != nil {
 		return "", fmt.Errorf("fallback function failed: %w", err)
 	}
 
-	// Store in cache for future requests
 	if err := r.Set(ctx, key, result, expiration); err != nil {
 		logger.GetLogger().Error(fmt.Sprintf("Failed to cache value: %v", err))
-		// Don't return error, just log it
 	}
 
 	return result, nil
 }
 
-// Delete removes a key from Redis synchronously (blocking)
 func (r *RedisCacheService) Delete(ctx context.Context, key string) error {
 	return r.client.Del(ctx, key).Err()
 }
 
-// Unlink removes a key from Redis asynchronously (non-blocking)
 func (r *RedisCacheService) Unlink(ctx context.Context, key string) error {
 	return r.client.Unlink(ctx, key).Err()
 }
 
-// DeletePattern removes all keys matching a pattern
 func (r *RedisCacheService) DeletePattern(ctx context.Context, pattern string) error {
 	var cursor uint64
 	for {
@@ -133,7 +192,6 @@ func (r *RedisCacheService) DeletePattern(ctx context.Context, pattern string) e
 	return nil
 }
 
-// Exists checks if a key exists in Redis
 func (r *RedisCacheService) Exists(ctx context.Context, key string) (bool, error) {
 	result, err := r.client.Exists(ctx, key).Result()
 	if err != nil {
@@ -142,37 +200,29 @@ func (r *RedisCacheService) Exists(ctx context.Context, key string) (bool, error
 	return result > 0, nil
 }
 
-// Close closes the Redis connection
 func (r *RedisCacheService) Close() error {
 	return r.client.Close()
 }
 
-// HealthCheck verifies Redis connectivity
 func (r *RedisCacheService) HealthCheck(ctx context.Context) error {
 	return r.client.Ping(ctx).Err()
 }
 
-// NewMutex creates a new distributed mutex using go-redsync
 func (r *RedisCacheService) NewMutex(name string, options ...redsync.Option) *redsync.Mutex {
 	return r.rs.NewMutex(name, options...)
 }
 
-// WithLock executes a function with a distributed lock using go-redsync
 func WithLock(cache RedisCacheService, lockName string, fn func() error, ttl time.Duration) error {
 	mutex := cache.NewMutex(lockName, redsync.WithExpiry(ttl))
 
-	// Acquire lock
 	if err := mutex.Lock(); err != nil {
 		return err
 	}
 
-	// Ensure lock is released
 	defer func() {
 		if _, err := mutex.Unlock(); err != nil {
-			// swallow unlock error
 		}
 	}()
 
-	// Execute the function
 	return fn()
 }

apps/jan-api-gateway/application/config/environment_variables/env.go

@@ -1,13 +1,13 @@
 package environment_variables
 
 import (
-	"fmt"
 	"os"
 	"reflect"
 	"strconv"
 	"strings"
 	"time"
 
+	"menlo.ai/jan-api-gateway/app/utils/logger"
 	"menlo.ai/jan-api-gateway/config"
 )
 
@@ -32,7 +32,7 @@ type EnvironmentVariable struct {
 	// Redis configuration
 	REDIS_URL      string
 	REDIS_PASSWORD string
-	REDIS_DB       string
+	REDIS_DB       int
 }
 
 func (ev *EnvironmentVariable) LoadFromEnv() {
@@ -43,7 +43,7 @@ func (ev *EnvironmentVariable) LoadFromEnv() {
 		envKey := field.Name
 		envValue := os.Getenv(envKey)
 		if envValue == "" {
-			fmt.Printf("Missing SYSENV: %s\n", envKey)
+			logger.GetLogger().Warnf("Missing SYSENV: %s", envKey)
 		}
 		if envValue != "" {
 			switch v.Field(i).Kind() {
@@ -52,14 +52,14 @@ func (ev *EnvironmentVariable) LoadFromEnv() {
 			case reflect.Int:
 				intV, err := strconv.Atoi(envValue)
 				if err != nil {
-					fmt.Printf("Invalid int value for %s: %s\n", envKey, envValue)
+					logger.GetLogger().Errorf("Invalid int value for %s: %s", envKey, envValue)
 				} else {
 					v.Field(i).SetInt(int64(intV))
 				}
 			case reflect.Bool:
 				boolVal, err := strconv.ParseBool(envValue)
 				if err != nil {
-					fmt.Printf("Invalid boolean value for %s: %s\n", envKey, envValue)
+					logger.GetLogger().Errorf("Invalid boolean value for %s: %s", envKey, envValue)
 				} else {
 					v.Field(i).SetBool(boolVal)
 				}
@@ -70,10 +70,10 @@ func (ev *EnvironmentVariable) LoadFromEnv() {
 					hosts := strings.Split(envValue, ",")
 					v.Field(i).Set(reflect.ValueOf(hosts))
 				} else {
-					fmt.Printf("Unsupported slice type for %s\n", field.Name)
+					logger.GetLogger().Errorf("Unsupported slice type for %s", field.Name)
 				}
 			default:
-				fmt.Printf("Unsupported field type: %s\n", field.Name)
+				logger.GetLogger().Errorf("Unsupported field type: %s", field.Name)
 			}
 		}
 	}