From 75c1b62a34c5ed6a63815f21a6a449c9fd2fb320 Mon Sep 17 00:00:00 2001 From: Pierre Date: Mon, 2 Feb 2026 00:03:11 +0100 Subject: [PATCH] bugfix on service discovery --- README.md | 2 - src/server.py | 39 +++----- src/storage/leader_client.py | 125 ++++++++++++++++---------- src/storage/service_discovery.py | 150 ++++++++++++++++++++----------- www/index.html | 119 +----------------------- 5 files changed, 192 insertions(+), 243 deletions(-) diff --git a/README.md b/README.md index 0fa9248..098ce3a 100644 --- a/README.md +++ b/README.md @@ -161,9 +161,7 @@ python server.py | Endpoint | Method | Description | |----------|--------|-------------| | `/api/stats` | GET | Library statistics | -| `/api/storage-status` | GET | KV connection status | | `/api/library` | GET | Full library list | -| `/api/library/refresh` | POST | Refresh library from KV | ### Stremio Protocol | Endpoint | Method | Description | diff --git a/src/server.py b/src/server.py index a8cd96a..c6310fd 100644 --- a/src/server.py +++ b/src/server.py @@ -13,7 +13,6 @@ Environment Variables: - MEDIA_DIR: Directory containing video files (default: /files/watch) - CACHE_DIR: Directory for transcoded segments (default: /data/cache) - PORT: HTTP server port (default: 7000) -- STORAGE_MODE: Storage backend 'leader' or 'redis' (default: redis) - META_CORE_PATH: Path to meta-core shared volume (default: /meta-core) - REDIS_URL: Redis connection URL (default: redis://localhost:6379) - REDIS_PREFIX: Redis key prefix (default: empty) @@ -183,20 +182,6 @@ class Handler(BaseHTTPRequestHandler): parsed = urlparse(self.path) path = unquote(parsed.path) - # Library refresh endpoint - if path == '/api/library/refresh': - try: - # Reconnect storage to refresh - storage = stremio.get_storage() - if hasattr(storage, 'disconnect'): - storage.disconnect() - stremio.init_storage() - # Reinitialize fileserver with new storage - fileserver.init(stremio.get_storage()) - return self.send_json({'status': 'ok', 'message': 'Library refreshed'}) - except Exception as e: - return self.send_json({'status': 'error', 'message': str(e)}, 500) - # Reset metrics if path == '/transcode/reset-metrics': transcoder.reset_metrics() @@ -235,11 +220,6 @@ class Handler(BaseHTTPRequestHandler): # === Dashboard API === - # Storage status - if path == '/api/storage-status': - storage = stremio.get_storage() - return self.send_json(storage.get_status() if storage else {'connected': False}) - # Library stats if path == '/api/stats': stats = stremio.get_library_stats() @@ -607,17 +587,24 @@ class Handler(BaseHTTPRequestHandler): all_services = sd.discover_all_services() for svc in all_services: - # Build dashboard URL from API URL - api_url = svc.get('api', '') + # 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.get('name', 'Unknown'), + '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: print(f"[Server] Error discovering services: {e}") @@ -631,9 +618,9 @@ class Handler(BaseHTTPRequestHandler): info = discovery.get_leader_info() if info: leader_info = { - 'host': info.host if hasattr(info, 'host') else info.get('host', ''), - 'api': info.api if hasattr(info, 'api') else info.get('api', ''), - 'http': info.http if hasattr(info, 'http') else info.get('http', ''), + '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}") diff --git a/src/storage/leader_client.py b/src/storage/leader_client.py index 021114f..b391130 100644 --- a/src/storage/leader_client.py +++ b/src/storage/leader_client.py @@ -98,32 +98,88 @@ class LeaderClient: 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.""" + # URL caching + self._cached_urls: Optional[URLsResponse] = None + self._urls_cache_time: float = 0 + self._urls_cache_ttl: float = 5.0 # 5 seconds + + def _get_api_url_from_file(self) -> Optional[str]: + """Read API URL from file (plain text format).""" try: with open(self.info_file_path, 'r') as f: content = f.read() + return content.strip() or None + except FileNotFoundError: + print(f"[LeaderClient] Leader info file not found: {self.info_file_path}") + return None + except Exception as e: + print(f"[LeaderClient] Error reading API URL from file: {e}") + return None - data = json.loads(content) + def _fetch_urls(self, api_url: str) -> Optional[URLsResponse]: + """Fetch URLs from meta-core /urls API with caching.""" + import time + # Check cache + now = time.time() + if self._cached_urls and (now - self._urls_cache_time) < self._urls_cache_ttl: + return self._cached_urls + + 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 fetch URLs: {response.status}") + return None + + data = json.loads(response.read().decode()) + + self._cached_urls = 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) + ) + self._urls_cache_time = now + return self._cached_urls + + except URLError as e: + print(f"[LeaderClient] Error calling /urls API: {e}") + return None + except Exception as e: + print(f"[LeaderClient] Error fetching URLs: {e}") + return None + + def get_leader_info(self) -> Optional[LeaderLockInfo]: + """Read leader info from file and /urls API.""" + try: + # Read API URL from file (plain text) + api_url = self._get_api_url_from_file() + if not api_url: + return None + + # Fetch full info from /urls API + urls = self._fetch_urls(api_url) + if not urls: + return None + + # Convert URLsResponse to LeaderLockInfo + import time 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) + hostname=urls.hostname, + base_url=urls.base_url, + api_url=urls.api_url, + redis_url=urls.redis_url, + webdav_url=urls.webdav_url, + timestamp=int(time.time() * 1000), + pid=0 # Unknown for remote leader ) 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 @@ -148,40 +204,15 @@ class LeaderClient: # First try using configured meta_core_url api_url = self.meta_core_url - # Fall back to reading from leader info + # Fall back to reading from file if not api_url: - info = self.get_leader_info() - api_url = info.api_url if info else None + api_url = self._get_api_url_from_file() 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 + return self._fetch_urls(api_url) def wait_for_leader(self, timeout_ms: int = 30000) -> LeaderLockInfo: """Wait for leader info to be available.""" @@ -232,7 +263,11 @@ class LeaderClient: def _on_file_change(self) -> None: """Handle file change event.""" - print("[LeaderClient] Leader info changed, reloading...") + print("[LeaderClient] Leader info changed, invalidating cache...") + # Invalidate cache to force fresh API call + self._cached_urls = None + self._urls_cache_time = 0 + self.get_leader_info() self._notify_change() diff --git a/src/storage/service_discovery.py b/src/storage/service_discovery.py index 3296f88..43a8799 100644 --- a/src/storage/service_discovery.py +++ b/src/storage/service_discovery.py @@ -1,8 +1,8 @@ """ Service Discovery for meta-stremio. -Python implementation of the service discovery pattern used in meta-sort. -Each service registers itself in /meta-core/services/{service-name}.json +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. Features: @@ -18,7 +18,7 @@ import json import socket import threading import time -from dataclasses import dataclass, field, asdict +from dataclasses import dataclass, asdict from datetime import datetime from pathlib import Path from typing import List, Optional, Dict, Callable @@ -35,17 +35,23 @@ BASE_URL = os.environ.get('BASE_URL', '') @dataclass class ServiceInfo: - """Service registration info stored in JSON file.""" + """ + Service registration info stored in JSON file. + + Simplified format matching TypeScript ServiceInfo interface: + - name: Service name (e.g., 'meta-stremio') + - hostname: Container/host hostname + - baseUrl: Base URL for the service (e.g., 'http://localhost:8182') + - status: Current status ('running' | 'stale' | 'stopped') + - lastHeartbeat: ISO timestamp of last heartbeat + - role: Optional role ('leader' | 'follower') - only used by meta-core + """ name: str - version: str - api: str # HTTP API URL (e.g., http://10.0.1.50:7000) - status: str = 'starting' # starting | running | stopping | stopped - pid: int = 0 - hostname: str = '' - startedAt: str = '' + hostname: str + baseUrl: str + status: str = 'running' # running | stale | stopped lastHeartbeat: str = '' - capabilities: List[str] = field(default_factory=list) - endpoints: Dict[str, str] = field(default_factory=dict) + role: Optional[str] = None # "leader", "follower", or None (for non-meta-core services) def to_dict(self) -> dict: """Convert to dictionary for JSON serialization.""" @@ -56,7 +62,7 @@ class ServiceDiscovery: """ Service Discovery using shared filesystem. - Each service registers itself in /meta-core/services/{service-name}.json + Each service registers itself in /meta-core/services/{service-name}-{hostname}.json and can discover other registered services. """ @@ -74,21 +80,20 @@ class ServiceDiscovery: ): self.meta_core_path = meta_core_path or META_CORE_PATH self.service_name = service_name or SERVICE_NAME - self.version = version or SERVICE_VERSION # Use base_url if provided, then api_url, then env BASE_URL, then auto-detect - self.api_url = base_url or api_url or BASE_URL or self._get_default_api_url() - self.capabilities = capabilities or ['streaming', 'transcoding', 'hls'] - self.endpoints = endpoints or { - 'health': '/health', - 'manifest': '/manifest.json', - 'dashboard': '/', - 'api': '/api', - } + self.base_url = base_url or api_url or BASE_URL or self._get_default_api_url() self.heartbeat_interval = heartbeat_interval self.stale_threshold = stale_threshold + # Keep for backwards compatibility but not used in simplified format + self._version = version or SERVICE_VERSION + self._capabilities = capabilities or [] + self._endpoints = endpoints or {} + + self.current_hostname = socket.gethostname() self.services_dir = os.path.join(self.meta_core_path, 'services') - self.service_file_path = os.path.join(self.services_dir, f'{self.service_name}.json') + # Use hostname-based file naming like TypeScript services + self.service_file_path = os.path.join(self.services_dir, f'{self.service_name}-{self.current_hostname}.json') self._heartbeat_thread: Optional[threading.Thread] = None self._stop_event = threading.Event() @@ -116,20 +121,14 @@ class ServiceDiscovery: """Ensure services directory exists.""" Path(self.services_dir).mkdir(parents=True, exist_ok=True) - def _build_service_info(self, status: str = 'starting') -> ServiceInfo: - """Build service info for this service.""" - now = datetime.utcnow().isoformat() + 'Z' + def _build_service_info(self, status: str = 'running') -> ServiceInfo: + """Build service info for this service (simplified format).""" return ServiceInfo( name=self.service_name, - version=self.version, - api=self.api_url, + hostname=self.current_hostname, + baseUrl=self.base_url, status=status, - pid=os.getpid(), - hostname=socket.gethostname(), - startedAt=now, - lastHeartbeat=now, - capabilities=self.capabilities, - endpoints=self.endpoints + lastHeartbeat=datetime.utcnow().isoformat() + 'Z' ) def register(self) -> None: @@ -236,22 +235,34 @@ class ServiceDiscovery: return True def discover_service(self, name: str) -> Optional[dict]: - """Discover a service by name.""" - service_path = os.path.join(self.services_dir, f'{name}.json') - + """ + 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: - with open(service_path, 'r') as f: - info = json.load(f) - - # Check if service is alive - if self._is_service_stale(info): - print(f'[ServiceDiscovery] Service {name} appears stale') + if not os.path.exists(self.services_dir): return None - return info + # 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 - except FileNotFoundError: return None + except Exception as e: print(f'[ServiceDiscovery] Error discovering {name}: {e}') return None @@ -259,6 +270,8 @@ class ServiceDiscovery: 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): @@ -268,11 +281,40 @@ class ServiceDiscovery: if not filename.endswith('.json'): continue - name = filename.replace('.json', '') - service = self.discover_service(name) + filepath = os.path.join(self.services_dir, filename) + try: + with open(filepath, 'r') as f: + info = json.load(f) - if service and service.get('status') == 'running': - services.append(service) + # 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}') @@ -285,8 +327,12 @@ class ServiceDiscovery: if not service: return False - health_endpoint = service.get('endpoints', {}).get('health', '/health') - url = f"{service['api']}{health_endpoint}" + # 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: @@ -312,7 +358,7 @@ class ServiceDiscovery: self.register() self.update_status('running') self.start_heartbeat() - print(f'[ServiceDiscovery] Started {self.service_name} at {self.api_url}') + print(f'[ServiceDiscovery] 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') diff --git a/www/index.html b/www/index.html index b9194cd..1137608 100644 --- a/www/index.html +++ b/www/index.html @@ -276,25 +276,6 @@ font-size: 0.85rem; } - /* Storage info */ - .storage-info { - display: flex; - flex-direction: column; - gap: 0.5rem; - } - - .storage-row { - display: flex; - justify-content: space-between; - align-items: center; - } - - .storage-type { - font-size: 1.1rem; - font-weight: 600; - color: var(--accent-primary); - } - /* Footer */ footer { text-align: center; @@ -421,33 +402,6 @@ - -
-

