diff --git a/src/storage/__init__.py b/src/storage/__init__.py index e3b7ed3..475bfe4 100644 --- a/src/storage/__init__.py +++ b/src/storage/__init__.py @@ -11,7 +11,7 @@ Also provides service discovery for inter-service communication. from .provider import StorageProvider, VideoMetadata from .redis_storage import RedisStorage from .leader_storage import LeaderStorage -from .leader_discovery import LeaderDiscovery, LeaderLockInfo +from .leader_client import LeaderClient, LeaderLockInfo from .service_discovery import ( ServiceDiscovery, ServiceInfo, @@ -24,7 +24,7 @@ __all__ = [ 'VideoMetadata', 'RedisStorage', 'LeaderStorage', - 'LeaderDiscovery', + 'LeaderClient', 'LeaderLockInfo', 'ServiceDiscovery', 'ServiceInfo', diff --git a/src/storage/leader_client.py b/src/storage/leader_client.py new file mode 100644 index 0000000..021114f --- /dev/null +++ b/src/storage/leader_client.py @@ -0,0 +1,260 @@ +""" +LeaderClient - Client for reading leader info from meta-core + +This replaces the old LeaderDiscovery.py by delegating leader election +to meta-core (Go sidecar) and just reading the results. + +Features: +- Reads kv-leader.info file for leader URLs +- Calls meta-core /urls API for current leader URLs +- Watches for leader changes via file system watcher +""" +from __future__ import annotations + +import os +import json +import threading +from dataclasses import dataclass +from pathlib import Path +from typing import Optional, Callable +from urllib.request import urlopen, Request +from urllib.error import URLError + +# Try to import watchdog for file watching (optional) +try: + from watchdog.observers import Observer + from watchdog.events import FileSystemEventHandler + WATCHDOG_AVAILABLE = True +except ImportError: + WATCHDOG_AVAILABLE = False + Observer = None + FileSystemEventHandler = object + + +@dataclass +class LeaderLockInfo: + """Leader lock file content format (kv-leader.info). + Written by meta-core leader, read by other services. + """ + hostname: str + base_url: str + api_url: str # meta-core API URL (port 9000) + redis_url: str + webdav_url: str + timestamp: int + pid: int + + +@dataclass +class URLsResponse: + """Response from meta-core /urls API.""" + hostname: str + base_url: str + api_url: str + redis_url: str + webdav_url: str + is_leader: bool + + +class LockFileHandler(FileSystemEventHandler): + """File system event handler for lock file changes.""" + + def __init__(self, callback: Callable[[], None], lock_filename: str): + self.callback = callback + self.lock_filename = lock_filename + + def on_modified(self, event): + if event.src_path.endswith(self.lock_filename): + self.callback() + + def on_created(self, event): + if event.src_path.endswith(self.lock_filename): + self.callback() + + def on_deleted(self, event): + if event.src_path.endswith(self.lock_filename): + self.callback() + + +class LeaderClient: + """ + LeaderClient for meta-stremio. + + Reads leader info from kv-leader.info file written by meta-core. + """ + + def __init__( + self, + meta_core_path: str, + meta_core_url: Optional[str] = None + ): + self.meta_core_path = meta_core_path + self.meta_core_url = meta_core_url + + self.info_file_path = os.path.join(meta_core_path, 'locks', 'kv-leader.info') + self.leader_info: Optional[LeaderLockInfo] = None + + self._file_observer = None + self._on_change_callbacks: list[Callable[[], None]] = [] + self._stop_event = threading.Event() + + def get_leader_info(self) -> Optional[LeaderLockInfo]: + """Read leader info from the info file.""" + try: + with open(self.info_file_path, 'r') as f: + content = f.read() + + data = json.loads(content) + + self.leader_info = LeaderLockInfo( + hostname=data.get('hostname', 'unknown'), + base_url=data.get('baseUrl', ''), + api_url=data.get('apiUrl', ''), + redis_url=data.get('redisUrl', ''), + webdav_url=data.get('webdavUrl', ''), + timestamp=data.get('timestamp', 0), + pid=data.get('pid', 0) + ) + + return self.leader_info + + except FileNotFoundError: + print(f"[LeaderClient] Leader info file not found: {self.info_file_path}") + return None + except json.JSONDecodeError as e: + print(f"[LeaderClient] Error parsing leader info: {e}") + return None + except Exception as e: + print(f"[LeaderClient] Error reading leader info: {e}") + return None + + def get_redis_url(self) -> Optional[str]: + """Get Redis URL from leader info.""" + info = self.get_leader_info() + return info.redis_url if info else None + + def get_webdav_url(self) -> Optional[str]: + """Get WebDAV URL from leader info.""" + info = self.get_leader_info() + return info.webdav_url if info else None + + def get_api_url(self) -> Optional[str]: + """Get meta-core API URL from leader info.""" + info = self.get_leader_info() + return info.api_url if info else None + + def get_urls(self) -> Optional[URLsResponse]: + """Call meta-core /urls API to get current URLs.""" + # First try using configured meta_core_url + api_url = self.meta_core_url + + # Fall back to reading from leader info + if not api_url: + info = self.get_leader_info() + api_url = info.api_url if info else None + + if not api_url: + print("[LeaderClient] No meta-core API URL available") + return None + + try: + url = f"{api_url}/urls" + req = Request(url, headers={'Accept': 'application/json'}) + with urlopen(req, timeout=5) as response: + if response.status != 200: + print(f"[LeaderClient] Failed to get URLs: {response.status}") + return None + + data = json.loads(response.read().decode()) + + return URLsResponse( + hostname=data.get('hostname', ''), + base_url=data.get('baseUrl', ''), + api_url=data.get('apiUrl', ''), + redis_url=data.get('redisUrl', ''), + webdav_url=data.get('webdavUrl', ''), + is_leader=data.get('isLeader', False) + ) + + except URLError as e: + print(f"[LeaderClient] Error calling /urls API: {e}") + return None + except Exception as e: + print(f"[LeaderClient] Error getting URLs: {e}") + return None + + def wait_for_leader(self, timeout_ms: int = 30000) -> LeaderLockInfo: + """Wait for leader info to be available.""" + import time + start_time = time.time() + poll_interval = 0.5 # 500ms + + while (time.time() - start_time) * 1000 < timeout_ms: + info = self.get_leader_info() + if info: + print(f"[LeaderClient] Leader found: {info.hostname} at {info.redis_url}") + return info + + print("[LeaderClient] Waiting for leader...") + time.sleep(poll_interval) + + raise TimeoutError(f"No leader found within {timeout_ms}ms") + + def start_watching(self) -> None: + """Start watching for leader changes.""" + if not WATCHDOG_AVAILABLE: + print("[LeaderClient] watchdog not available, file watching disabled") + return + + try: + lock_dir = os.path.dirname(self.info_file_path) + + handler = LockFileHandler( + callback=self._on_file_change, + lock_filename='kv-leader.info' + ) + + self._file_observer = Observer() + self._file_observer.schedule(handler, lock_dir, recursive=False) + self._file_observer.start() + print(f"[LeaderClient] Watching for leader changes in {lock_dir}") + + except Exception as e: + print(f"[LeaderClient] Could not start file watcher: {e}") + + def stop_watching(self) -> None: + """Stop watching for leader changes.""" + if self._file_observer: + self._file_observer.stop() + self._file_observer.join(timeout=2.0) + self._file_observer = None + print("[LeaderClient] Stopped watching for leader changes") + + def _on_file_change(self) -> None: + """Handle file change event.""" + print("[LeaderClient] Leader info changed, reloading...") + self.get_leader_info() + self._notify_change() + + def on_change(self, callback: Callable[[], None]) -> 'LeaderClient': + """Register callback for leader changes.""" + self._on_change_callbacks.append(callback) + return self + + def _notify_change(self) -> None: + """Notify all change callbacks.""" + for callback in self._on_change_callbacks: + try: + callback() + except Exception as e: + print(f"[LeaderClient] Error in change callback: {e}") + + def get_cached_leader_info(self) -> Optional[LeaderLockInfo]: + """Get cached leader info (without re-reading file).""" + return self.leader_info + + def close(self) -> None: + """Clean up resources.""" + self._stop_event.set() + self.stop_watching() + self._on_change_callbacks = [] diff --git a/src/storage/leader_discovery.py b/src/storage/leader_discovery.py deleted file mode 100644 index c3e2428..0000000 --- a/src/storage/leader_discovery.py +++ /dev/null @@ -1,300 +0,0 @@ -""" -Leader Discovery for meta-stremio (FOLLOWER-only) - -This is a Python implementation of the leader discovery pattern used in meta-fuse. -meta-stremio never spawns Redis - it only connects to the leader's Redis instance. - -Architecture: -- Reads info file at META_CORE_PATH/locks/kv-leader.info -- The leader holds flock on kv-leader.lock, writes metadata to kv-leader.info -- Parses leader info (Redis URL, HTTP URL) -- Watches for leader changes via polling and file watching -- Triggers reconnection callbacks on leader failure/change -""" -from __future__ import annotations - -import os -import json -import time -import threading -from dataclasses import dataclass -from pathlib import Path -from typing import Optional, Callable -from urllib.request import urlopen -from urllib.error import URLError - -# Try to import watchdog for file watching (optional) -try: - from watchdog.observers import Observer - from watchdog.events import FileSystemEventHandler - WATCHDOG_AVAILABLE = True -except ImportError: - WATCHDOG_AVAILABLE = False - Observer = None - FileSystemEventHandler = object - - -@dataclass -class LeaderLockInfo: - """Leader lock file content format.""" - host: str - api: str # Redis connection URL (e.g., redis://10.0.1.50:6379) - http: str # HTTP API URL for the leader service - timestamp: int - pid: int - - -class LockFileHandler(FileSystemEventHandler): - """File system event handler for lock file changes.""" - - def __init__(self, callback: Callable[[], None], lock_filename: str): - self.callback = callback - self.lock_filename = lock_filename - - def on_modified(self, event): - if event.src_path.endswith(self.lock_filename): - self.callback() - - def on_created(self, event): - if event.src_path.endswith(self.lock_filename): - self.callback() - - def on_deleted(self, event): - if event.src_path.endswith(self.lock_filename): - self.callback() - - -class LeaderDiscovery: - """ - Leader Discovery for meta-stremio. - - Discovers and monitors the KV leader via a lock file in the shared filesystem. - """ - - def __init__( - self, - meta_core_path: str, - health_check_interval: float = 5.0, - max_failures: int = 3 - ): - self.meta_core_path = meta_core_path - self.health_check_interval = health_check_interval - self.max_failures = max_failures - - self.lock_file_path = os.path.join(meta_core_path, 'locks', 'kv-leader.info') - self.leader_info: Optional[LeaderLockInfo] = None - - self._is_started = False - self._stop_event = threading.Event() - self._health_check_thread: Optional[threading.Thread] = None - self._file_observer = None - self._consecutive_failures = 0 - - # Event callbacks - self._on_leader_found_callback: Optional[Callable[[LeaderLockInfo], None]] = None - self._on_leader_lost_callback: Optional[Callable[[], None]] = None - - def start(self) -> None: - """Start watching for leader.""" - if self._is_started: - print("[LeaderDiscovery] Already started") - return - - print(f"[LeaderDiscovery] Starting, lock file: {self.lock_file_path}") - - # Ensure lock directory exists - self._ensure_lock_dir() - - # Try to read existing leader info - self._check_for_leader() - - # Start file watcher (if watchdog is available) - self._start_file_watcher() - - # Start health check thread - self._start_health_check() - - self._is_started = True - - def stop(self) -> None: - """Stop watching.""" - if not self._is_started: - return - - print("[LeaderDiscovery] Stopping...") - - self._stop_event.set() - - # Stop health check thread - if self._health_check_thread and self._health_check_thread.is_alive(): - self._health_check_thread.join(timeout=2.0) - - # Stop file observer - if self._file_observer: - self._file_observer.stop() - self._file_observer.join(timeout=2.0) - self._file_observer = None - - self._is_started = False - - def _ensure_lock_dir(self) -> None: - """Ensure lock directory exists.""" - lock_dir = os.path.dirname(self.lock_file_path) - try: - os.makedirs(lock_dir, exist_ok=True) - except OSError as e: - print(f"[LeaderDiscovery] Could not create lock directory: {e}") - - def _check_for_leader(self) -> None: - """Check for leader by reading lock file.""" - try: - with open(self.lock_file_path, 'r') as f: - content = f.read() - - data = json.loads(content) - - # Validate leader info - if data and data.get('api') and data.get('http'): - info = LeaderLockInfo( - host=data.get('host', 'unknown'), - api=data['api'], - http=data['http'], - timestamp=data.get('timestamp', 0), - pid=data.get('pid', 0) - ) - - is_new = ( - self.leader_info is None or - self.leader_info.api != info.api - ) - - self.leader_info = info - self._consecutive_failures = 0 - - if is_new: - print(f"[LeaderDiscovery] Leader found: {info.host} at {info.api}") - if self._on_leader_found_callback: - self._on_leader_found_callback(info) - - except FileNotFoundError: - # Lock file doesn't exist - no leader yet - if self.leader_info: - print("[LeaderDiscovery] Lock file removed, leader lost") - self._handle_leader_lost() - - except json.JSONDecodeError as e: - print(f"[LeaderDiscovery] Error parsing lock file: {e}") - - except Exception as e: - print(f"[LeaderDiscovery] Error reading lock file: {e}") - - def _is_leader_healthy(self) -> bool: - """Check if current leader is still healthy.""" - if not self.leader_info: - return False - - try: - health_url = f"{self.leader_info.http}/health" - with urlopen(health_url, timeout=5) as response: - return response.status == 200 - except (URLError, TimeoutError, Exception): - return False - - def _handle_leader_lost(self) -> None: - """Handle leader being lost.""" - self.leader_info = None - if self._on_leader_lost_callback: - self._on_leader_lost_callback() - - def _start_file_watcher(self) -> None: - """Start watching lock file for changes.""" - if not WATCHDOG_AVAILABLE: - print("[LeaderDiscovery] watchdog not available, using polling only") - return - - try: - lock_dir = os.path.dirname(self.lock_file_path) - handler = LockFileHandler( - callback=self._check_for_leader, - lock_filename='kv-leader.info' - ) - - self._file_observer = Observer() - self._file_observer.schedule(handler, lock_dir, recursive=False) - self._file_observer.start() - print(f"[LeaderDiscovery] File watcher started for {lock_dir}") - - except Exception as e: - print(f"[LeaderDiscovery] Could not start file watcher: {e}") - - def _start_health_check(self) -> None: - """Start health check loop.""" - def health_check_loop(): - while not self._stop_event.is_set(): - try: - if not self.leader_info: - # No leader known, try to find one - self._check_for_leader() - else: - # Check if current leader is healthy - healthy = self._is_leader_healthy() - - if healthy: - self._consecutive_failures = 0 - else: - self._consecutive_failures += 1 - print(f"[LeaderDiscovery] Leader health check failed ({self._consecutive_failures}/{self.max_failures})") - - if self._consecutive_failures >= self.max_failures: - print("[LeaderDiscovery] Leader appears dead, marking as lost") - self._handle_leader_lost() - - # Try to find new leader - self._check_for_leader() - - except Exception as e: - print(f"[LeaderDiscovery] Health check error: {e}") - - self._stop_event.wait(self.health_check_interval) - - self._health_check_thread = threading.Thread( - target=health_check_loop, - daemon=True, - name="leader-health-check" - ) - self._health_check_thread.start() - - # ======================================================================== - # Event Registration - # ======================================================================== - - def on_leader_found(self, callback: Callable[[LeaderLockInfo], None]) -> 'LeaderDiscovery': - """Register callback for when leader is found/changed.""" - self._on_leader_found_callback = callback - - # If we already have leader info, call immediately - if self.leader_info: - callback(self.leader_info) - - return self - - def on_leader_lost(self, callback: Callable[[], None]) -> 'LeaderDiscovery': - """Register callback for when leader is lost.""" - self._on_leader_lost_callback = callback - return self - - # ======================================================================== - # Getters - # ======================================================================== - - def get_leader_info(self) -> Optional[LeaderLockInfo]: - """Get current leader info.""" - return self.leader_info - - def get_redis_url(self) -> Optional[str]: - """Get Redis URL of current leader.""" - return self.leader_info.api if self.leader_info else None - - def has_leader(self) -> bool: - """Check if a leader is known.""" - return self.leader_info is not None diff --git a/src/storage/leader_storage.py b/src/storage/leader_storage.py index 7ca7a45..2c357a9 100644 --- a/src/storage/leader_storage.py +++ b/src/storage/leader_storage.py @@ -1,7 +1,7 @@ """ Leader-aware Storage Provider for meta-stremio. -Uses leader discovery to automatically connect to the KV leader. +Uses LeaderClient to read leader info from meta-core. Handles reconnection when the leader changes or becomes unavailable. """ from __future__ import annotations @@ -12,7 +12,7 @@ import threading from typing import List, Optional, Callable from .provider import StorageProvider, VideoMetadata -from .leader_discovery import LeaderDiscovery, LeaderLockInfo +from .leader_client import LeaderClient, LeaderLockInfo try: import redis @@ -32,8 +32,8 @@ class LeaderStorage(StorageProvider): """ Storage provider that discovers and connects to the KV leader. - Uses leader discovery to automatically find the Redis instance - managed by meta-sort. Handles reconnection on leader failure. + Uses LeaderClient to read leader info from meta-core. + Handles reconnection on leader failure. """ def __init__( @@ -51,7 +51,7 @@ class LeaderStorage(StorageProvider): self._prefix = prefix or REDIS_PREFIX self._override_url = redis_url # Direct Redis URL (skip leader discovery) - self._leader_discovery: Optional[LeaderDiscovery] = None + self._leader_client: Optional[LeaderClient] = None self._client: Optional[redis.Redis] = None self._connected = False self._lock = threading.Lock() @@ -61,31 +61,39 @@ class LeaderStorage(StorageProvider): self._on_disconnect_callbacks: List[Callable[[], None]] = [] def connect(self) -> None: - """Connect to the storage backend via leader discovery.""" + """Connect to the storage backend via leader client.""" # If direct URL provided, skip leader discovery if self._override_url: print(f"[LeaderStorage] Using direct Redis URL: {self._override_url}") self._connect_to_redis(self._override_url) return - # Use leader discovery - self._leader_discovery = LeaderDiscovery( + # Use leader client + self._leader_client = LeaderClient( meta_core_path=self._meta_core_path ) - # Set up callbacks - self._leader_discovery.on_leader_found(self._on_leader_found) - self._leader_discovery.on_leader_lost(self._on_leader_lost) + # Set up change callback + self._leader_client.on_change(self._on_leader_change) - # Start discovery - self._leader_discovery.start() + # Wait for leader and connect + try: + info = self._leader_client.wait_for_leader(30000) + print(f"[LeaderStorage] Connecting to leader at {info.redis_url}...") + self._connect_to_redis(info.redis_url) + except TimeoutError as e: + print(f"[LeaderStorage] Failed to find leader: {e}") + return + + # Start watching for leader changes + self._leader_client.start_watching() def disconnect(self) -> None: """Disconnect from the storage backend.""" - # Stop leader discovery - if self._leader_discovery: - self._leader_discovery.stop() - self._leader_discovery = None + # Stop leader client + if self._leader_client: + self._leader_client.close() + self._leader_client = None # Disconnect Redis self._disconnect_redis() @@ -102,17 +110,21 @@ class LeaderStorage(StorageProvider): self._connected = False return False - def _on_leader_found(self, info: LeaderLockInfo) -> None: - """Handle leader found event.""" - print(f"[LeaderStorage] Connecting to leader at {info.api}...") - self._connect_to_redis(info.api) - - def _on_leader_lost(self) -> None: - """Handle leader lost event.""" - print("[LeaderStorage] Leader lost, disconnecting...") + def _on_leader_change(self) -> None: + """Handle leader change event.""" + print("[LeaderStorage] Leader changed, reconnecting...") self._disconnect_redis() self._notify_disconnect() + # Reconnect to new leader + if self._leader_client: + try: + info = self._leader_client.wait_for_leader(30000) + print(f"[LeaderStorage] Reconnecting to leader at {info.redis_url}...") + self._connect_to_redis(info.redis_url) + except TimeoutError as e: + print(f"[LeaderStorage] Failed to reconnect: {e}") + def _connect_to_redis(self, url: str) -> None: """Connect to Redis.""" with self._lock: @@ -523,13 +535,15 @@ class LeaderStorage(StorageProvider): def get_status(self) -> dict: """Get storage status for dashboard.""" leader_info = None - if self._leader_discovery: - info = self._leader_discovery.get_leader_info() + if self._leader_client: + info = self._leader_client.get_cached_leader_info() if info: leader_info = { - 'host': info.host, - 'api': info.api, - 'http': info.http, + 'hostname': info.hostname, + 'base_url': info.base_url, + 'api_url': info.api_url, + 'redis_url': info.redis_url, + 'webdav_url': info.webdav_url, } status = { diff --git a/src/stremio.py b/src/stremio.py index 56b7fae..ebdd2d8 100644 --- a/src/stremio.py +++ b/src/stremio.py @@ -22,11 +22,10 @@ from urllib.parse import quote from typing import Optional, List, Dict from collections import defaultdict -from storage import StorageProvider, VideoMetadata, RedisStorage, LeaderStorage +from storage import StorageProvider, VideoMetadata, LeaderStorage # Configuration MEDIA_DIR = os.environ.get('MEDIA_DIR', '/files/watch') -STORAGE_MODE = os.environ.get('STORAGE_MODE', 'redis').lower() # 'leader' or 'redis' # Poster dimensions for Stremio POSTER_WIDTH = 342 # Stremio standard poster width @@ -68,23 +67,15 @@ def get_poster_url(cid: Optional[str], width: Optional[int] = None) -> str: def init_storage() -> StorageProvider: - """Initialize the storage provider based on configuration. + """Initialize the storage provider. - Storage modes: - - 'leader': Use leader discovery to find the KV store (recommended) - - 'redis': Direct Redis connection using REDIS_URL + Uses LeaderStorage which reads leader info from meta-core. """ global _storage - if STORAGE_MODE == 'redis': - _storage = RedisStorage() - _storage.connect() - print(f"[Stremio] Using Redis storage") - else: - # Default to leader discovery - _storage = LeaderStorage() - _storage.connect() - print(f"[Stremio] Using leader discovery storage") + _storage = LeaderStorage() + _storage.connect() + print("[Stremio] Using leader storage (reads leader info from meta-core)") return _storage