Scaling Strategies

Redis Cluster Management

import redis
import rediscluster
from typing import Dict, List, Any, Optional
import hashlib
import asyncio

class RedisClusterManager:
    """Manage Redis cluster for horizontal cache scaling"""

    def __init__(self, cluster_nodes: List[Dict[str, Any]]):
        self.cluster_nodes = cluster_nodes
        self.cluster = None
        self.node_clients = {}
        self.hash_slots = 16384  # Redis cluster hash slots

    async def initialize(self):
        """Initialize Redis cluster connection"""

        startup_nodes = [
            {"host": node["host"], "port": node["port"]}
            for node in self.cluster_nodes
        ]

        self.cluster = rediscluster.RedisCluster(
            startup_nodes=startup_nodes,
            decode_responses=True,
            skip_full_coverage_check=True,
            health_check_interval=30,
            retry_on_timeout=True,
            socket_keepalive=True,
            socket_keepalive_options={
                1: 1,  # TCP_KEEPIDLE
                2: 3,  # TCP_KEEPINTVL
                3: 5   # TCP_KEEPCNT
            }
        )

        # Initialize individual node clients for monitoring
        for node in self.cluster_nodes:
            node_key = f"{node['host']}:{node['port']}"
            self.node_clients[node_key] = redis.Redis(
                host=node['host'],
                port=node['port'],
                decode_responses=True
            )

    async def set_with_scaling(self, key: str, value: Any, ttl: Optional[int] = None):
        """Set value with cluster scaling considerations"""

        try:
            serialized_value = self._serialize_value(value)

            if ttl:
                await self.cluster.setex(key, ttl, serialized_value)
            else:
                await self.cluster.set(key, serialized_value)

        except rediscluster.exceptions.ClusterDownError:
            # Handle cluster scaling events
            await self._handle_cluster_scaling()
            raise

    async def get_with_scaling(self, key: str) -> Optional[Any]:
        """Get value with cluster scaling considerations"""

        try:
            value = await self.cluster.get(key)
            if value:
                return self._deserialize_value(value)
            return None

        except rediscluster.exceptions.ClusterDownError:
            await self._handle_cluster_scaling()
            raise

    async def mget_with_scaling(self, keys: List[str]) -> List[Optional[Any]]:
        """Multi-get with cluster scaling optimization"""

        # Group keys by cluster slots for efficient retrieval
        slot_groups = self._group_keys_by_slot(keys)

        results = {}
        for slot, slot_keys in slot_groups.items():
            try:
                slot_results = await self.cluster.mget(slot_keys)
                for key, value in zip(slot_keys, slot_results):
                    results[key] = self._deserialize_value(value) if value else None
            except Exception as e:
                logging.warning(f"Failed to get keys from slot {slot}: {e}")
                # Set None for failed keys
                for key in slot_keys:
                    results[key] = None

        # Return results in original key order
        return [results.get(key) for key in keys]

    def _group_keys_by_slot(self, keys: List[str]) -> Dict[int, List[str]]:
        """Group keys by Redis cluster hash slots"""

        slot_groups = {}
        for key in keys:
            slot = self._calculate_slot(key)
            if slot not in slot_groups:
                slot_groups[slot] = []
            slot_groups[slot].append(key)

        return slot_groups

    def _calculate_slot(self, key: str) -> int:
        """Calculate Redis cluster slot for key"""

        # Redis cluster key hashing
        if '{' in key and '}' in key:
            # Hash tag support
            start = key.find('{')
            end = key.find('}', start)
            if end > start + 1:
                key = key[start + 1:end]

        return rediscluster.crc16.crc16(key.encode('utf-8')) % self.hash_slots

    async def _handle_cluster_scaling(self):
        """Handle cluster scaling events"""

        try:
            # Refresh cluster topology
            await self.cluster.cluster_nodes()
            logging.info("Redis cluster topology refreshed")

        except Exception as e:
            logging.error(f"Failed to refresh cluster topology: {e}")

    async def scale_cluster(self, new_nodes: List[Dict[str, Any]]):
        """Scale Redis cluster by adding new nodes"""

        for node in new_nodes:
            try:
                # Add node to cluster
                await self._add_node_to_cluster(node)

                # Update node clients
                node_key = f"{node['host']}:{node['port']}"
                self.node_clients[node_key] = redis.Redis(
                    host=node['host'],
                    port=node['port'],
                    decode_responses=True
                )

                logging.info(f"Added node {node_key} to Redis cluster")

            except Exception as e:
                logging.error(f"Failed to add node {node}: {e}")

        # Rebalance cluster
        await self._rebalance_cluster()

    async def _add_node_to_cluster(self, node: Dict[str, Any]):
        """Add new node to existing cluster"""

        # This would integrate with Redis cluster management commands
        # Implementation depends on your cluster management setup
        pass

    async def _rebalance_cluster(self):
        """Rebalance cluster slots after scaling"""

        # This would implement cluster slot rebalancing
        # Implementation depends on your cluster management setup
        pass

    async def get_cluster_stats(self) -> Dict[str, Any]:
        """Get comprehensive cluster statistics"""

        stats = {
            'nodes': {},
            'total_memory': 0,
            'total_keys': 0,
            'cluster_health': 'unknown'
        }

        healthy_nodes = 0
        total_nodes = len(self.node_clients)

        for node_key, client in self.node_clients.items():
            try:
                info = await client.info()
                node_stats = {
                    'memory_used': info.get('used_memory', 0),
                    'keys': info.get('db0', {}).get('keys', 0),
                    'connected_clients': info.get('connected_clients', 0),
                    'ops_per_sec': info.get('instantaneous_ops_per_sec', 0),
                    'status': 'healthy'
                }

                stats['nodes'][node_key] = node_stats
                stats['total_memory'] += node_stats['memory_used']
                stats['total_keys'] += node_stats['keys']
                healthy_nodes += 1

            except Exception as e:
                stats['nodes'][node_key] = {
                    'status': 'unhealthy',
                    'error': str(e)
                }

        # Determine cluster health
        health_ratio = healthy_nodes / total_nodes
        if health_ratio >= 0.9:
            stats['cluster_health'] = 'healthy'
        elif health_ratio >= 0.7:
            stats['cluster_health'] = 'degraded'
        else:
            stats['cluster_health'] = 'critical'

        stats['healthy_nodes'] = healthy_nodes
        stats['total_nodes'] = total_nodes
        stats['health_ratio'] = health_ratio

        return stats

    def _serialize_value(self, value: Any) -> str:
        """Serialize value for Redis storage"""
        import json
        return json.dumps(value, default=str)

    def _deserialize_value(self, value: str) -> Any:
        """Deserialize value from Redis storage"""
        import json
        try:
            return json.loads(value)
        except (json.JSONDecodeError, TypeError):
            return value


