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)}