package main

import (
    "context"
    "errors"
    "fmt"
    "log"
    "sync"
    "time"

    "github.com/redis/go-redis/v9" // Compatible with Dragonfly 1.20 (Redis protocol compliant)
)

// CacheClient wraps redis.Client to provide unified access to Redis 9 and Dragonfly 1.20
// Both backends are protocol-compatible, so the same client works with minimal config changes
type CacheClient struct {
    client *redis.Client
    mu     sync.RWMutex
    opts   *CacheOptions
}

// CacheOptions configures the cache client behavior
type CacheOptions struct {
    Addr         string        // Redis/Dragonfly instance address (e.g., "localhost:6379")
    Password     string        // Auth password (empty for no auth)
    DB           int           // Database number (0-15 for Redis, ignored by Dragonfly)
    PoolSize     int           // Max connections in pool (default 10)
    MinIdleConns int           // Min idle connections (default 5)
    MaxRetries   int           // Max retry attempts for failed operations (default 3)
    RetryDelay   time.Duration // Delay between retries (default 100ms)
}

// NewCacheClient initializes a new cache client for Redis 9 or Dragonfly 1.20
// Verifies backend compatibility on startup by checking the server version
func NewCacheClient(opts *CacheOptions) (*CacheClient, error) {
    if opts == nil {
        return nil, errors.New("cache options cannot be nil")
    }
    if opts.Addr == "" {
        return nil, errors.New("cache address is required")
    }

    // Set defaults for optional fields
    if opts.PoolSize == 0 {
        opts.PoolSize = 10
    }
    if opts.MinIdleConns == 0 {
        opts.MinIdleConns = 5
    }
    if opts.MaxRetries == 0 {
        opts.MaxRetries = 3
    }
    if opts.RetryDelay == 0 {
        opts.RetryDelay = 100 * time.Millisecond
    }

    client := redis.NewClient(&redis.Options{
        Addr:         opts.Addr,
        Password:     opts.Password,
        DB:           opts.DB,
        PoolSize:     opts.PoolSize,
        MinIdleConns: opts.MinIdleConns,
        MaxRetries:   opts.MaxRetries,
    })

    // Verify backend is reachable and get version for compatibility check
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()

    info, err := client.Info(ctx, "server").Result()
    if err != nil {
        return nil, fmt.Errorf("failed to connect to cache backend: %w", err)
    }

    // Log server version for debugging (Redis 9.x or Dragonfly 1.20.x expected)
    log.Printf("Connected to cache backend: %s", info)
    return &CacheClient{client: client, opts: opts}, nil
}

// Get retrieves a value from the cache with retry logic
func (c *CacheClient) Get(ctx context.Context, key string) (string, error) {
    if key == "" {
        return "", errors.New("cache key cannot be empty")
    }

    var (
        val string
        err error
    )
    for attempt := 0; attempt <= c.opts.MaxRetries; attempt++ {
        val, err = c.client.Get(ctx, key).Result()
        if err == nil {
            return val, nil
        }
        if errors.Is(err, redis.Nil) {
            return "", fmt.Errorf("key %s not found: %w", key, err)
        }
        if attempt < c.opts.MaxRetries {
            time.Sleep(c.opts.RetryDelay)
            log.Printf("Retrying Get for key %s (attempt %d/%d)", key, attempt+1, c.opts.MaxRetries)
        }
    }
    return "", fmt.Errorf("failed to get key %s after %d retries: %w", key, c.opts.MaxRetries, err)
}

// Set stores a value in the cache with TTL, with retry logic
func (c *CacheClient) Set(ctx context.Context, key string, value string, ttl time.Duration) error {
    if key == "" {
        return errors.New("cache key cannot be empty")
    }
    if value == "" {
        return errors.New("cache value cannot be empty")
    }

    var err error
    for attempt := 0; attempt <= c.opts.MaxRetries; attempt++ {
        err = c.client.Set(ctx, key, value, ttl).Err()
        if err == nil {
            return nil
        }
        if attempt < c.opts.MaxRetries {
            time.Sleep(c.opts.RetryDelay)
            log.Printf("Retrying Set for key %s (attempt %d/%d)", key, attempt+1, c.opts.MaxRetries)
        }
    }
    return fmt.Errorf("failed to set key %s after %d retries: %w", key, c.opts.MaxRetries, err)
}

// Close releases all resources used by the cache client
func (c *CacheClient) Close() error {
    c.mu.Lock()
    defer c.mu.Unlock()
    return c.client.Close()
}

