From 2c772fd820df1cc16b237391036c817d61693ca5 Mon Sep 17 00:00:00 2001 From: Pierre Date: Sat, 7 Mar 2026 14:44:07 +0100 Subject: [PATCH] bugfix event --- src/server.py | 68 ++----- src/storage/__init__.py | 20 +- src/storage/leader_client.py | 17 ++ ...e_discovery.py => service_registration.py} | 192 ++++-------------- www/index.html | 7 +- 5 files changed, 92 insertions(+), 212 deletions(-) rename src/storage/{service_discovery.py => service_registration.py} (56%) diff --git a/src/server.py b/src/server.py index 38f71a5..ddb705e 100644 --- a/src/server.py +++ b/src/server.py @@ -569,59 +569,35 @@ class Handler(BaseHTTPRequestHandler): self.wfile.write(content.encode('utf-8')) def handle_services_api(self): - """Return list of discovered services for dashboard navigation.""" - services = [] - leader_info = None + """Return list of discovered services for dashboard navigation. + + Fetches services from meta-core (single source of truth) instead of + doing local filesystem-based discovery. + """ + from storage.leader_client import get_leader_client + from urllib.request import urlopen + from urllib.error import URLError try: - sd = get_service_discovery() - if sd and sd.is_started(): - all_services = sd.discover_all_services() + leader_client = get_leader_client() + leader_info = leader_client.get_leader_info() if leader_client else None - for svc in all_services: - # Filter out follower meta-core instances (only show leader) - svc_name = svc.get('name', '') - svc_role = svc.get('role', '') - if svc_name == 'meta-core' and svc_role != 'leader': - continue - - # Build dashboard URL from base URL (supports both 'baseUrl' and legacy 'api') - api_url = svc.get('baseUrl', '') or svc.get('api', '') - dashboard_path = svc.get('endpoints', {}).get('dashboard', '/') - - services.append({ - 'name': svc_name or 'Unknown', - 'url': api_url + dashboard_path if api_url else '', - 'api': api_url, - 'status': svc.get('status', 'unknown'), - 'capabilities': svc.get('capabilities', []), - 'version': svc.get('version', ''), - 'role': svc_role or None, - }) + if leader_info and leader_info.api_url: + # Fetch services from meta-core API + url = f"{leader_info.api_url}/services?current=meta-stremio" + with urlopen(url, timeout=5) as response: + data = json.loads(response.read().decode()) + return self.send_json(data) + except URLError as e: + print(f"[Server] Error fetching services from meta-core: {e}") except Exception as e: - print(f"[Server] Error discovering services: {e}") - - # Get leader info from leader storage - try: - storage = stremio.get_storage() - if storage and hasattr(storage, '_leader_discovery'): - discovery = storage._leader_discovery - if discovery: - info = discovery.get_leader_info() - if info: - leader_info = { - 'host': info.hostname if hasattr(info, 'hostname') else info.get('hostname', ''), - 'api': info.api_url if hasattr(info, 'api_url') else info.get('apiUrl', ''), - 'http': info.base_url if hasattr(info, 'base_url') else info.get('baseUrl', ''), - } - except Exception as e: - print(f"[Server] Error getting leader info: {e}") + print(f"[Server] Error in handle_services_api: {e}") + # Fallback: return empty list if meta-core is unavailable return self.send_json({ - 'services': services, + 'services': [], 'current': 'meta-stremio', - 'leader': leader_info, - 'isLeader': False, # meta-stremio is never the leader + 'count': 0, }) def handle_file(self, cid: str, width: int = None): diff --git a/src/storage/__init__.py b/src/storage/__init__.py index 475bfe4..323fb2e 100644 --- a/src/storage/__init__.py +++ b/src/storage/__init__.py @@ -5,20 +5,24 @@ Provides a unified interface for reading video metadata from different backends: - LeaderStorage: Discovers and connects to the KV leader (recommended) - RedisStorage: Reads from Redis with direct URL (legacy) -Also provides service discovery for inter-service communication. +Also provides service registration for inter-service communication. +Service discovery is centralized in meta-core. """ from .provider import StorageProvider, VideoMetadata from .redis_storage import RedisStorage from .leader_storage import LeaderStorage -from .leader_client import LeaderClient, LeaderLockInfo -from .service_discovery import ( - ServiceDiscovery, +from .leader_client import LeaderClient, LeaderLockInfo, get_leader_client +from .service_registration import ( + ServiceRegistration, ServiceInfo, - get_service_discovery, - init_service_discovery + get_service_discovery, # Kept for backward compatibility + init_service_discovery # Kept for backward compatibility ) +# Backward compatibility alias +ServiceDiscovery = ServiceRegistration + __all__ = [ 'StorageProvider', 'VideoMetadata', @@ -26,7 +30,9 @@ __all__ = [ 'LeaderStorage', 'LeaderClient', 'LeaderLockInfo', - 'ServiceDiscovery', + 'get_leader_client', + 'ServiceRegistration', + 'ServiceDiscovery', # Backward compatibility alias 'ServiceInfo', 'get_service_discovery', 'init_service_discovery', diff --git a/src/storage/leader_client.py b/src/storage/leader_client.py index 92dfa0c..4698dde 100644 --- a/src/storage/leader_client.py +++ b/src/storage/leader_client.py @@ -304,3 +304,20 @@ class LeaderClient: self._stop_event.set() self.stop_watching() self._on_change_callbacks = [] + + +# Singleton instance +_leader_client: Optional[LeaderClient] = None + + +def get_leader_client() -> Optional[LeaderClient]: + """Get or create the LeaderClient singleton. + + Returns None if META_CORE_PATH is not set. + """ + global _leader_client + if _leader_client is None: + meta_core_path = os.environ.get('META_CORE_PATH', '/meta-core') + meta_core_url = os.environ.get('META_CORE_URL') + _leader_client = LeaderClient(meta_core_path, meta_core_url) + return _leader_client diff --git a/src/storage/service_discovery.py b/src/storage/service_registration.py similarity index 56% rename from src/storage/service_discovery.py rename to src/storage/service_registration.py index dc85a03..d81e0dc 100644 --- a/src/storage/service_discovery.py +++ b/src/storage/service_registration.py @@ -1,14 +1,14 @@ """ -Service Discovery for meta-stremio. +Service Registration for meta-stremio. -Python implementation of the service discovery pattern used in meta-sort/meta-fuse. -Each service registers itself in /meta-core/services/{service-name}-{hostname}.json -and can discover other registered services. +Python implementation of the service registration pattern used in meta-sort/meta-fuse. +Each service registers itself in /meta-core/services/{service-name}-{hostname}.json. + +Service discovery is centralized in meta-core - this module only handles registration. +For service discovery, use the /api/services endpoint (fetches from meta-core). Features: - Service registration with heartbeat -- Discovery of all running services -- Stale service detection (60 second threshold) - Automatic heartbeat loop """ from __future__ import annotations @@ -62,7 +62,7 @@ class ServiceInfo: return {k: v for k, v in result.items() if v is not None} -class ServiceDiscovery: +class ServiceRegistration: """ Service Discovery using shared filesystem. @@ -144,7 +144,7 @@ class ServiceDiscovery: with open(self.service_file_path, 'w') as f: json.dump(info.to_dict(), f, indent=2) - print(f'[ServiceDiscovery] Registered {self.service_name}') + print(f'[ServiceRegistration] Registered {self.service_name}') def update_status(self, status: str) -> None: """Update service status.""" @@ -159,7 +159,7 @@ class ServiceDiscovery: json.dump(info, f, indent=2) except Exception as e: - print(f'[ServiceDiscovery] Failed to update status: {e}') + print(f'[ServiceRegistration] Failed to update status: {e}') def heartbeat(self) -> None: """Send heartbeat (update lastHeartbeat timestamp).""" @@ -177,7 +177,7 @@ class ServiceDiscovery: self.register() self.update_status('running') except Exception as e: - print(f'[ServiceDiscovery] Heartbeat failed: {e}') + print(f'[ServiceRegistration] Heartbeat failed: {e}') def _heartbeat_loop(self) -> None: """Background heartbeat loop.""" @@ -185,7 +185,7 @@ class ServiceDiscovery: try: self.heartbeat() except Exception as e: - print(f'[ServiceDiscovery] Heartbeat error: {e}') + print(f'[ServiceRegistration] Heartbeat error: {e}') self._stop_event.wait(self.heartbeat_interval) @@ -214,135 +214,9 @@ class ServiceDiscovery: try: self.update_status('stopped') - print(f'[ServiceDiscovery] Unregistered {self.service_name}') + print(f'[ServiceRegistration] Unregistered {self.service_name}') except Exception as e: - print(f'[ServiceDiscovery] Failed to unregister: {e}') - - # ======================================================================== - # Discovery Methods - # ======================================================================== - - def _is_service_stale(self, info: dict) -> bool: - """Check if a service is stale based on lastHeartbeat.""" - try: - last_heartbeat_str = info.get('lastHeartbeat', '') - if not last_heartbeat_str: - return True - - # Parse ISO timestamp - last_heartbeat_str = last_heartbeat_str.rstrip('Z') - last_heartbeat = datetime.fromisoformat(last_heartbeat_str) - age = (datetime.utcnow() - last_heartbeat).total_seconds() - - return age > self.stale_threshold - except Exception: - return True - - def discover_service(self, name: str) -> Optional[dict]: - """ - Discover a service by name. - Looks for files matching pattern: {name}-*.json (hostname-based naming) - Falls back to exact match {name}.json for backwards compatibility. - """ - try: - if not os.path.exists(self.services_dir): - return None - - # First try exact match for backward compatibility - exact_path = os.path.join(self.services_dir, f'{name}.json') - if os.path.exists(exact_path): - with open(exact_path, 'r') as f: - info = json.load(f) - if not self._is_service_stale(info) and info.get('status') == 'running': - return info - - # Search for hostname-based files: name-*.json - for filename in os.listdir(self.services_dir): - if filename.startswith(f'{name}-') and filename.endswith('.json'): - filepath = os.path.join(self.services_dir, filename) - with open(filepath, 'r') as f: - info = json.load(f) - if not self._is_service_stale(info) and info.get('status') == 'running': - return info - - return None - - except Exception as e: - print(f'[ServiceDiscovery] Error discovering {name}: {e}') - return None - - def discover_all_services(self) -> List[dict]: - """Discover all registered services.""" - services = [] - seen_names = set() - meta_core_instances = [] # Collect all meta-core instances to pick leader - - try: - if not os.path.exists(self.services_dir): - return services - - for filename in os.listdir(self.services_dir): - if not filename.endswith('.json'): - continue - - filepath = os.path.join(self.services_dir, filename) - try: - with open(filepath, 'r') as f: - info = json.load(f) - - # Skip stale or non-running services - if self._is_service_stale(info): - continue - if info.get('status') != 'running': - continue - - name = info.get('name', '') - - # Special handling for meta-core: collect all instances - # to pick the leader later (filesystem order is unreliable) - if name == 'meta-core': - meta_core_instances.append(info) - continue - - # Deduplicate by service name (keep first found) - if name and name not in seen_names: - seen_names.add(name) - services.append(info) - - except Exception as e: - print(f'[ServiceDiscovery] Failed to read {filename}: {e}') - - # Add meta-core: prefer leader, fallback to first instance - if meta_core_instances: - leader = next( - (s for s in meta_core_instances if s.get('role') == 'leader'), - meta_core_instances[0] # Fallback if no leader found - ) - services.append(leader) - - except Exception as e: - print(f'[ServiceDiscovery] Error discovering services: {e}') - - return services - - def is_service_healthy(self, name: str) -> bool: - """Check if a service is healthy via HTTP ping.""" - service = self.discover_service(name) - if not service: - return False - - # Use baseUrl (new format) or api (legacy format) - base_url = service.get('baseUrl', '') or service.get('api', '') - if not base_url: - return False - - url = f"{base_url}/health" - - try: - with urlopen(url, timeout=5) as response: - return response.status == 200 - except Exception: - return False + print(f'[ServiceRegistration] Failed to unregister: {e}') # ======================================================================== # Lifecycle @@ -362,12 +236,12 @@ class ServiceDiscovery: self.register() self.update_status('running') self.start_heartbeat() - print(f'[ServiceDiscovery] Started {self.service_name}-{self.current_hostname} at {self.base_url}') + print(f'[ServiceRegistration] Started {self.service_name}-{self.current_hostname} at {self.base_url}') except Exception as e: - print(f'[ServiceDiscovery] Registration failed (read-only?): {e}') - print(f'[ServiceDiscovery] Running in discovery-only mode') + print(f'[ServiceRegistration] Registration failed (read-only?): {e}') + print(f'[ServiceRegistration] Running in discovery-only mode') else: - print(f'[ServiceDiscovery] Services directory not writable, running in discovery-only mode') + print(f'[ServiceRegistration] Services directory not writable, running in discovery-only mode') self._is_started = True @@ -394,24 +268,30 @@ class ServiceDiscovery: # Singleton instance -_service_discovery: Optional[ServiceDiscovery] = None +_service_registration: Optional[ServiceRegistration] = None -def get_service_discovery() -> ServiceDiscovery: - """Get or create the service discovery singleton.""" - global _service_discovery - if _service_discovery is None: - _service_discovery = ServiceDiscovery() - return _service_discovery +def get_service_discovery() -> ServiceRegistration: + """Get or create the service registration singleton. + + Note: Function name kept as 'get_service_discovery' for backward compatibility. + """ + global _service_registration + if _service_registration is None: + _service_registration = ServiceRegistration() + return _service_registration def init_service_discovery( api_url: str = None, base_url: str = None, **kwargs -) -> ServiceDiscovery: - """Initialize and start the service discovery singleton.""" - global _service_discovery - _service_discovery = ServiceDiscovery(api_url=api_url, base_url=base_url, **kwargs) - _service_discovery.start() - return _service_discovery +) -> ServiceRegistration: + """Initialize and start the service registration singleton. + + Note: Function name kept as 'init_service_discovery' for backward compatibility. + """ + global _service_registration + _service_registration = ServiceRegistration(api_url=api_url, base_url=base_url, **kwargs) + _service_registration.start() + return _service_registration diff --git a/www/index.html b/www/index.html index 1137608..8c1a3d6 100644 --- a/www/index.html +++ b/www/index.html @@ -552,8 +552,9 @@ 'meta-sort': '📁', 'meta-fuse': '🗂️', 'meta-stremio': '🎬', - 'meta-orbit': '🌐', 'meta-core': '⚙️', + 'meta-dup': '🔍', + 'meta-orbit': '🌐', 'default': '📦' }; @@ -577,10 +578,10 @@ const activeClass = isActive ? ' active' : ''; // Use external URL for other services, current path for self - const href = isActive ? '#' : svc.url; + const href = isActive ? '#' : svc.baseUrl; return ` - + ${icon} ${formatServiceName(svc.name)}