diff --git a/apps/channels/tasks.py b/apps/channels/tasks.py index 52956d53..59122d9f 100755 --- a/apps/channels/tasks.py +++ b/apps/channels/tasks.py @@ -2494,7 +2494,7 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): md = r.hgetall(metadata_key) if md: def _d(bkey, cast=str): - v = md.get(key) + v = md.get(bkey) try: if v is None: return None diff --git a/apps/proxy/ts_proxy/client_manager.py b/apps/proxy/ts_proxy/client_manager.py index e2cfd460..e1c92256 100644 --- a/apps/proxy/ts_proxy/client_manager.py +++ b/apps/proxy/ts_proxy/client_manager.py @@ -428,8 +428,7 @@ class ClientManager: client_id_list = list(client_ids) pipe = redis_client.pipeline() for cid in client_id_list: - cid_str = cid.decode('utf-8') - pipe.exists(RedisKeys.client_metadata(channel_id, cid_str)) + pipe.exists(RedisKeys.client_metadata(channel_id, cid)) results = pipe.execute() stale_ids = [ diff --git a/apps/proxy/ts_proxy/server.py b/apps/proxy/ts_proxy/server.py index b3449693..5316ec58 100644 --- a/apps/proxy/ts_proxy/server.py +++ b/apps/proxy/ts_proxy/server.py @@ -940,14 +940,14 @@ class ProxyServer: metadata = self.redis_client.hgetall(metadata_key) if metadata: # Calculate runtime from init_time - if b'init_time' in metadata: + if 'init_time' in metadata: try: init_time = float(metadata['init_time']) runtime = round(time.time() - init_time, 2) except Exception: pass # Get total bytes transferred - if b'total_bytes' in metadata: + if 'total_bytes' in metadata: try: total_bytes = int(metadata['total_bytes']) except Exception: @@ -1113,9 +1113,9 @@ class ProxyServer: # Also get init time as a fallback init_time = None - if metadata and b'init_time' in metadata: + if metadata and 'init_time' in metadata: try: - init_time = float(metadata[b'init_time']) + init_time = float(metadata['init_time']) except (ValueError, TypeError): pass @@ -1400,7 +1400,7 @@ class ProxyServer: real_count = max(0, client_count - len(stale_ids)) if real_count <= 0: # No real clients remain — safe to clean up. - state = metadata.get(b'state', b'unknown').decode('utf-8') if b'state' in metadata else 'unknown' + state = metadata.get('state', 'unknown') logger.warning( f"Orphaned channel {channel_id} (state: {state}, " f"owner: {owner}) had {client_count} ghost client(s) " diff --git a/apps/proxy/ts_proxy/services/channel_service.py b/apps/proxy/ts_proxy/services/channel_service.py index 0a0945d6..d0478e8f 100644 --- a/apps/proxy/ts_proxy/services/channel_service.py +++ b/apps/proxy/ts_proxy/services/channel_service.py @@ -382,8 +382,8 @@ class ChannelService: metadata = proxy_server.redis_client.hgetall(metadata_key) # Extract state and owner - state = metadata.get(ChannelMetadataField.STATE.encode(), 'unknown') - owner = metadata.get(ChannelMetadataField.OWNER.encode(), 'unknown') + state = metadata.get(ChannelMetadataField.STATE, 'unknown') + owner = metadata.get(ChannelMetadataField.OWNER, 'unknown') # Valid states indicate channel is running properly valid_states = [ChannelState.ACTIVE, ChannelState.WAITING_FOR_CLIENTS, ChannelState.CONNECTING] @@ -432,13 +432,13 @@ class ChannelService: try: # Use factory to parse the line based on stream type parsed_data = LogParserFactory.parse(stream_type, stream_info_line) - + if not parsed_data: return # Update Redis and database with parsed data ChannelService._update_stream_info_in_redis( - channel_id, + channel_id, parsed_data.get('video_codec'), parsed_data.get('resolution'), parsed_data.get('width'), diff --git a/apps/proxy/ts_proxy/views.py b/apps/proxy/ts_proxy/views.py index fabc1053..d94a03cf 100644 --- a/apps/proxy/ts_proxy/views.py +++ b/apps/proxy/ts_proxy/views.py @@ -127,7 +127,7 @@ def stream_ts(request, channel_id, user=None): ) # Unknown/empty state - check if owner is alive else: - owner_field = ChannelMetadataField.OWNER.encode("utf-8") + owner_field = ChannelMetadataField.OWNER if owner_field in metadata: owner = metadata[owner_field] owner_heartbeat_key = f"ts_proxy:worker:{owner}:heartbeat" diff --git a/apps/proxy/utils.py b/apps/proxy/utils.py index 0ea1bd27..1c71b5cd 100644 --- a/apps/proxy/utils.py +++ b/apps/proxy/utils.py @@ -11,7 +11,7 @@ def attempt_stream_termination(user_id, requesting_client_id, active_connections try: logger.info("[stream limits]" f"[{requesting_client_id}] User {user_id} has {len(active_connections)} active connections, checking termination candidates") - user_limit_settings = CoreSettings.get_user_limit_settings() + user_limit_settings = CoreSettings.get_user_limits_settings() terminate_oldest = user_limit_settings.get("terminate_oldest", True) prioritize_single = user_limit_settings.get("prioritize_single_client_channels", True) ignore_same_channel = user_limit_settings.get("ignore_same_channel_connections", False) @@ -79,74 +79,58 @@ def get_user_active_connections(user_id): connections = [] try: - cursor = 0 - # Grab live streams - while True: - cursor, keys = redis_client.scan(cursor=cursor, match="ts_proxy:channel:*:clients:*", count=1000) + for key in redis_client.scan_iter(match="ts_proxy:channel:*:clients:*", count=1000): + parts = key.split(':') + if len(parts) >= 5: + channel_id = parts[2] + client_id = parts[4] - for key in keys: - parts = key.split(':') - if len(parts) >= 5: - channel_id = parts[2] - client_id = parts[4] + client_user_id = redis_client.hget(key, 'user_id') + connected_at = redis_client.hget(key, 'connected_at') - client_user_id = redis_client.hget(key, 'user_id') - connected_at = redis_client.hget(key, 'connected_at') + logger.info(f"[stream limits] user_id = {user_id}") + logger.info(f"[stream limits] channel_id = {channel_id}") + logger.info(f"[stream limits] client_id = {client_id}") - logger.info(f"[stream limits] user_id = {user_id}") - logger.info(f"[stream limits] channel_id = {channel_id}") - logger.info(f"[stream limits] client_id = {client_id}") - - if client_user_id and int(client_user_id) == user_id: - try: - logger.info(f"[stream limits] Found LIVE connection for user {user_id} on channel {channel_id} with client ID {client_id}") - connected_at = float(connected_at) if connected_at else 0 - connections.append({ - 'media_id': channel_id, - 'client_id': client_id, - 'connected_at': connected_at, - 'type': 'live', - }) - except (ValueError, TypeError): - pass - - if cursor == 0: - break - - - cursor = 0 + if client_user_id and int(client_user_id) == user_id: + try: + logger.info(f"[stream limits] Found LIVE connection for user {user_id} on channel {channel_id} with client ID {client_id}") + connected_at = float(connected_at) if connected_at else 0 + connections.append({ + 'media_id': channel_id, + 'client_id': client_id, + 'connected_at': connected_at, + 'type': 'live', + }) + except (ValueError, TypeError): + pass # Grab VOD - while True: - cursor, keys = redis_client.scan(cursor=cursor, match="vod_persistent_connections:*", count=1000) + for key in redis_client.scan_iter(match="vod_persistent_connection:*", count=1000): + parts = key.split(':') + if len(parts) >= 2: + client_id = parts[1] - for key in keys: - parts = key.split(':') - if len(parts) >= 2: - client_id = parts[1] + client_user_id = redis_client.hget(key, 'user_id') + connected_at = redis_client.hget(key, 'created_at') + content_uuid = redis_client.hget(key, 'content_uuid') - client_user_id = redis_client.hget(key, 'user_id') - connected_at = redis_client.hget(key, 'created_at') + logger.info(f"[stream limits] user_id = {user_id}") + logger.info(f"[stream limits] client_id = {client_id}") - logger.info(f"[stream limits] user_id = {user_id}") - logger.info(f"[stream limits] client_id = {client_id}") - - if client_user_id and int(client_user_id) == user_id: - try: - logger.info(f"[stream limits] Found VOD connection for user {user_id} on channel {channel_id} with client ID {client_id}") - connected_at = float(connected_at) if connected_at else 0 - connections.append({ - 'media_id': channel_id, - 'client_id': client_id, - 'connected_at': connected_at, - 'type': 'vod', - }) - except (ValueError, TypeError): - pass - - if cursor == 0: - break + if client_user_id and int(client_user_id) == user_id: + try: + logger.info(f"[stream limits] Found VOD connection for user {user_id} on content {content_uuid} with client ID {client_id}") + connected_at = float(connected_at) if connected_at else 0 + connections.append({ + 'media_id': content_uuid or client_id, + 'client_id': client_id, + 'connected_at': connected_at, + 'type': 'vod', + }) + except (ValueError, TypeError): + pass return connections @@ -159,14 +143,12 @@ def check_user_stream_limits(user, client_id): # Check user stream limits if user and user.stream_limit > 0: logger.info("[stream limits]" f"[{client_id}] User {user.username} (ID: {user.id}) is requesting a stream (stream_limit: {user.stream_limit})") - user_limit_settings = CoreSettings.get_user_limit_settings() + user_limit_settings = CoreSettings.get_user_limits_settings() active_connections = get_user_active_connections(user.id) unique_channel_count = set([conn['media_id'] for conn in active_connections]) user_stream_count = len(unique_channel_count) if user_limit_settings.get("ignore_same_channel_connections", False) else len(active_connections) - print(active_connections) - logger.info(f"[stream limits]" f"[{client_id}] User {user.username} currently has {len(active_connections)} active connections across {len(unique_channel_count)} unique channels (counting method: {'unique channels' if user_limit_settings.get('ignore_same_channel_connections', False) else 'total connections'})") if user.stream_limit > 0 and user_stream_count >= user.stream_limit: diff --git a/core/utils.py b/core/utils.py index 4d65aa91..583075c3 100644 --- a/core/utils.py +++ b/core/utils.py @@ -57,6 +57,8 @@ class RedisClient: redis_host = os.environ.get("REDIS_HOST", getattr(settings, 'REDIS_HOST', 'localhost')) redis_port = int(os.environ.get("REDIS_PORT", getattr(settings, 'REDIS_PORT', 6379))) redis_db = int(os.environ.get("REDIS_DB", getattr(settings, 'REDIS_DB', 0))) + redis_password = os.environ.get("REDIS_PASSWORD", getattr(settings, 'REDIS_PASSWORD', '')) + redis_user = os.environ.get("REDIS_USER", getattr(settings, 'REDIS_USER', '')) # Use standardized settings socket_timeout = getattr(settings, 'REDIS_SOCKET_TIMEOUT', 5) @@ -65,17 +67,23 @@ class RedisClient: socket_keepalive = getattr(settings, 'REDIS_SOCKET_KEEPALIVE', True) retry_on_timeout = getattr(settings, 'REDIS_RETRY_ON_TIMEOUT', True) + # TLS params from settings (empty dict when TLS is disabled) + ssl_params = getattr(settings, 'REDIS_SSL_PARAMS', {}) + # Create Redis client with better defaults client = redis.Redis( host=redis_host, port=redis_port, db=redis_db, + password=redis_password if redis_password else None, + username=redis_user if redis_user else None, socket_timeout=socket_timeout, socket_connect_timeout=socket_connect_timeout, socket_keepalive=socket_keepalive, health_check_interval=health_check_interval, retry_on_timeout=retry_on_timeout, - decode_responses=decode_responses + decode_responses=decode_responses, + **ssl_params ) # Validate connection with ping @@ -87,84 +95,34 @@ class RedisClient: client.config_set('save', '') # Disable RDB snapshots client.config_set('appendonly', 'no') # Disable AOF logging - # Set optimal memory settings with environment variable support - # Get max memory from environment or use a larger default (512MB instead of 256MB) - #max_memory = os.environ.get('REDIS_MAX_MEMORY', '512mb') - #eviction_policy = os.environ.get('REDIS_EVICTION_POLICY', 'allkeys-lru') - - # TLS params from settings (empty dict when TLS is disabled) - ssl_params = getattr(settings, 'REDIS_SSL_PARAMS', {}) - - # Create Redis client with better defaults - client = redis.Redis( - host=redis_host, - port=redis_port, - db=redis_db, - password=redis_password if redis_password else None, - username=redis_user if redis_user else None, - socket_timeout=socket_timeout, - socket_connect_timeout=socket_connect_timeout, - socket_keepalive=socket_keepalive, - health_check_interval=health_check_interval, - retry_on_timeout=retry_on_timeout, - decode_responses=decode_responses, - **ssl_params - ) - - #logger.info(f"Redis configured with maxmemory={max_memory}, policy={eviction_policy}") - # Disable protected mode when in debug mode if os.environ.get('DISPATCHARR_DEBUG', '').lower() == 'true': client.config_set('protected-mode', 'no') # Disable protected mode in debug logger.warning("Redis protected mode disabled for debug environment") - # Set optimal memory settings with environment variable support - # Get max memory from environment or use a larger default (512MB instead of 256MB) - #max_memory = os.environ.get('REDIS_MAX_MEMORY', '512mb') - #eviction_policy = os.environ.get('REDIS_EVICTION_POLICY', 'allkeys-lru') - - # Apply memory settings - #client.config_set('maxmemory-policy', eviction_policy) - #client.config_set('maxmemory', max_memory) - - #logger.info(f"Redis configured with maxmemory={max_memory}, policy={eviction_policy}") - - # Disable protected mode when in debug mode - if os.environ.get('DISPATCHARR_DEBUG', '').lower() == 'true': - client.config_set('protected-mode', 'no') # Disable protected mode in debug - logger.warning("Redis protected mode disabled for debug environment") - - logger.trace("Redis persistence disabled for better performance") - except redis.exceptions.ResponseError as e: - # Improve error handling for Redis configuration errors - if "OOM" in str(e): - logger.error(f"Redis OOM during configuration: {e}") - # Try to increase maxmemory as an emergency measure - try: - client.config_set('maxmemory', '768mb') - logger.warning("Applied emergency Redis memory increase to 768MB") - except: - pass - else: - logger.error(f"Redis configuration error: {e}") - - logger.info(f"Connected to Redis at {redis_host}:{redis_port}/{redis_db}") - - cls._client = client - break - - except (ConnectionError, TimeoutError) as e: - retry_count += 1 - _tls_hint = _REDIS_TLS_HINT if ssl_params else "" - if retry_count >= max_retries: - logger.error(f"Failed to connect to Redis after {max_retries} attempts: {e}{_tls_hint}") - return None + logger.trace("Redis persistence disabled for better performance") + except redis.exceptions.ResponseError as e: + # Improve error handling for Redis configuration errors + if "OOM" in str(e): + logger.error(f"Redis OOM during configuration: {e}") + # Try to increase maxmemory as an emergency measure + try: + client.config_set('maxmemory', '768mb') + logger.warning("Applied emergency Redis memory increase to 768MB") + except: + pass else: logger.error(f"Redis configuration error: {e}") - except Exception as e: - _tls_hint = _REDIS_TLS_HINT if ssl_params else "" - logger.error(f"Unexpected error connecting to Redis: {e}{_tls_hint}") + logger.info(f"Connected to Redis at {redis_host}:{redis_port}/{redis_db}") + + return client + + except (ConnectionError, TimeoutError) as e: + retry_count += 1 + _tls_hint = _REDIS_TLS_HINT if ssl_params else "" + if retry_count >= max_retries: + logger.error(f"Failed to connect to Redis after {max_retries} attempts: {e}{_tls_hint}") return None else: # Use exponential backoff for retries @@ -173,10 +131,15 @@ class RedisClient: time.sleep(wait_time) except Exception as e: - logger.error(f"Unexpected error connecting to Redis: {e}") + _tls_hint = "" + try: + _tls_hint = _REDIS_TLS_HINT if ssl_params else "" + except NameError: + pass + logger.error(f"Unexpected error connecting to Redis: {e}{_tls_hint}") return None - return client + return None @classmethod def get_client(cls, max_retries=5, retry_interval=1): diff --git a/debug_redis_check.py b/debug_redis_check.py new file mode 100644 index 00000000..9deab6d4 --- /dev/null +++ b/debug_redis_check.py @@ -0,0 +1,21 @@ +#!/usr/bin/env python +import os, sys +os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'dispatcharr.settings') +import django +django.setup() + +from core.utils import RedisClient + +r = RedisClient.get_client() +print(f"Client: {r}") +print(f"decode_responses: {r.connection_pool.connection_kwargs.get('decode_responses')}") + +keys = list(r.scan_iter(match='ts_proxy:channel:*:metadata', count=100)) +print(f"Found {len(keys)} metadata keys") +for k in keys[:3]: + print(f" Key: {k!r}") + data = r.hgetall(k) + field_names = list(data.keys())[:8] + print(f" Fields ({len(data)} total): {field_names}") + state = data.get('state', 'MISSING') + print(f" state={state!r}")