func main() {
    // Example usage: Connect to Redis 9 or Dragonfly 1.20 by changing the Addr
    opts := &CacheOptions{
        Addr:         "localhost:6379", // Change to Dragonfly address (e.g., "localhost:6380") to test Dragonfly
        Password:     "",
        DB:           0,
        PoolSize:     20,
        MinIdleConns: 10,
        MaxRetries:   3,
        RetryDelay:   100 * time.Millisecond,
    }

    client, err := NewCacheClient(opts)
    if err != nil {
        log.Fatalf("Failed to initialize cache client: %v", err)
    }
    defer client.Close()

    ctx := context.Background()
    // Test Set operation
    err = client.Set(ctx, "benchmark:key:1", "test-value", 1*time.Hour)
    if err != nil {
        log.Fatalf("Failed to set key: %v", err)
    }

    // Test Get operation
    val, err := client.Get(ctx, "benchmark:key:1")
    if err != nil {
        log.Fatalf("Failed to get key: %v", err)
    }
    log.Printf("Retrieved value: %s", val)
}

import redis
import time
import json
import argparse
import subprocess
from typing import Dict, List, Tuple
from dataclasses import dataclass

# Dragonfly is Redis protocol compatible, so redis-py works with both backends
# GitHub repo for redis-py: https://github.com/redis/redis-py

@dataclass
class BenchmarkConfig:
    """Configuration for cache benchmark runs"""
    host: str
    port: int
    password: str
    db: int
    payload_size: int  # Bytes
    num_keys: int
    num_operations: int
    read_write_ratio: float  # 0.8 = 80% GET, 20% SET
    warmup_seconds: int = 30
    test_seconds: int = 300

class CacheBenchmarker:
    """Runs throughput and latency benchmarks against Redis 9 or Dragonfly 1.20"""

    def __init__(self, config: BenchmarkConfig):
        self.config = config
        self.client = redis.Redis(
            host=config.host,
            port=config.port,
            password=config.password,
            db=config.db,
            socket_timeout=5,
            socket_connect_timeout=5,
            retry_on_timeout=True,
            health_check_interval=30
        )
        self.results: Dict[str, float] = {}

    def _generate_payload(self, size: int) -> bytes:
        """Generate a random payload of specified size"""
        import os
        return os.urandom(size)

    def _warmup(self) -> None:
        """Run warmup workload to prime cache and connection pool"""
        print(f"Starting warmup for {self.config.warmup_seconds} seconds...")
        start = time.time()
        while (time.time() - start) < self.config.warmup_seconds:
            key = f"warmup:{int(time.time() * 1000)}"
            payload = self._generate_payload(self.config.payload_size)
            try:
                self.client.set(key, payload, ex=60)
                self.client.get(key)
            except redis.RedisError as e:
                print(f"Warmup error: {e}")
        print("Warmup complete.")

    def run_throughput_test(self) -> Tuple[float, float]:
        """Run throughput test and return (ops_per_sec, p99_latency_ms)"""
        print(f"Running throughput test: {self.config.num_operations} operations, "
              f"{self.config.read_write_ratio*100}% GET, {self.config.payload_size}B payload")

        latencies: List[float] = []
        success_ops = 0
        failed_ops = 0
        start_time = time.time()

        for i in range(self.config.num_operations):
            key = f"bench:{i % self.config.num_keys}"
            is_read = (i % int(1 / self.config.read_write_ratio)) < 1  # Simplified R/W split
            op_start = time.time()

            try:
                if is_read:
                    self.client.get(key)
                else:
                    payload = self._generate_payload(self.config.payload_size)
                    self.client.set(key, payload, ex=3600)
                op_end = time.time()
                latencies.append((op_end - op_start) * 1000)  # Convert to ms
                success_ops += 1
            except redis.RedisError as e:
                failed_ops += 1
                if failed_ops % 100 == 0:
                    print(f"Operation failed: {e}")

        total_time = time.time() - start_time
        ops_per_sec = success_ops / total_time if total_time > 0 else 0
        p99_latency = sorted(latencies)[int(len(latencies) * 0.99)] if latencies else 0

        print(f"Throughput: {ops_per_sec:.0f} ops/sec")
        print(f"p99 Latency: {p99_latency:.2f} ms")
        print(f"Failed operations: {failed_ops}")

        return ops_per_sec, p99_latency

    def run_memory_test(self) -> float:
        """Store 10M items and return total memory used in GB"""
        print(f"Running memory test: storing {self.config.num_keys} items of {self.config.payload_size}B")
        start_mem = self._get_used_memory()

        for i in range(self.config.num_keys):
            key = f"mem:{i}"
            payload = self._generate_payload(self.config.payload_size)
            try:
                self.client.set(key, payload, ex=86400)  # 24h TTL
            except redis.RedisError as e:
                print(f"Failed to set key {key}: {e}")

        end_mem = self._get_used_memory()
        memory_used_gb = (end_mem - start_mem) / (1024 ** 3)
        print(f"Memory used for {self.config.num_keys} items: {memory_used_gb:.2f} GB")
        return memory_used_gb

    def _get_used_memory(self) -> int:
        """Get used memory in bytes from the cache backend"""
        info = self.client.info("memory")
        return info["used_memory"]

    def run_all_benchmarks(self) -> None:
        """Run all benchmark suites and print results"""
        self._warmup()
        ops_per_sec, p99_latency = self.run_throughput_test()
        memory_used = self.run_memory_test()

        self.results = {
            "ops_per_sec": ops_per_sec,
            "p99_latency_ms": p99_latency,
            "memory_used_gb": memory_used,
            "backend": self._get_backend_version()
        }

        print("\n=== Benchmark Results ===")
        print(json.dumps(self.results, indent=2))

    def _get_backend_version(self) -> str:
        """Get the cache backend version string"""
        info = self.client.info("server")
        return info.get("redis_version", "unknown")

    def close(self) -> None:
        """Close the Redis client connection"""
        self.client.close()

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Benchmark Redis 9 or Dragonfly 1.20")
    parser.add_argument("--host", default="localhost", help="Cache host")
    parser.add_argument("--port", type=int, default=6379, help="Cache port (6379 for Redis, 6380 for Dragonfly)")
    parser.add_argument("--password", default="", help="Cache password")
    parser.add_argument("--payload-size", type=int, default=1024, help="Payload size in bytes")
    parser.add_argument("--num-keys", type=int, default=100000, help="Number of unique keys")
    parser.add_argument("--num-ops", type=int, default=1000000, help="Number of operations")
    parser.add_argument("--rw-ratio", type=float, default=0.8, help="Read/write ratio (0.8 = 80% GET)")
    args = parser.parse_args()

    config = BenchmarkConfig(
        host=args.host,
        port=args.port,
        password=args.password,
        db=0,
        payload_size=args.payload_size,
        num_keys=args.num_keys,
        num_operations=args.num_ops,
        read_write_ratio=args.rw_ratio
    )

    benchmarker = CacheBenchmarker(config)
    try:
        benchmarker.run_all_benchmarks()
    finally:
        benchmarker.close()

