bugfix event

This commit is contained in:
Pierre
2026-03-07 14:44:07 +01:00
parent f26fea797c
commit 51f57af68e
5 changed files with 92 additions and 212 deletions
+22 -46
View File
@@ -569,59 +569,35 @@ class Handler(BaseHTTPRequestHandler):
self.wfile.write(content.encode('utf-8')) self.wfile.write(content.encode('utf-8'))
def handle_services_api(self): def handle_services_api(self):
"""Return list of discovered services for dashboard navigation.""" """Return list of discovered services for dashboard navigation.
services = []
leader_info = None 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: try:
sd = get_service_discovery() leader_client = get_leader_client()
if sd and sd.is_started(): leader_info = leader_client.get_leader_info() if leader_client else None
all_services = sd.discover_all_services()
for svc in all_services: if leader_info and leader_info.api_url:
# Filter out follower meta-core instances (only show leader) # Fetch services from meta-core API
svc_name = svc.get('name', '') url = f"{leader_info.api_url}/services?current=meta-stremio"
svc_role = svc.get('role', '') with urlopen(url, timeout=5) as response:
if svc_name == 'meta-core' and svc_role != 'leader': data = json.loads(response.read().decode())
continue return self.send_json(data)
except URLError as e:
# Build dashboard URL from base URL (supports both 'baseUrl' and legacy 'api') print(f"[Server] Error fetching services from meta-core: {e}")
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,
})
except Exception as e: except Exception as e:
print(f"[Server] Error discovering services: {e}") print(f"[Server] Error in handle_services_api: {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}")
# Fallback: return empty list if meta-core is unavailable
return self.send_json({ return self.send_json({
'services': services, 'services': [],
'current': 'meta-stremio', 'current': 'meta-stremio',
'leader': leader_info, 'count': 0,
'isLeader': False, # meta-stremio is never the leader
}) })
def handle_file(self, cid: str, width: int = None): def handle_file(self, cid: str, width: int = None):
+13 -7
View File
@@ -5,20 +5,24 @@ Provides a unified interface for reading video metadata from different backends:
- LeaderStorage: Discovers and connects to the KV leader (recommended) - LeaderStorage: Discovers and connects to the KV leader (recommended)
- RedisStorage: Reads from Redis with direct URL (legacy) - 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 .provider import StorageProvider, VideoMetadata
from .redis_storage import RedisStorage from .redis_storage import RedisStorage
from .leader_storage import LeaderStorage from .leader_storage import LeaderStorage
from .leader_client import LeaderClient, LeaderLockInfo from .leader_client import LeaderClient, LeaderLockInfo, get_leader_client
from .service_discovery import ( from .service_registration import (
ServiceDiscovery, ServiceRegistration,
ServiceInfo, ServiceInfo,
get_service_discovery, get_service_discovery, # Kept for backward compatibility
init_service_discovery init_service_discovery # Kept for backward compatibility
) )
# Backward compatibility alias
ServiceDiscovery = ServiceRegistration
__all__ = [ __all__ = [
'StorageProvider', 'StorageProvider',
'VideoMetadata', 'VideoMetadata',
@@ -26,7 +30,9 @@ __all__ = [
'LeaderStorage', 'LeaderStorage',
'LeaderClient', 'LeaderClient',
'LeaderLockInfo', 'LeaderLockInfo',
'ServiceDiscovery', 'get_leader_client',
'ServiceRegistration',
'ServiceDiscovery', # Backward compatibility alias
'ServiceInfo', 'ServiceInfo',
'get_service_discovery', 'get_service_discovery',
'init_service_discovery', 'init_service_discovery',
+17
View File
@@ -304,3 +304,20 @@ class LeaderClient:
self._stop_event.set() self._stop_event.set()
self.stop_watching() self.stop_watching()
self._on_change_callbacks = [] 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
@@ -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. 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 Each service registers itself in /meta-core/services/{service-name}-{hostname}.json.
and can discover other registered services.
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: Features:
- Service registration with heartbeat - Service registration with heartbeat
- Discovery of all running services
- Stale service detection (60 second threshold)
- Automatic heartbeat loop - Automatic heartbeat loop
""" """
from __future__ import annotations from __future__ import annotations
@@ -62,7 +62,7 @@ class ServiceInfo:
return {k: v for k, v in result.items() if v is not None} return {k: v for k, v in result.items() if v is not None}
class ServiceDiscovery: class ServiceRegistration:
""" """
Service Discovery using shared filesystem. Service Discovery using shared filesystem.
@@ -144,7 +144,7 @@ class ServiceDiscovery:
with open(self.service_file_path, 'w') as f: with open(self.service_file_path, 'w') as f:
json.dump(info.to_dict(), f, indent=2) 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: def update_status(self, status: str) -> None:
"""Update service status.""" """Update service status."""
@@ -159,7 +159,7 @@ class ServiceDiscovery:
json.dump(info, f, indent=2) json.dump(info, f, indent=2)
except Exception as e: except Exception as e:
print(f'[ServiceDiscovery] Failed to update status: {e}') print(f'[ServiceRegistration] Failed to update status: {e}')
def heartbeat(self) -> None: def heartbeat(self) -> None:
"""Send heartbeat (update lastHeartbeat timestamp).""" """Send heartbeat (update lastHeartbeat timestamp)."""
@@ -177,7 +177,7 @@ class ServiceDiscovery:
self.register() self.register()
self.update_status('running') self.update_status('running')
except Exception as e: except Exception as e:
print(f'[ServiceDiscovery] Heartbeat failed: {e}') print(f'[ServiceRegistration] Heartbeat failed: {e}')
def _heartbeat_loop(self) -> None: def _heartbeat_loop(self) -> None:
"""Background heartbeat loop.""" """Background heartbeat loop."""
@@ -185,7 +185,7 @@ class ServiceDiscovery:
try: try:
self.heartbeat() self.heartbeat()
except Exception as e: except Exception as e:
print(f'[ServiceDiscovery] Heartbeat error: {e}') print(f'[ServiceRegistration] Heartbeat error: {e}')
self._stop_event.wait(self.heartbeat_interval) self._stop_event.wait(self.heartbeat_interval)
@@ -214,135 +214,9 @@ class ServiceDiscovery:
try: try:
self.update_status('stopped') self.update_status('stopped')
print(f'[ServiceDiscovery] Unregistered {self.service_name}') print(f'[ServiceRegistration] Unregistered {self.service_name}')
except Exception as e: except Exception as e:
print(f'[ServiceDiscovery] Failed to unregister: {e}') print(f'[ServiceRegistration] 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
# ======================================================================== # ========================================================================
# Lifecycle # Lifecycle
@@ -362,12 +236,12 @@ class ServiceDiscovery:
self.register() self.register()
self.update_status('running') self.update_status('running')
self.start_heartbeat() 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: except Exception as e:
print(f'[ServiceDiscovery] Registration failed (read-only?): {e}') print(f'[ServiceRegistration] Registration failed (read-only?): {e}')
print(f'[ServiceDiscovery] Running in discovery-only mode') print(f'[ServiceRegistration] Running in discovery-only mode')
else: 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 self._is_started = True
@@ -394,24 +268,30 @@ class ServiceDiscovery:
# Singleton instance # Singleton instance
_service_discovery: Optional[ServiceDiscovery] = None _service_registration: Optional[ServiceRegistration] = None
def get_service_discovery() -> ServiceDiscovery: def get_service_discovery() -> ServiceRegistration:
"""Get or create the service discovery singleton.""" """Get or create the service registration singleton.
global _service_discovery
if _service_discovery is None: Note: Function name kept as 'get_service_discovery' for backward compatibility.
_service_discovery = ServiceDiscovery() """
return _service_discovery global _service_registration
if _service_registration is None:
_service_registration = ServiceRegistration()
return _service_registration
def init_service_discovery( def init_service_discovery(
api_url: str = None, api_url: str = None,
base_url: str = None, base_url: str = None,
**kwargs **kwargs
) -> ServiceDiscovery: ) -> ServiceRegistration:
"""Initialize and start the service discovery singleton.""" """Initialize and start the service registration singleton.
global _service_discovery
_service_discovery = ServiceDiscovery(api_url=api_url, base_url=base_url, **kwargs) Note: Function name kept as 'init_service_discovery' for backward compatibility.
_service_discovery.start() """
return _service_discovery global _service_registration
_service_registration = ServiceRegistration(api_url=api_url, base_url=base_url, **kwargs)
_service_registration.start()
return _service_registration
+4 -3
View File
@@ -552,8 +552,9 @@
'meta-sort': '📁', 'meta-sort': '📁',
'meta-fuse': '🗂️', 'meta-fuse': '🗂️',
'meta-stremio': '🎬', 'meta-stremio': '🎬',
'meta-orbit': '🌐',
'meta-core': '⚙️', 'meta-core': '⚙️',
'meta-dup': '🔍',
'meta-orbit': '🌐',
'default': '📦' 'default': '📦'
}; };
@@ -577,10 +578,10 @@
const activeClass = isActive ? ' active' : ''; const activeClass = isActive ? ' active' : '';
// Use external URL for other services, current path for self // Use external URL for other services, current path for self
const href = isActive ? '#' : svc.url; const href = isActive ? '#' : svc.baseUrl;
return ` return `
<a href="${href}" class="service-link${activeClass}" onclick="event.preventDefault(); ${isActive ? '' : `window.location.href='${svc.url}'`}"> <a href="${href}" class="service-link${activeClass}" onclick="event.preventDefault(); ${isActive ? '' : `window.location.href='${svc.baseUrl}'`}">
<span class="service-icon">${icon}</span> <span class="service-icon">${icon}</span>
<span>${formatServiceName(svc.name)}</span> <span>${formatServiceName(svc.name)}</span>
<span class="service-status"></span> <span class="service-status"></span>