💾 Storage Status

-
-
- Loading... - - - Checking... - -
- -
- Prefix - meta-sort: -
-
- -
-
-
-

📚 Library

@@ -538,50 +492,6 @@ }); } - async function fetchStorageStatus() { - try { - const res = await fetch('/api/storage-status'); - const data = await res.json(); - - const statusEl = document.getElementById('storageStatus'); - const statusText = document.getElementById('storageStatusText'); - const typeEl = document.getElementById('storageType'); - - // Display storage type - const typeNames = { - 'leader': 'Leader Discovery', - 'redis': 'Redis Direct' - }; - typeEl.textContent = typeNames[data.type] || data.type; - - if (data.connected) { - statusEl.className = 'status status-connected'; - statusText.textContent = 'Connected'; - } else { - statusEl.className = 'status status-disconnected'; - statusText.textContent = 'Disconnected'; - } - - // Show leader info for leader mode - const redisUrlRow = document.getElementById('redisUrlRow'); - if (data.leader && data.leader.api) { - redisUrlRow.style.display = 'flex'; - document.getElementById('redisUrl').textContent = data.leader.host + ' (' + data.leader.api.replace('redis://', '') + ')'; - } else if (data.type === 'redis' && data.url) { - redisUrlRow.style.display = 'flex'; - document.getElementById('redisUrl').textContent = data.url.replace(/\/\/[^:]+:[^@]+@/, '//***:***@'); - } else { - redisUrlRow.style.display = 'none'; - } - - if (data.prefix) { - document.getElementById('storagePrefix').textContent = data.prefix; - } - } catch (e) { - console.error('Failed to fetch storage status:', e); - } - } - async function fetchStats() { try { const res = await fetch('/api/stats'); @@ -637,39 +547,13 @@ } } - async function refreshLibrary() { - try { - const btn = event.target.closest('.btn'); - btn.disabled = true; - btn.innerHTML = '⏳ Refreshing...'; - - await fetch('/api/library/refresh', { method: 'POST' }); - - await fetchStats(); - await fetchStorageStatus(); - - btn.innerHTML = '✓ Refreshed!'; - setTimeout(() => { - btn.disabled = false; - btn.innerHTML = '🔄 Refresh Library'; - }, 2000); - } catch (e) { - console.error('Failed to refresh library:', e); - const btn = event.target.closest('.btn'); - btn.disabled = false; - btn.innerHTML = '❌ Failed'; - setTimeout(() => { - btn.innerHTML = '🔄 Refresh Library'; - }, 2000); - } - } - // Service icons mapping const serviceIcons = { 'meta-sort': '📁', 'meta-fuse': '🗂️', 'meta-stremio': '🎬', 'meta-orbit': '🌐', + 'meta-core': '⚙️', 'default': '📦' }; @@ -722,7 +606,6 @@ } // Initial fetch - fetchStorageStatus(); fetchStats(); fetchMetrics(); fetchServices();