import redis
import hashlib
import time
import argparse
import sys
from typing import Optional, Tuple

# GitHub repository for redis-py: https://github.com/redis/redis-py
# This script migrates data from Redis 9 to Dragonfly 1.20 with validation

class CacheMigrator:
    def __init__(self, src_host: str, src_port: int, dst_host: str, dst_port: int, 
                 password: str = "", batch_size: int = 1000, validate: bool = True):
        self.src = redis.Redis(host=src_host, port=src_port, password=password, db=0, socket_timeout=10)
        self.dst = redis.Redis(host=dst_host, port=dst_port, password=password, db=0, socket_timeout=10)
        self.batch_size = batch_size
        self.validate = validate
        self.total_keys = 0
        self.migrated_keys = 0
        self.failed_keys = 0
        self.validated_keys = 0

    def _get_key_checksum(self, client: redis.Redis, key: str) -> Optional[str]:
        """Get MD5 checksum of a key's value for validation"""
        try:
            value = client.get(key)
            if value is None:
                return None
            return hashlib.md5(value).hexdigest()
        except redis.RedisError:
            return None

    def _scan_keys(self) -> list:
        """Scan all keys in the source Redis instance"""
        print("Scanning source keys...")
        keys = []
        cursor = 0
        while True:
            cursor, batch = self.src.scan(cursor, count=self.batch_size)
            keys.extend(batch)
            if cursor == 0:
                break
        self.total_keys = len(keys)
        print(f"Found {self.total_keys} keys to migrate")
        return keys

    def _migrate_batch(self, keys: list) -> None:
        """Migrate a batch of keys from source to destination"""
        pipe = self.src.pipeline()
        for key in keys:
            pipe.type(key)
            pipe.ttl(key)
            pipe.get(key)  # Only handles string keys for simplicity; extend for hashes, sets, etc.
        results = pipe.execute()

        dst_pipe = self.dst.pipeline()
        for i, key in enumerate(keys):
            key_type = results[i*3]
            ttl = results[i*3 + 1]
            value = results[i*3 + 2]

            if key_type != "string":
                print(f"Skipping non-string key {key} (type: {key_type})")
                self.failed_keys += 1
                continue
            if value is None:
                continue  # Key expired during scan

            # Set value with original TTL
            if ttl == -1:
                dst_pipe.set(key, value)
            elif ttl > 0:
                dst_pipe.setex(key, ttl, value)
            else:
                dst_pipe.set(key, value)  # TTL -2 means no expiry

        try:
            dst_pipe.execute()
            self.migrated_keys += len(keys) - self.failed_keys
        except redis.RedisError as e:
            print(f"Failed to migrate batch: {e}")
            self.failed_keys += len(keys)

    def _validate_migration(self, keys: list) -> None:
        """Validate migrated keys by comparing checksums"""
        if not self.validate:
            return
        print("Validating migrated keys...")
        for key in keys:
            src_checksum = self._get_key_checksum(self.src, key)
            dst_checksum = self._get_key_checksum(self.dst, key)
            if src_checksum != dst_checksum:
                print(f"Validation failed for key {key}")
                self.failed_keys += 1
            else:
                self.validated_keys += 1

    def run_migration(self) -> None:
        """Run the full migration process"""
        start_time = time.time()
        # Verify source and destination are reachable
        try:
            src_version = self.src.info("server")["redis_version"]
            dst_version = self.dst.info("server")["redis_version"]
            print(f"Source: Redis {src_version} at {self.src.host}:{self.src.port}")
            print(f"Destination: Dragonfly {dst_version} at {self.dst.host}:{self.dst.port}")
        except redis.RedisError as e:
            print(f"Failed to connect to source or destination: {e}")
            sys.exit(1)

        keys = self._scan_keys()
        if not keys:
            print("No keys to migrate")
            return

        # Migrate in batches
        for i in range(0, len(keys), self.batch_size):
            batch = keys[i:i + self.batch_size]
            print(f"Migrating batch {i//self.batch_size + 1}/{(len(keys) + self.batch_size -1)//self.batch_size}")
            self._migrate_batch(batch)
            progress = (i + len(batch)) / len(keys) * 100
            print(f"Progress: {progress:.1f}% ({self.migrated_keys}/{self.total_keys} keys)")

        # Validate if enabled
        if self.validate:
            self._validate_migration(keys)

        total_time = time.time() - start_time
        print("\n=== Migration Complete ===")
        print(f"Total time: {total_time:.1f} seconds")
        print(f"Migrated keys: {self.migrated_keys}")
        print(f"Failed keys: {self.failed_keys}")
        if self.validate:
            print(f"Validated keys: {self.validated_keys}")

    def close(self) -> None:
        self.src.close()
        self.dst.close()

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Migrate data from Redis 9 to Dragonfly 1.20")
    parser.add_argument("--src-host", default="localhost", help="Source Redis host")
    parser.add_argument("--src-port", type=int, default=6379, help="Source Redis port")
    parser.add_argument("--dst-host", default="localhost", help="Destination Dragonfly host")
    parser.add_argument("--dst-port", type=int, default=6380, help="Destination Dragonfly port")
    parser.add_argument("--password", default="", help="Auth password for both instances")
    parser.add_argument("--batch-size", type=int, default=1000, help="Migration batch size")
    parser.add_argument("--no-validate", action="store_true", help="Disable migration validation")
    args = parser.parse_args()

    migrator = CacheMigrator(
        src_host=args.src_host,
        src_port=args.src_port,
        dst_host=args.dst_host,
        dst_port=args.dst_port,
        password=args.password,
        batch_size=args.batch_size,
        validate=not args.no_validate
    )
    try:
        migrator.run_migration()
    finally:
        migrator.close()

# Dragonfly 1.20 launch command with tuned thread count for 16 vCPU instance
./dragonfly \
  --thread_count=16 \
  --port=6380 \
  --max_memory=28GB \
  --log_level=info \
  --enable_tls=false

# Redis 9 redis.conf snippet to enable IO threading
io-threads 4
io-threads-do-reads yes
maxmemory 28GB
maxmemory-policy allkeys-lru
appendonly yes
cluster-enabled no

# Python snippet to test command compatibility
import redis

def test_command_compatibility(host: str, port: int):
    client = redis.Redis(host=host, port=port)
    test_commands = [
        ("SET", "compat:test", "1"),
        ("GET", "compat:test"),
        ("GEOADD", "cities", 13.361389, 38.115556, "Palermo"),
        ("CLUSTER", "nodes"),  # Will fail on Dragonfly
    ]
    for cmd in test_commands:
        try:
            getattr(client, cmd[0].lower())(*cmd[1:])
            print(f"Command {cmd[0]} supported")
        except redis.RedisError as e:
            print(f"Command {cmd[0]} unsupported: {e}")
    client.close()

test_command_compatibility("localhost", 6380)  # Test Dragonfly