class CacheScalingStrategy:
    """Strategy for scaling cache layer based on load"""

    def __init__(self, redis_manager: RedisClusterManager):
        self.redis_manager = redis_manager
        self.scaling_thresholds = {
            'memory_utilization': 0.8,  # 80%
            'ops_per_second': 10000,    # 10K ops/sec per node
            'response_time': 10         # 10ms average response time
        }

    async def evaluate_scaling_needs(self) -> Dict[str, Any]:
        """Evaluate if cache layer needs scaling"""

        stats = await self.redis_manager.get_cluster_stats()

        scaling_recommendation = {
            'action': 'none',
            'reason': '',
            'recommended_nodes': 0,
            'current_performance': stats
        }

        # Check memory utilization
        if self._check_memory_pressure(stats):
            scaling_recommendation['action'] = 'scale_out'
            scaling_recommendation['reason'] = 'High memory utilization'
            scaling_recommendation['recommended_nodes'] = self._calculate_memory_scaling(stats)

        # Check operations per second
        elif self._check_ops_pressure(stats):
            scaling_recommendation['action'] = 'scale_out'
            scaling_recommendation['reason'] = 'High operations load'
            scaling_recommendation['recommended_nodes'] = self._calculate_ops_scaling(stats)

        # Check for over-provisioning
        elif self._check_underutilization(stats):
            scaling_recommendation['action'] = 'scale_in'
            scaling_recommendation['reason'] = 'Underutilized resources'
            scaling_recommendation['recommended_nodes'] = self._calculate_scale_in(stats)

        return scaling_recommendation

    def _check_memory_pressure(self, stats: Dict[str, Any]) -> bool:
        """Check if memory utilization is too high"""

        for node_key, node_stats in stats['nodes'].items():
            if node_stats.get('status') == 'healthy':
                # Assume 1GB max memory per node (configurable)
                max_memory = 1024 * 1024 * 1024
                memory_ratio = node_stats.get('memory_used', 0) / max_memory

                if memory_ratio > self.scaling_thresholds['memory_utilization']:
                    return True

        return False

    def _check_ops_pressure(self, stats: Dict[str, Any]) -> bool:
        """Check if operations load is too high"""

        for node_key, node_stats in stats['nodes'].items():
            if node_stats.get('status') == 'healthy':
                ops_per_sec = node_stats.get('ops_per_sec', 0)

                if ops_per_sec > self.scaling_thresholds['ops_per_second']:
                    return True

        return False

    def _check_underutilization(self, stats: Dict[str, Any]) -> bool:
        """Check if resources are underutilized"""

        healthy_nodes = [
            node_stats for node_stats in stats['nodes'].values()
            if node_stats.get('status') == 'healthy'
        ]

        if len(healthy_nodes) <= 3:  # Minimum cluster size
            return False

        # Check if all nodes are significantly underutilized
        underutilized_count = 0
        for node_stats in healthy_nodes:
            max_memory = 1024 * 1024 * 1024
            memory_ratio = node_stats.get('memory_used', 0) / max_memory
            ops_per_sec = node_stats.get('ops_per_sec', 0)

            if (memory_ratio < 0.3 and  # < 30% memory
                ops_per_sec < 1000):    # < 1K ops/sec
                underutilized_count += 1

        return underutilized_count >= len(healthy_nodes) * 0.7  # 70% underutilized

    def _calculate_memory_scaling(self, stats: Dict[str, Any]) -> int:
        """Calculate number of nodes needed for memory scaling"""

        total_memory = stats.get('total_memory', 0)
        healthy_nodes = stats.get('healthy_nodes', 1)

        # Target 60% memory utilization after scaling
        target_utilization = 0.6
        max_memory_per_node = 1024 * 1024 * 1024  # 1GB

        required_nodes = int(total_memory / (max_memory_per_node * target_utilization))
        additional_nodes = max(1, required_nodes - healthy_nodes)

        return additional_nodes

    def _calculate_ops_scaling(self, stats: Dict[str, Any]) -> int:
        """Calculate number of nodes needed for operations scaling"""

        # Find max ops per second across nodes
        max_ops = max(
            node_stats.get('ops_per_sec', 0)
            for node_stats in stats['nodes'].values()
            if node_stats.get('status') == 'healthy'
        )

        if max_ops > self.scaling_thresholds['ops_per_second']:
            # Add nodes to distribute load
            scale_factor = max_ops / self.scaling_thresholds['ops_per_second']
            return int(scale_factor)

        return 1

    def _calculate_scale_in(self, stats: Dict[str, Any]) -> int:
        """Calculate how many nodes can be removed"""

        healthy_nodes = stats.get('healthy_nodes', 0)
        min_nodes = 3  # Minimum for Redis cluster

        if healthy_nodes <= min_nodes:
            return 0

        # Conservative scale-in: remove at most 1 node at a time
        return min(1, healthy_nodes - min_nodes)