Mirrors/Dispatcharr
1
0
Fork 1
mirror of https://github.com/Dispatcharr/Dispatcharr.git synced 2026-07-20 01:59:50 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

Merge pull request #42 from SergeantPanda/proxy-stats

Browse source 
Proxy stats
This commit is contained in:
SergeantPanda 2025-03-27 08:42:18 -05:00 • committed by GitHub
commit cc23ee9fbf
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 126 additions and 78 deletions
Download patch file Download diff file Expand all files Collapse all files

44
apps/proxy/ts_proxy/channel_status.py
View file

@ -3,7 +3,7 @@ import time
import re
from . import proxy_server
from .redis_keys import RedisKeys
from .constants import TS_PACKET_SIZE
from .constants import TS_PACKET_SIZE, ChannelMetadataField
from redis.exceptions import ConnectionError, TimeoutError
from .utils import get_logger
@ -35,28 +35,32 @@ class ChannelStatus:
info = {
'channel_id': channel_id,
'state': metadata.get(b'state', b'unknown').decode('utf-8'),
'url': metadata.get(b'url', b'').decode('utf-8'),
'profile': metadata.get(b'profile', b'unknown').decode('utf-8'),
'started_at': metadata.get(b'init_time', b'0').decode('utf-8'),
'owner': metadata.get(b'owner', b'unknown').decode('utf-8'),
'state': metadata.get(ChannelMetadataField.STATE.encode('utf-8'), b'unknown').decode('utf-8'),
'url': metadata.get(ChannelMetadataField.URL.encode('utf-8'), b'').decode('utf-8'),
'profile': metadata.get(ChannelMetadataField.STREAM_PROFILE.encode('utf-8'),
metadata.get(ChannelMetadataField.PROFILE.encode('utf-8'), b'unknown')).decode('utf-8'),
'started_at': metadata.get(ChannelMetadataField.INIT_TIME.encode('utf-8'), b'0').decode('utf-8'),
'owner': metadata.get(ChannelMetadataField.OWNER.encode('utf-8'), b'unknown').decode('utf-8'),
'buffer_index': int(buffer_index_value.decode('utf-8')) if buffer_index_value else 0,
}
# Add timing information
if b'state_changed_at' in metadata:
state_changed_at = float(metadata[b'state_changed_at'].decode('utf-8'))
state_changed_field = ChannelMetadataField.STATE_CHANGED_AT.encode('utf-8')
if state_changed_field in metadata:
state_changed_at = float(metadata[state_changed_field].decode('utf-8'))
info['state_changed_at'] = state_changed_at
info['state_duration'] = time.time() - state_changed_at
if b'init_time' in metadata:
created_at = float(metadata[b'init_time'].decode('utf-8'))
init_time_field = ChannelMetadataField.INIT_TIME.encode('utf-8')
if init_time_field in metadata:
created_at = float(metadata[init_time_field].decode('utf-8'))
info['started_at'] = created_at
info['uptime'] = time.time() - created_at
# Add data throughput information
if b'total_bytes' in metadata:
total_bytes = int(metadata[b'total_bytes'].decode('utf-8'))
total_bytes_field = ChannelMetadataField.TOTAL_BYTES.encode('utf-8')
if total_bytes_field in metadata:
total_bytes = int(metadata[total_bytes_field].decode('utf-8'))
info['total_bytes'] = total_bytes
# Format total bytes in human-readable form
@ -87,7 +91,7 @@ class ChannelStatus:
for client_id in client_ids:
client_id_str = client_id.decode('utf-8')
client_key = f"ts_proxy:channel:{channel_id}:clients:{client_id_str}"
client_key = RedisKeys.client_metadata(channel_id, client_id_str)
client_data = proxy_server.redis_client.hgetall(client_key)
if client_data:
@ -261,23 +265,23 @@ class ChannelStatus:
client_count = proxy_server.redis_client.scard(client_set_key) or 0
# Calculate uptime
created_at = float(metadata.get(b'init_time', b'0').decode('utf-8'))
created_at = float(metadata.get(ChannelMetadataField.INIT_TIME.encode('utf-8'), b'0').decode('utf-8'))
uptime = time.time() - created_at if created_at > 0 else 0
# Simplified info
info = {
'channel_id': channel_id,
'state': metadata.get(b'state', b'unknown').decode('utf-8'),
'url': metadata.get(b'url', b'').decode('utf-8'),
'profile': metadata.get(b'profile', b'unknown').decode('utf-8'),
'owner': metadata.get(b'owner', b'unknown').decode('utf-8'),
'state': metadata.get(ChannelMetadataField.STATE.encode('utf-8'), b'unknown').decode('utf-8'),
'url': metadata.get(ChannelMetadataField.URL.encode('utf-8'), b'').decode('utf-8'),
'profile': metadata.get(ChannelMetadataField.PROFILE.encode('utf-8'), b'unknown').decode('utf-8'),
'owner': metadata.get(ChannelMetadataField.OWNER.encode('utf-8'), b'unknown').decode('utf-8'),
'buffer_index': int(buffer_index_value.decode('utf-8')) if buffer_index_value else 0,
'client_count': client_count,
'uptime': uptime
}
# Add data throughput information to basic info
total_bytes_bytes = proxy_server.redis_client.hget(metadata_key, 'total_bytes')
total_bytes_bytes = proxy_server.redis_client.hget(metadata_key, ChannelMetadataField.TOTAL_BYTES.encode('utf-8'))
if total_bytes_bytes:
total_bytes = int(total_bytes_bytes.decode('utf-8'))
info['total_bytes'] = total_bytes
@ -307,7 +311,7 @@ class ChannelStatus:
# Get up to 10 clients for the basic view
for client_id in list(client_ids)[:10]:
client_id_str = client_id.decode('utf-8')
client_key = f"ts_proxy:channel:{channel_id}:clients:{client_id_str}"
client_key = RedisKeys.client_metadata(channel_id, client_id_str)
# Efficient way - just retrieve the essentials
client_info = {

39
apps/proxy/ts_proxy/constants.py
View file

@ -35,6 +35,45 @@ class StreamType:
TS = "ts"
UNKNOWN = "unknown"
# Channel metadata field names stored in Redis
class ChannelMetadataField:
# Basic fields
URL = "url"
USER_AGENT = "user_agent"
STATE = "state"
OWNER = "owner"
STREAM_ID = "stream_id"
# Profile fields
STREAM_PROFILE = "stream_profile"
M3U_PROFILE = "m3u_profile"
# Status and error fields
ERROR_MESSAGE = "error_message"
ERROR_TIME = "error_time"
STATE_CHANGED_AT = "state_changed_at"
INIT_TIME = "init_time"
CONNECTION_READY_TIME = "connection_ready_time"
# Buffer and data tracking
BUFFER_CHUNKS = "buffer_chunks"
TOTAL_BYTES = "total_bytes"
# Stream switching
STREAM_SWITCH_TIME = "stream_switch_time"
STREAM_SWITCH_REASON = "stream_switch_reason"
# Client metadata fields
CONNECTED_AT = "connected_at"
LAST_ACTIVE = "last_active"
BYTES_SENT = "bytes_sent"
AVG_RATE_KBPS = "avg_rate_KBps"
CURRENT_RATE_KBPS = "current_rate_KBps"
IP_ADDRESS = "ip_address"
WORKER_ID = "worker_id"
CHUNKS_SENT = "chunks_sent"
STATS_UPDATED_AT = "stats_updated_at"
# TS packet constants
TS_PACKET_SIZE = 188
TS_SYNC_BYTE = 0x47

40
apps/proxy/ts_proxy/services/channel_service.py
View file

@ -11,7 +11,7 @@ from apps.channels.models import Channel
from apps.proxy.config import TSConfig as Config
from .. import proxy_server
from ..redis_keys import RedisKeys
from ..constants import EventType, ChannelState
from ..constants import EventType, ChannelState, ChannelMetadataField
from ..url_utils import get_stream_info_for_switch
logger = logging.getLogger("ts_proxy")
@ -42,19 +42,19 @@ class ChannelService:
# Check if metadata already exists
if proxy_server.redis_client.exists(metadata_key):
# Just update the existing metadata with stream_id
proxy_server.redis_client.hset(metadata_key, "stream_id", str(stream_id))
proxy_server.redis_client.hset(metadata_key, ChannelMetadataField.STREAM_ID, str(stream_id))
logger.info(f"Pre-set stream ID {stream_id} in Redis for channel {channel_id}")
else:
# Create initial metadata with essential values
initial_metadata = {
"stream_id": str(stream_id),
ChannelMetadataField.STREAM_ID: str(stream_id),
"temp_init": str(time.time())
}
proxy_server.redis_client.hset(metadata_key, mapping=initial_metadata)
logger.info(f"Created initial metadata with stream_id {stream_id} for channel {channel_id}")
# Verify the stream_id was set
stream_id_value = proxy_server.redis_client.hget(metadata_key, "stream_id")
stream_id_value = proxy_server.redis_client.hget(metadata_key, ChannelMetadataField.STREAM_ID)
if stream_id_value:
logger.info(f"Verified stream_id {stream_id_value.decode('utf-8')} is now set in Redis")
else:
@ -68,9 +68,9 @@ class ChannelService:
metadata_key = RedisKeys.channel_metadata(channel_id)
update_data = {}
if profile_value:
update_data["profile"] = profile_value
update_data[ChannelMetadataField.STREAM_PROFILE] = profile_value
if stream_id:
update_data["stream_id"] = str(stream_id)
update_data[ChannelMetadataField.STREAM_ID] = str(stream_id)
if update_data:
proxy_server.redis_client.hset(metadata_key, mapping=update_data)
@ -220,8 +220,8 @@ class ChannelService:
channel_info = {"state": state}
# Immediately mark as stopping in metadata so clients detect it faster
proxy_server.redis_client.hset(metadata_key, "state", ChannelState.STOPPING)
proxy_server.redis_client.hset(metadata_key, "state_changed_at", str(time.time()))
proxy_server.redis_client.hset(metadata_key, ChannelMetadataField.STATE, ChannelState.STOPPING)
proxy_server.redis_client.hset(metadata_key, ChannelMetadataField.STATE_CHANGED_AT, str(time.time()))
except Exception as e:
logger.error(f"Error fetching channel state: {e}")
@ -350,8 +350,8 @@ class ChannelService:
metadata = proxy_server.redis_client.hgetall(metadata_key)
# Extract state and owner
state = metadata.get(b'state', b'unknown').decode('utf-8')
owner = metadata.get(b'owner', b'unknown').decode('utf-8')
state = metadata.get(ChannelMetadataField.STATE.encode(), b'unknown').decode('utf-8')
owner = metadata.get(ChannelMetadataField.OWNER.encode(), b'unknown').decode('utf-8')
# Valid states indicate channel is running properly
valid_states = [ChannelState.ACTIVE, ChannelState.WAITING_FOR_CLIENTS, ChannelState.CONNECTING]
@ -360,7 +360,7 @@ class ChannelService:
return False, state, owner, {"error": f"Invalid state: {state}"}
# Check if owner is still active
owner_heartbeat_key = f"ts_proxy:worker:{owner}:heartbeat"
owner_heartbeat_key = RedisKeys.worker_heartbeat(owner)
owner_alive = proxy_server.redis_client.exists(owner_heartbeat_key)
if not owner_alive:
@ -407,21 +407,21 @@ class ChannelService:
# Use the appropriate method based on the key type
if key_type == 'hash':
proxy_server.redis_client.hset(metadata_key, "url", url)
proxy_server.redis_client.hset(metadata_key, ChannelMetadataField.URL, url)
if user_agent:
proxy_server.redis_client.hset(metadata_key, "user_agent", user_agent)
proxy_server.redis_client.hset(metadata_key, ChannelMetadataField.USER_AGENT, user_agent)
elif key_type == 'none': # Key doesn't exist yet
# Create new hash with all required fields
metadata = {"url": url}
metadata = {ChannelMetadataField.URL: url}
if user_agent:
metadata["user_agent"] = user_agent
metadata[ChannelMetadataField.USER_AGENT] = user_agent
proxy_server.redis_client.hset(metadata_key, mapping=metadata)
else:
# If key exists with wrong type, delete it and recreate
proxy_server.redis_client.delete(metadata_key)
metadata = {"url": url}
metadata = {ChannelMetadataField.URL: url}
if user_agent:
metadata["user_agent"] = user_agent
metadata[ChannelMetadataField.USER_AGENT] = user_agent
proxy_server.redis_client.hset(metadata_key, mapping=metadata)
# Set switch request flag to ensure all workers see it
@ -438,7 +438,7 @@ class ChannelService:
return False
switch_request = {
"event": EventType.STREAM_SWITCH, # Use constant instead of string
"event": EventType.STREAM_SWITCH,
"channel_id": channel_id,
"url": new_url,
"user_agent": user_agent,
@ -459,7 +459,7 @@ class ChannelService:
return False
stop_request = {
"event": EventType.CHANNEL_STOP, # Use constant instead of string
"event": EventType.CHANNEL_STOP,
"channel_id": channel_id,
"requester_worker_id": proxy_server.worker_id,
"timestamp": time.time()
@ -480,7 +480,7 @@ class ChannelService:
return False
stop_request = {
"event": EventType.CLIENT_STOP, # Use constant instead of string
"event": EventType.CLIENT_STOP,
"channel_id": channel_id,
"client_id": client_id,
"requester_worker_id": proxy_server.worker_id,

11
apps/proxy/ts_proxy/stream_generator.py
View file

@ -11,6 +11,7 @@ from . import proxy_server
from .utils import create_ts_packet, get_logger
from .redis_keys import RedisKeys
from .utils import get_logger
from .constants import ChannelMetadataField
logger = get_logger()
@ -298,11 +299,11 @@ class StreamGenerator:
try:
client_key = RedisKeys.client_metadata(self.channel_id, self.client_id)
stats = {
"chunks_sent": str(self.chunks_sent),
"bytes_sent": str(self.bytes_sent),
"avg_rate_KBps": str(round(avg_rate, 1)),
"current_rate_KBps": str(round(self.current_rate, 1)),
"stats_updated_at": str(current_time)
ChannelMetadataField.CHUNKS_SENT: str(self.chunks_sent),
ChannelMetadataField.BYTES_SENT: str(self.bytes_sent),
ChannelMetadataField.AVG_RATE_KBPS: str(round(avg_rate, 1)),
ChannelMetadataField.CURRENT_RATE_KBPS: str(round(self.current_rate, 1)),
ChannelMetadataField.STATS_UPDATED_AT: str(current_time)
}
proxy_server.redis_client.hset(client_key, mapping=stats)
# No need to set expiration as client heartbeat will refresh this key

43
apps/proxy/ts_proxy/stream_manager.py
View file

@ -15,7 +15,7 @@ from core.models import UserAgent, CoreSettings
from .stream_buffer import StreamBuffer
from .utils import detect_stream_type, get_logger
from .redis_keys import RedisKeys
from .constants import ChannelState, EventType, StreamType, TS_PACKET_SIZE
from .constants import ChannelState, EventType, StreamType, ChannelMetadataField, TS_PACKET_SIZE
from .config_helper import ConfigHelper
from .url_utils import get_alternate_streams, get_stream_info_for_switch
@ -284,10 +284,10 @@ class StreamManager:
# Update metadata to indicate error state
update_data = {
"state": ChannelState.ERROR,
"state_changed_at": str(time.time()),
"error_message": error_message,
"error_time": str(time.time())
ChannelMetadataField.STATE: ChannelState.ERROR,
ChannelMetadataField.STATE_CHANGED_AT: str(time.time()),
ChannelMetadataField.ERROR_MESSAGE: error_message,
ChannelMetadataField.ERROR_TIME: str(time.time())
}
self.buffer.redis_client.hset(metadata_key, mapping=update_data)
logger.info(f"Updated channel {self.channel_id} state to ERROR in Redis after stream failure")
@ -398,13 +398,13 @@ class StreamManager:
metadata_key = RedisKeys.channel_metadata(self.channel_id)
# Use hincrby to atomically increment the total_bytes field
self.buffer.redis_client.hincrby(metadata_key, "total_bytes", self.bytes_processed)
self.buffer.redis_client.hincrby(metadata_key, ChannelMetadataField.TOTAL_BYTES, self.bytes_processed)
# Reset local counter after updating Redis
self.bytes_processed = 0
self.last_bytes_update = now
logger.debug(f"Updated total_bytes in Redis for channel {self.channel_id}")
logger.debug(f"Updated {ChannelMetadataField.TOTAL_BYTES} in Redis for channel {self.channel_id}")
except Exception as e:
logger.error(f"Error updating bytes processed: {e}")
@ -743,8 +743,9 @@ class StreamManager:
current_state = None
try:
metadata = redis_client.hgetall(metadata_key)
if metadata and b'state' in metadata:
current_state = metadata[b'state'].decode('utf-8')
state_field = ChannelMetadataField.STATE.encode('utf-8')
if metadata and state_field in metadata:
current_state = metadata[state_field].decode('utf-8')
except Exception as e:
logger.error(f"Error checking current state: {e}")
@ -758,8 +759,8 @@ class StreamManager:
# Not enough buffer yet - set to connecting state if not already
if current_state != ChannelState.CONNECTING:
update_data = {
"state": ChannelState.CONNECTING,
"state_changed_at": current_time
ChannelMetadataField.STATE: ChannelState.CONNECTING,
ChannelMetadataField.STATE_CHANGED_AT: current_time
}
redis_client.hset(metadata_key, mapping=update_data)
logger.info(f"Channel {channel_id} connected but waiting for buffer to fill: {current_buffer_index}/{initial_chunks_needed} chunks")
@ -772,10 +773,10 @@ class StreamManager:
# We have enough buffer, proceed with state change
update_data = {
"state": ChannelState.WAITING_FOR_CLIENTS,
"connection_ready_time": current_time,
"state_changed_at": current_time,
"buffer_chunks": str(current_buffer_index)
ChannelMetadataField.STATE: ChannelState.WAITING_FOR_CLIENTS,
ChannelMetadataField.CONNECTION_READY_TIME: current_time,
ChannelMetadataField.STATE_CHANGED_AT: current_time,
ChannelMetadataField.BUFFER_CHUNKS: str(current_buffer_index)
}
redis_client.hset(metadata_key, mapping=update_data)
@ -885,12 +886,12 @@ class StreamManager:
if hasattr(self.buffer, 'redis_client') and self.buffer.redis_client:
metadata_key = RedisKeys.channel_metadata(self.channel_id)
self.buffer.redis_client.hset(metadata_key, mapping={
"url": new_url,
"user_agent": new_user_agent,
"profile": stream_info['profile'],
"stream_id": str(stream_id),
"stream_switch_time": str(time.time()),
"stream_switch_reason": "max_retries_exceeded"
ChannelMetadataField.URL: new_url,
ChannelMetadataField.USER_AGENT: new_user_agent,
ChannelMetadataField.STREAM_PROFILE: stream_info['profile'],
ChannelMetadataField.STREAM_ID: str(stream_id),
ChannelMetadataField.STREAM_SWITCH_TIME: str(time.time()),
ChannelMetadataField.STREAM_SWITCH_REASON: "max_retries_exceeded"
})
# Log the switch

9
apps/proxy/ts_proxy/url_utils.py
View file

@ -33,10 +33,10 @@ def generate_stream_url(channel_id: str) -> Tuple[str, str, bool]:
# Get the M3U account profile for URL pattern
stream = get_object_or_404(Stream, pk=stream_id)
profile = get_object_or_404(M3UAccountProfile, pk=profile_id)
m3u_profile = get_object_or_404(M3UAccountProfile, pk=profile_id)
# Get the appropriate user agent
m3u_account = M3UAccount.objects.get(id=profile.m3u_account.id)
m3u_account = M3UAccount.objects.get(id=m3u_profile.m3u_account.id)
stream_user_agent = UserAgent.objects.get(id=m3u_account.user_agent.id).user_agent
if stream_user_agent is None:
@ -45,7 +45,7 @@ def generate_stream_url(channel_id: str) -> Tuple[str, str, bool]:
# Generate stream URL based on the selected profile
input_url = stream.url
stream_url = transform_url(input_url, profile.search_pattern, profile.replace_pattern)
stream_url = transform_url(input_url, m3u_profile.search_pattern, m3u_profile.replace_pattern)
# Check if transcoding is needed
stream_profile = channel.get_stream_profile()
@ -54,7 +54,8 @@ def generate_stream_url(channel_id: str) -> Tuple[str, str, bool]:
else:
transcode = True
# Get profile name as string
# Get profile name as string - use id for backward compatibility
# but we'll store it in the STREAM_PROFILE field
profile_value = stream_profile.id
return stream_url, stream_user_agent, transcode, profile_value

18
apps/proxy/ts_proxy/views.py
View file

@ -18,7 +18,7 @@ from apps.m3u.models import M3UAccount, M3UAccountProfile
from core.models import UserAgent, CoreSettings, PROXY_PROFILE_NAME
from rest_framework.decorators import api_view, permission_classes
from rest_framework.permissions import IsAuthenticated
from .constants import ChannelState, EventType, StreamType
from .constants import ChannelState, EventType, StreamType, ChannelMetadataField
from .config_helper import ConfigHelper
from .services.channel_service import ChannelService
from .url_utils import generate_stream_url, transform_url, get_stream_info_for_switch
@ -56,15 +56,17 @@ def stream_ts(request, channel_id):
metadata_key = RedisKeys.channel_metadata(channel_id)
if proxy_server.redis_client.exists(metadata_key):
metadata = proxy_server.redis_client.hgetall(metadata_key)
if b'state' in metadata:
channel_state = metadata[b'state'].decode('utf-8')
state_field = ChannelMetadataField.STATE.encode('utf-8')
if state_field in metadata:
channel_state = metadata[state_field].decode('utf-8')
# Only skip initialization if channel is in a healthy state
valid_states = [ChannelState.ACTIVE, ChannelState.WAITING_FOR_CLIENTS]
if channel_state in valid_states:
# Verify the owner is still active
if b'owner' in metadata:
owner = metadata[b'owner'].decode('utf-8')
owner_field = ChannelMetadataField.OWNER.encode('utf-8')
if owner_field in metadata:
owner = metadata[owner_field].decode('utf-8')
owner_heartbeat_key = f"ts_proxy:worker:{owner}:heartbeat"
if proxy_server.redis_client.exists(owner_heartbeat_key):
# Owner is active and channel is in good state
@ -134,9 +136,9 @@ def stream_ts(request, channel_id):
if proxy_server.redis_client:
metadata_key = RedisKeys.channel_metadata(channel_id)
url_bytes = proxy_server.redis_client.hget(metadata_key, "url")
ua_bytes = proxy_server.redis_client.hget(metadata_key, "user_agent")
profile_bytes = proxy_server.redis_client.hget(metadata_key, "profile")
url_bytes = proxy_server.redis_client.hget(metadata_key, ChannelMetadataField.URL)
ua_bytes = proxy_server.redis_client.hget(metadata_key, ChannelMetadataField.USER_AGENT)
profile_bytes = proxy_server.redis_client.hget(metadata_key, ChannelMetadataField.STREAM_PROFILE)
if url_bytes:
url = url_bytes.decode('utf-8')