mirror of
https://github.com/Dispatcharr/Dispatcharr.git
synced 2026-07-18 00:55:50 +00:00
4fc12bca4a
- Improved logging in the Stream model for better debugging of profile evaluations. - Introduced a new method `_stream_assignment_is_reusable` in the Channel model to determine if existing stream assignments can be reused, enhancing efficiency. - Updated the release logic in `release_profile_slot` to utilize stored credential keys, reducing unnecessary database lookups. - Simplified error handling in the `get_stream_info_for_switch` function to ensure proper stream release on exceptions. - Enhanced tests for connection pool management and error handling in the ServerGroupsTable component to improve reliability and user feedback.
540 lines
22 KiB
Python
540 lines
22 KiB
Python
import time
|
|
from .server import ProxyServer
|
|
from .redis_keys import RedisKeys
|
|
from .constants import TS_PACKET_SIZE, ChannelMetadataField
|
|
from redis.exceptions import ConnectionError, TimeoutError
|
|
from .utils import get_logger
|
|
from .client_manager import ClientManager
|
|
from django.db import DatabaseError
|
|
|
|
logger = get_logger()
|
|
|
|
class ChannelStatus:
|
|
|
|
@staticmethod
|
|
def _calculate_bitrate(total_bytes, duration):
|
|
"""Calculate bitrate in Kbps based on total bytes and duration in seconds"""
|
|
if duration <= 0:
|
|
return 0
|
|
|
|
# Convert bytes to bits (x8) and divide by duration to get bits per second
|
|
# Then divide by 1000 to get Kbps
|
|
return (total_bytes * 8) / duration / 1000
|
|
|
|
def get_detailed_channel_info(channel_id):
|
|
proxy_server = ProxyServer.get_instance()
|
|
|
|
# Get channel metadata
|
|
metadata_key = RedisKeys.channel_metadata(channel_id)
|
|
metadata = proxy_server.redis_client.hgetall(metadata_key)
|
|
|
|
if not metadata:
|
|
return None
|
|
|
|
# Basic channel info
|
|
buffer_index_key = RedisKeys.buffer_index(channel_id)
|
|
buffer_index_value = proxy_server.redis_client.get(buffer_index_key)
|
|
|
|
info = {
|
|
'channel_id': channel_id,
|
|
'state': metadata.get(ChannelMetadataField.STATE, 'unknown'),
|
|
'url': metadata.get(ChannelMetadataField.URL, ''),
|
|
'stream_profile': metadata.get(ChannelMetadataField.STREAM_PROFILE, ''),
|
|
'started_at': metadata.get(ChannelMetadataField.INIT_TIME, '0'),
|
|
'owner': metadata.get(ChannelMetadataField.OWNER, 'unknown'),
|
|
'buffer_index': int(buffer_index_value) if buffer_index_value else 0,
|
|
}
|
|
|
|
# Add stream ID and name information
|
|
stream_id_bytes = metadata.get(ChannelMetadataField.STREAM_ID)
|
|
if stream_id_bytes:
|
|
try:
|
|
stream_id = int(stream_id_bytes)
|
|
info['stream_id'] = stream_id
|
|
|
|
# Look up stream name from database
|
|
try:
|
|
from apps.channels.models import Stream
|
|
stream = Stream.objects.filter(id=stream_id).first()
|
|
if stream:
|
|
info['stream_name'] = stream.name
|
|
except (ImportError, DatabaseError) as e:
|
|
logger.warning(f"Failed to get stream name for ID {stream_id}: {e}")
|
|
except ValueError:
|
|
logger.warning(f"Invalid stream_id format in Redis: {stream_id_bytes}")
|
|
|
|
# Add M3U profile information
|
|
m3u_profile_id_bytes = metadata.get(ChannelMetadataField.M3U_PROFILE)
|
|
if m3u_profile_id_bytes:
|
|
try:
|
|
m3u_profile_id = int(m3u_profile_id_bytes)
|
|
info['m3u_profile_id'] = m3u_profile_id
|
|
|
|
# Look up M3U profile name from database
|
|
try:
|
|
from apps.m3u.models import M3UAccountProfile
|
|
m3u_profile = M3UAccountProfile.objects.filter(id=m3u_profile_id).first()
|
|
if m3u_profile:
|
|
info['m3u_profile_name'] = m3u_profile.name
|
|
except (ImportError, DatabaseError) as e:
|
|
logger.warning(f"Failed to get M3U profile name for ID {m3u_profile_id}: {e}")
|
|
except ValueError:
|
|
logger.warning(f"Invalid m3u_profile_id format in Redis: {m3u_profile_id_bytes}")
|
|
|
|
# Add timing information
|
|
state_changed_field = ChannelMetadataField.STATE_CHANGED_AT
|
|
if state_changed_field in metadata:
|
|
state_changed_at = float(metadata[state_changed_field])
|
|
info['state_changed_at'] = state_changed_at
|
|
info['state_duration'] = time.time() - state_changed_at
|
|
|
|
init_time_field = ChannelMetadataField.INIT_TIME
|
|
if init_time_field in metadata:
|
|
created_at = float(metadata[init_time_field])
|
|
info['started_at'] = created_at
|
|
info['uptime'] = time.time() - created_at
|
|
|
|
# Add data throughput information
|
|
total_bytes_field = ChannelMetadataField.TOTAL_BYTES
|
|
if total_bytes_field in metadata:
|
|
total_bytes = int(metadata[total_bytes_field])
|
|
info['total_bytes'] = total_bytes
|
|
|
|
# Format total bytes in human-readable form
|
|
if total_bytes < 1024:
|
|
info['total_data'] = f"{total_bytes} B"
|
|
elif total_bytes < 1024 * 1024:
|
|
info['total_data'] = f"{total_bytes / 1024:.2f} KB"
|
|
elif total_bytes < 1024 * 1024 * 1024:
|
|
info['total_data'] = f"{total_bytes / (1024 * 1024):.2f} MB"
|
|
else:
|
|
info['total_data'] = f"{total_bytes / (1024 * 1024 * 1024):.2f} GB"
|
|
|
|
# Calculate average bitrate if we have uptime
|
|
if 'uptime' in info and info['uptime'] > 0:
|
|
avg_bitrate = ChannelStatus._calculate_bitrate(total_bytes, info['uptime'])
|
|
info['avg_bitrate_kbps'] = avg_bitrate
|
|
|
|
# Format in Mbps if over 1000 Kbps
|
|
if avg_bitrate > 1000:
|
|
info['avg_bitrate'] = f"{avg_bitrate / 1000:.2f} Mbps"
|
|
else:
|
|
info['avg_bitrate'] = f"{avg_bitrate:.2f} Kbps"
|
|
|
|
# Get client information
|
|
client_set_key = RedisKeys.clients(channel_id)
|
|
client_ids = proxy_server.redis_client.smembers(client_set_key)
|
|
clients = []
|
|
|
|
stale_client_ids = []
|
|
for client_id in client_ids:
|
|
client_id_str = client_id
|
|
client_key = RedisKeys.client_metadata(channel_id, client_id_str)
|
|
client_data = proxy_server.redis_client.hgetall(client_key)
|
|
|
|
if not client_data:
|
|
# Metadata hash expired but SET entry persists (ghost client).
|
|
stale_client_ids.append(client_id)
|
|
continue
|
|
|
|
client_info = {
|
|
'client_id': client_id_str,
|
|
'user_agent': client_data.get('user_agent', 'unknown'),
|
|
'worker_id': client_data.get('worker_id', 'unknown'),
|
|
'ip_address': client_data.get('ip_address', 'unknown'),
|
|
'user_id': client_data.get('user_id', '0'),
|
|
'output_format': client_data.get('output_format', 'mpegts'),
|
|
}
|
|
|
|
raw_profile_id = client_data.get('output_profile_id')
|
|
if raw_profile_id and raw_profile_id not in ('None', '0', ''):
|
|
client_info['output_profile_id'] = int(raw_profile_id)
|
|
else:
|
|
client_info['output_profile_id'] = None
|
|
|
|
if 'connected_at' in client_data:
|
|
client_info['connected_at'] = float(client_data['connected_at'])
|
|
|
|
if 'last_active' in client_data:
|
|
last_active = float(client_data['last_active'])
|
|
client_info['last_active'] = last_active
|
|
client_info['last_active_ago'] = time.time() - last_active
|
|
|
|
# Add transfer rate statistics
|
|
if 'bytes_sent' in client_data:
|
|
client_info['bytes_sent'] = int(client_data['bytes_sent'])
|
|
|
|
# Add average transfer rate
|
|
if 'avg_rate_KBps' in client_data:
|
|
client_info['avg_rate_KBps'] = float(client_data['avg_rate_KBps'])
|
|
elif 'transfer_rate_KBps' in client_data: # For backward compatibility
|
|
client_info['avg_rate_KBps'] = float(client_data['transfer_rate_KBps'])
|
|
|
|
# Add current transfer rate
|
|
if 'current_rate_KBps' in client_data:
|
|
client_info['current_rate_KBps'] = float(client_data['current_rate_KBps'])
|
|
|
|
clients.append(client_info)
|
|
|
|
# Clean up stale SET entries so SCARD stays accurate.
|
|
if stale_client_ids:
|
|
proxy_server.redis_client.srem(client_set_key, *stale_client_ids)
|
|
logger.info(
|
|
f"Removed {len(stale_client_ids)} ghost client(s) from "
|
|
f"channel {channel_id} client set"
|
|
)
|
|
|
|
info['clients'] = clients
|
|
info['client_count'] = len(clients)
|
|
|
|
# Get buffer health with improved diagnostics
|
|
buffer_stats = {
|
|
'chunks': info['buffer_index'],
|
|
'diagnostics': {}
|
|
}
|
|
|
|
# Sample a few recent chunks to check sizes with better error handling
|
|
if info['buffer_index'] > 0:
|
|
try:
|
|
sample_chunks = min(5, info['buffer_index'])
|
|
chunk_sizes = []
|
|
chunk_keys_found = []
|
|
chunk_keys_missing = []
|
|
|
|
# Check if the keys exist before getting
|
|
for i in range(info['buffer_index']-sample_chunks+1, info['buffer_index']+1):
|
|
chunk_key = RedisKeys.buffer_chunk(channel_id, i)
|
|
|
|
# Check if key exists first
|
|
if proxy_server.redis_client.exists(chunk_key):
|
|
chunk_data = proxy_server.redis_client.get(chunk_key)
|
|
if chunk_data:
|
|
chunk_size = len(chunk_data)
|
|
chunk_sizes.append(chunk_size)
|
|
chunk_keys_found.append(i)
|
|
|
|
# Check for TS alignment (packets are 188 bytes)
|
|
ts_packets = chunk_size // 188
|
|
ts_aligned = chunk_size % 188 == 0
|
|
|
|
# Add for first chunk only to avoid too much data
|
|
if len(chunk_keys_found) == 1:
|
|
buffer_stats['diagnostics']['first_chunk'] = {
|
|
'index': i,
|
|
'size': chunk_size,
|
|
'ts_packets': ts_packets,
|
|
'aligned': ts_aligned,
|
|
'first_byte': chunk_data[0] if chunk_size > 0 else None
|
|
}
|
|
else:
|
|
chunk_keys_missing.append(i)
|
|
|
|
# Add detailed diagnostics
|
|
if chunk_sizes:
|
|
buffer_stats['avg_chunk_size'] = sum(chunk_sizes) / len(chunk_sizes)
|
|
buffer_stats['recent_chunk_sizes'] = chunk_sizes
|
|
buffer_stats['keys_found'] = chunk_keys_found
|
|
buffer_stats['keys_missing'] = chunk_keys_missing
|
|
|
|
# Calculate data rate
|
|
total_data = sum(chunk_sizes)
|
|
buffer_stats['total_sample_bytes'] = total_data
|
|
|
|
# Add TS packet analysis
|
|
total_ts_packets = total_data // TS_PACKET_SIZE
|
|
buffer_stats['estimated_ts_packets'] = total_ts_packets
|
|
buffer_stats['is_ts_aligned'] = all(size % TS_PACKET_SIZE == 0 for size in chunk_sizes)
|
|
else:
|
|
# If no chunks found, scan for keys to help debug
|
|
all_buffer_keys = []
|
|
cursor = 0
|
|
|
|
buffer_key_pattern = f"live:channel:{channel_id}:input:buffer:chunk:*"
|
|
|
|
while True:
|
|
cursor, keys = proxy_server.redis_client.scan(cursor, match=buffer_key_pattern, count=100)
|
|
if keys:
|
|
all_buffer_keys.extend([k for k in keys])
|
|
if cursor == 0 or len(all_buffer_keys) >= 20: # Limit to 20 keys
|
|
break
|
|
|
|
buffer_stats['diagnostics']['all_buffer_keys'] = all_buffer_keys[:20] # First 20 keys
|
|
buffer_stats['diagnostics']['total_buffer_keys'] = len(all_buffer_keys)
|
|
|
|
except Exception as e:
|
|
# Capture any errors for diagnostics
|
|
buffer_stats['error'] = str(e)
|
|
buffer_stats['diagnostics']['exception'] = str(e)
|
|
|
|
# Add TTL information to see if chunks are expiring
|
|
chunk_ttl_key = RedisKeys.buffer_chunk(channel_id, info['buffer_index'])
|
|
chunk_ttl = proxy_server.redis_client.ttl(chunk_ttl_key)
|
|
buffer_stats['latest_chunk_ttl'] = chunk_ttl
|
|
|
|
info['buffer_stats'] = buffer_stats
|
|
|
|
# Get local worker info if available
|
|
if channel_id in proxy_server.stream_managers:
|
|
manager = proxy_server.stream_managers[channel_id]
|
|
info['local_manager'] = {
|
|
'healthy': manager.healthy,
|
|
'connected': manager.connected,
|
|
'last_data_time': manager.last_data_time,
|
|
'last_data_age': time.time() - manager.last_data_time
|
|
}
|
|
|
|
# Add FFmpeg stream information
|
|
video_codec = metadata.get(ChannelMetadataField.VIDEO_CODEC)
|
|
if video_codec:
|
|
info['video_codec'] = video_codec
|
|
|
|
resolution = metadata.get(ChannelMetadataField.RESOLUTION)
|
|
if resolution:
|
|
info['resolution'] = resolution
|
|
|
|
source_fps = metadata.get(ChannelMetadataField.SOURCE_FPS)
|
|
if source_fps:
|
|
info['source_fps'] = source_fps
|
|
|
|
pixel_format = metadata.get(ChannelMetadataField.PIXEL_FORMAT)
|
|
if pixel_format:
|
|
info['pixel_format'] = pixel_format
|
|
|
|
source_bitrate = metadata.get(ChannelMetadataField.SOURCE_BITRATE)
|
|
if source_bitrate:
|
|
info['source_bitrate'] = source_bitrate
|
|
|
|
audio_codec = metadata.get(ChannelMetadataField.AUDIO_CODEC)
|
|
if audio_codec:
|
|
info['audio_codec'] = audio_codec
|
|
|
|
sample_rate = metadata.get(ChannelMetadataField.SAMPLE_RATE)
|
|
if sample_rate:
|
|
info['sample_rate'] = sample_rate
|
|
|
|
audio_channels = metadata.get(ChannelMetadataField.AUDIO_CHANNELS)
|
|
if audio_channels:
|
|
info['audio_channels'] = audio_channels
|
|
|
|
audio_bitrate = metadata.get(ChannelMetadataField.AUDIO_BITRATE)
|
|
if audio_bitrate:
|
|
info['audio_bitrate'] = audio_bitrate
|
|
|
|
|
|
# Add FFmpeg performance stats
|
|
ffmpeg_speed = metadata.get(ChannelMetadataField.FFMPEG_SPEED)
|
|
if ffmpeg_speed:
|
|
info['ffmpeg_speed'] = ffmpeg_speed
|
|
|
|
ffmpeg_fps = metadata.get(ChannelMetadataField.FFMPEG_FPS)
|
|
if ffmpeg_fps:
|
|
info['ffmpeg_fps'] = ffmpeg_fps
|
|
|
|
actual_fps = metadata.get(ChannelMetadataField.ACTUAL_FPS)
|
|
if actual_fps:
|
|
info['actual_fps'] = actual_fps
|
|
|
|
ffmpeg_bitrate = metadata.get(ChannelMetadataField.FFMPEG_BITRATE)
|
|
if ffmpeg_bitrate:
|
|
info['ffmpeg_bitrate'] = ffmpeg_bitrate
|
|
|
|
stream_type = metadata.get(ChannelMetadataField.STREAM_TYPE)
|
|
if stream_type:
|
|
info['stream_type'] = stream_type
|
|
|
|
|
|
return info
|
|
|
|
@staticmethod
|
|
def _execute_redis_command(command_func):
|
|
"""Execute Redis command with error handling"""
|
|
proxy_server = ProxyServer.get_instance()
|
|
|
|
if not proxy_server.redis_client:
|
|
return None
|
|
|
|
try:
|
|
return command_func()
|
|
except (ConnectionError, TimeoutError) as e:
|
|
logger.warning(f"Redis connection error in ChannelStatus: {e}")
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"Redis command error in ChannelStatus: {e}")
|
|
return None
|
|
|
|
@staticmethod
|
|
def get_basic_channel_info(channel_id):
|
|
"""Get basic channel information with Redis error handling"""
|
|
proxy_server = ProxyServer.get_instance()
|
|
|
|
try:
|
|
# Use _execute_redis_command for Redis operations
|
|
metadata_key = RedisKeys.channel_metadata(channel_id)
|
|
metadata = ChannelStatus._execute_redis_command(
|
|
lambda: proxy_server.redis_client.hgetall(metadata_key)
|
|
)
|
|
|
|
if not metadata:
|
|
return None
|
|
|
|
# Basic channel info only - omit diagnostics and details
|
|
buffer_index_key = RedisKeys.buffer_index(channel_id)
|
|
buffer_index_value = proxy_server.redis_client.get(buffer_index_key)
|
|
|
|
# Count clients (using efficient count method)
|
|
client_set_key = RedisKeys.clients(channel_id)
|
|
client_count = proxy_server.redis_client.scard(client_set_key) or 0
|
|
|
|
# Calculate uptime
|
|
init_time_bytes = metadata.get(ChannelMetadataField.INIT_TIME, '0')
|
|
created_at = float(init_time_bytes)
|
|
uptime = time.time() - created_at if created_at > 0 else 0
|
|
|
|
# Simplified info
|
|
info = {
|
|
'channel_id': channel_id,
|
|
'state': metadata.get(ChannelMetadataField.STATE),
|
|
'url': metadata.get(ChannelMetadataField.URL, ""),
|
|
'stream_profile': metadata.get(ChannelMetadataField.STREAM_PROFILE, ""),
|
|
'owner': metadata.get(ChannelMetadataField.OWNER),
|
|
'buffer_index': int(buffer_index_value) if buffer_index_value else 0,
|
|
'client_count': client_count,
|
|
'uptime': uptime,
|
|
'started_at': created_at if created_at > 0 else None,
|
|
}
|
|
|
|
channel_name = metadata.get(ChannelMetadataField.CHANNEL_NAME)
|
|
if channel_name:
|
|
info['channel_name'] = channel_name
|
|
|
|
stream_id_bytes = metadata.get(ChannelMetadataField.STREAM_ID)
|
|
if stream_id_bytes:
|
|
try:
|
|
info['stream_id'] = int(stream_id_bytes)
|
|
except ValueError:
|
|
logger.warning(f"Invalid stream_id format in Redis: {stream_id_bytes}")
|
|
|
|
stream_name = metadata.get(ChannelMetadataField.STREAM_NAME)
|
|
if stream_name:
|
|
info['stream_name'] = stream_name
|
|
|
|
# Add data throughput information to basic info
|
|
# TOTAL_BYTES is already present in the hgetall result — avoid a redundant round-trip
|
|
total_bytes_bytes = metadata.get(ChannelMetadataField.TOTAL_BYTES)
|
|
if total_bytes_bytes:
|
|
total_bytes = int(total_bytes_bytes)
|
|
info['total_bytes'] = total_bytes
|
|
|
|
# Calculate and add bitrate
|
|
if uptime > 0:
|
|
avg_bitrate = ChannelStatus._calculate_bitrate(total_bytes, uptime)
|
|
info['avg_bitrate_kbps'] = avg_bitrate
|
|
|
|
# Format for display
|
|
if avg_bitrate > 1000:
|
|
info['avg_bitrate'] = f"{avg_bitrate / 1000:.2f} Mbps"
|
|
else:
|
|
info['avg_bitrate'] = f"{avg_bitrate:.2f} Kbps"
|
|
|
|
# Quick health check if available locally
|
|
if channel_id in proxy_server.stream_managers:
|
|
manager = proxy_server.stream_managers[channel_id]
|
|
info['healthy'] = manager.healthy
|
|
|
|
# Get concise client information
|
|
clients = []
|
|
client_ids = proxy_server.redis_client.smembers(client_set_key)
|
|
|
|
# Remove ghost SET entries before building the client list.
|
|
# Pass the already-fetched client_ids to avoid a redundant SMEMBERS.
|
|
stale_client_ids = ClientManager.remove_ghost_clients(
|
|
proxy_server.redis_client, channel_id, client_ids=client_ids
|
|
)
|
|
if stale_client_ids:
|
|
client_count = max(0, client_count - len(stale_client_ids))
|
|
|
|
# Build concise client list (up to 10) from remaining live clients.
|
|
if client_ids:
|
|
for client_id in list(client_ids)[:10]:
|
|
if client_id in stale_client_ids:
|
|
continue
|
|
|
|
client_key = RedisKeys.client_metadata(channel_id, client_id)
|
|
|
|
# Fetch only the fields we need in one round-trip (hmget returns a list
|
|
# in the same order as the requested keys; values are None if absent)
|
|
ua, ip, connected_at, user_id, output_format, raw_profile_id = (
|
|
proxy_server.redis_client.hmget(
|
|
client_key,
|
|
'user_agent', 'ip_address', 'connected_at', 'user_id',
|
|
'output_format', 'output_profile_id',
|
|
)
|
|
)
|
|
|
|
client_info = {
|
|
'client_id': client_id,
|
|
'user_agent': ua,
|
|
'output_format': output_format or 'mpegts',
|
|
}
|
|
|
|
if ip:
|
|
client_info['ip_address'] = ip
|
|
|
|
if connected_at:
|
|
client_info['connected_at'] = float(connected_at)
|
|
|
|
if user_id:
|
|
client_info['user_id'] = user_id
|
|
|
|
if raw_profile_id and raw_profile_id not in ('None', '0', ''):
|
|
client_info['output_profile_id'] = int(raw_profile_id)
|
|
else:
|
|
client_info['output_profile_id'] = None
|
|
|
|
clients.append(client_info)
|
|
|
|
# Add clients to info
|
|
info['clients'] = clients
|
|
info['client_count'] = client_count
|
|
|
|
# Add M3U profile ID from Redis metadata (name resolved on frontend from playlists store)
|
|
m3u_profile_id = metadata.get(ChannelMetadataField.M3U_PROFILE)
|
|
if m3u_profile_id:
|
|
try:
|
|
info['m3u_profile_id'] = int(m3u_profile_id)
|
|
except ValueError:
|
|
logger.warning(f"Invalid m3u_profile_id format in Redis: {m3u_profile_id}")
|
|
|
|
# Add stream info to basic info as well
|
|
video_codec = metadata.get(ChannelMetadataField.VIDEO_CODEC)
|
|
if video_codec:
|
|
info['video_codec'] = video_codec
|
|
|
|
resolution = metadata.get(ChannelMetadataField.RESOLUTION)
|
|
if resolution:
|
|
info['resolution'] = resolution
|
|
|
|
source_fps = metadata.get(ChannelMetadataField.SOURCE_FPS)
|
|
if source_fps:
|
|
info['source_fps'] = float(source_fps)
|
|
|
|
ffmpeg_speed = metadata.get(ChannelMetadataField.FFMPEG_SPEED)
|
|
if ffmpeg_speed:
|
|
info['ffmpeg_speed'] = float(ffmpeg_speed)
|
|
|
|
audio_codec = metadata.get(ChannelMetadataField.AUDIO_CODEC)
|
|
if audio_codec:
|
|
info['audio_codec'] = audio_codec
|
|
|
|
audio_channels = metadata.get(ChannelMetadataField.AUDIO_CHANNELS)
|
|
if audio_channels:
|
|
info['audio_channels'] = audio_channels
|
|
|
|
stream_type = metadata.get(ChannelMetadataField.STREAM_TYPE)
|
|
if stream_type:
|
|
info['stream_type'] = stream_type
|
|
|
|
return info
|
|
except Exception as e:
|
|
logger.error(f"Error getting channel info: {e}", exc_info=True)
|
|
return None
|