feat: add leader instance identification with operational metadata

- Add instance metadata (port, PID, hostname) to Redis leader election
- Expose is_leader boolean field in /health endpoint for cluster monitoring
- Store JSON metadata in Redis instead of plain UUID for operational visibility
- Add startup logging of instance metadata
- Make /health endpoint async to support Redis leader status checks
- Update tests to async for health check endpoints
- Maintain backward compatibility with legacy UUID-only format

Closes #3838

Signed-off-by: Bogdan-Marius-Catanus <bogdan-marius.catanus@ibm.com>

Author: Bogdan-Marius-Catanus

mcpgateway/main.py

@@ -10317,28 +10317,40 @@ async def reset_metrics(entity: Optional[str] = None, entity_id: Optional[int] =
 # Healthcheck      #
 ####################
 @app.get("/health")
-def healthcheck(response: Response = None):
+async def healthcheck(response: Response = None):
     """
     Perform a basic health check to verify database connectivity.
+    Now includes leader status for cluster monitoring.
 
-    Sync function so FastAPI runs it in a threadpool, avoiding event loop blocking.
+    Async function to allow checking Redis leader status.
     Uses a dedicated session to avoid cross-thread issues and double-commit
     from get_db dependency. All DB operations happen in the same thread.
 
     Args:
         response: Optional response object used to attach runtime-mode headers.
 
     Returns:
-        A dictionary with the health status and optional error message.
+        A dictionary with the health status, leader status, and optional error message.
     """
     db = SessionLocal()
     try:
         db.execute(text("SELECT 1"))
         # Explicitly commit to release PgBouncer backend connection in transaction mode.
         db.commit()
+
+        # Check leader status using async method
+        is_leader = False
+        if hasattr(gateway_service, "is_leader"):
+            try:
+                is_leader = await gateway_service.is_leader()
+            except Exception:
+                # Fallback to False if leader check fails (e.g., in test environments)
+                is_leader = False
+
         if response is not None:
             _apply_runtime_mode_headers(response)
-        return {"status": "healthy", "mcp_runtime": _mcp_runtime_status_payload()}
+
+        return {"status": "healthy", "is_leader": is_leader, "mcp_runtime": _mcp_runtime_status_payload()}
     except Exception as e:
         # Rollback, then invalidate if rollback fails (mirrors get_db cleanup).
         try:
@@ -10352,7 +10364,8 @@ def healthcheck(response: Response = None):
         logger.error(error_message)
         if response is not None:
             _apply_runtime_mode_headers(response)
-        return {"status": "unhealthy", "error": error_message, "mcp_runtime": _mcp_runtime_status_payload()}
+
+        return {"status": "unhealthy", "is_leader": False, "error": error_message, "mcp_runtime": _mcp_runtime_status_payload()}
     finally:
         db.close()
 

mcpgateway/services/gateway_service.py

@@ -45,9 +45,11 @@
 import asyncio
 import binascii
 from datetime import datetime, timezone
+import json
 import logging
 import mimetypes
 import os
+import socket
 import ssl
 import tempfile
 import time
@@ -471,15 +473,17 @@ def __init__(self) -> None:
 
         # Leader election settings from config
         if self.redis_url and REDIS_AVAILABLE:
-            self._instance_id = str(uuid.uuid4())  # Unique ID for this process
+            # Metadata for this instance
+            self._instance_metadata = {"instance_id": str(uuid.uuid4()), "port": settings.port, "pid": os.getpid(), "hostname": socket.gethostname()}
+            self._instance_id = self._instance_metadata["instance_id"]
             self._leader_key = settings.redis_leader_key
             self._leader_ttl = settings.redis_leader_ttl
             self._leader_heartbeat_interval = settings.redis_leader_heartbeat_interval
             self._leader_heartbeat_task: Optional[asyncio.Task] = None
             self._follower_election_task: Optional[asyncio.Task] = None
 
-            # Log instance mapping for debugging
-            logger.info(f"Instance started: instance_id={self._instance_id}, port={settings.port}, pid={os.getpid()}")
+            # Log instance mapping
+            logger.info(f"Instance started: {json.dumps(self._instance_metadata)}")
 
         # Always initialize file lock as fallback (used if Redis connection fails at runtime)
         if settings.cache_type != "none":
@@ -587,7 +591,8 @@ async def initialize(self) -> None:
             except Exception as e:
                 raise ConnectionError(f"Redis ping failed: {e}") from e
 
-            is_leader = await self._redis_client.set(self._leader_key, self._instance_id, ex=self._leader_ttl, nx=True)
+            # Store all instance metadate in redis
+            is_leader = await self._redis_client.set(self._leader_key, json.dumps(self._instance_metadata), ex=self._leader_ttl, nx=True)
             if is_leader:
                 logger.info("Acquired Redis leadership. Starting health check and heartbeat tasks.")
                 self._health_check_task = asyncio.create_task(self._run_health_checks(user_email))
@@ -653,7 +658,7 @@ async def shutdown(self) -> None:
                     return 0
                 end
                 """
-                result = await self._redis_client.eval(release_script, 1, self._leader_key, self._instance_id)
+                result = await self._redis_client.eval(release_script, 1, self._leader_key, json.dumps(self._instance_metadata))
                 if result:
                     logger.info("Released Redis leadership on shutdown")
             except Exception as e:
@@ -3941,8 +3946,18 @@ async def _run_leader_heartbeat(self) -> None:
                     continue
 
                 # Check if we're still the leader
-                current_leader = await self._redis_client.get(self._leader_key)
-                if current_leader != self._instance_id:
+                current_leader_raw = await self._redis_client.get(self._leader_key)
+                if current_leader_raw:
+                    try:
+                        current_leader_data = json.loads(current_leader_raw)
+                        current_leader_id = current_leader_data.get("instance_id")
+                    except (json.JSONDecodeError, AttributeError):
+                        # Fallback for old UUID-only format
+                        current_leader_id = current_leader_raw
+                else:
+                    current_leader_id = None
+
+                if current_leader_id != self._instance_id:
                     logger.info("Lost Redis leadership, stopping heartbeat")
                     self._start_follower_election()
                     return
@@ -3985,7 +4000,7 @@ async def _run_follower_election(self, user_email: str) -> None:
                     continue
 
                 # Attempt to acquire leadership
-                is_leader = await self._redis_client.set(self._leader_key, self._instance_id, ex=self._leader_ttl, nx=True)
+                is_leader = await self._redis_client.set(self._leader_key, json.dumps(self._instance_metadata), ex=self._leader_ttl, nx=True)
 
                 if is_leader:
                     logger.info("Acquired Redis leadership via follower election. Starting health check and heartbeat.")
@@ -4034,8 +4049,18 @@ async def _run_health_checks(self, user_email: str) -> None:
                 if self._redis_client and settings.cache_type == "redis":
                     # Redis-based leader check (async, decode_responses=True returns strings)
                     # Note: Leader key TTL refresh is handled by _run_leader_heartbeat task
-                    current_leader = await self._redis_client.get(self._leader_key)
-                    if current_leader != self._instance_id:
+                    current_leader_raw = await self._redis_client.get(self._leader_key)
+                    if current_leader_raw:
+                        try:
+                            current_leader_data = json.loads(current_leader_raw)
+                            current_leader_id = current_leader_data.get("instance_id")
+                        except (json.JSONDecodeError, AttributeError):
+                            # Fallback for old UUID-only format
+                            current_leader_id = current_leader_raw
+                    else:
+                        current_leader_id = None
+
+                    if current_leader_id != self._instance_id:
                         return
 
                     # Run health checks
@@ -4087,6 +4112,67 @@ async def _run_health_checks(self, user_email: str) -> None:
                 logger.error(f"Unexpected error in health check loop: {str(e)}")
                 await asyncio.sleep(self._health_check_interval)
 
+    def is_leader_sync(self) -> bool:
+        """Check if this instance is the current leader (synchronous version for health checks).
+
+        Returns:
+            bool: True if this instance holds the leader lock, False otherwise.
+        """
+        if not self._redis_client or not hasattr(self, "_leader_key"):
+            # Fallback to file lock for non-Redis setups
+            return True
+
+        try:
+            # Use sync Redis client method if available
+            import redis
+            if isinstance(self._redis_client, redis.asyncio.Redis):
+                # For async Redis client, we can't do sync call - return True as fallback
+                return True
+            
+            # For sync Redis client (shouldn't happen in current setup, but safe fallback)
+            current_leader_raw = self._redis_client.get(self._leader_key)
+            if not current_leader_raw:
+                return False
+
+            try:
+                current_leader_data = json.loads(current_leader_raw)
+                current_leader_id = current_leader_data.get("instance_id")
+            except (json.JSONDecodeError, AttributeError):
+                # Fallback for old UUID-only format
+                current_leader_id = current_leader_raw
+
+            return current_leader_id == self._instance_id
+        except Exception as e:
+            logger.warning(f"Error checking leader status: {e}")
+            return True  # Fail open for health checks
+
+    async def is_leader(self) -> bool:
+        """Check if this instance is the current leader (async version).
+
+        Returns:
+            bool: True if this instance holds the leader lock, False otherwise.
+        """
+        if not self._redis_client or not hasattr(self, "_leader_key"):
+            # Fallback to file lock for non-Redis setups
+            return True
+
+        try:
+            current_leader_raw = await self._redis_client.get(self._leader_key)
+            if not current_leader_raw:
+                return False
+
+            try:
+                current_leader_data = json.loads(current_leader_raw)
+                current_leader_id = current_leader_data.get("instance_id")
+            except (json.JSONDecodeError, AttributeError):
+                # Fallback for old UUID-only format
+                current_leader_id = current_leader_raw
+
+            return current_leader_id == self._instance_id
+        except Exception as e:
+            logger.warning(f"Error checking leader status: {e}")
+            return False
+
     def _get_auth_headers(self) -> Dict[str, str]:
         """Get default headers for gateway requests (no authentication).
 

tests/unit/mcpgateway/test_main.py

@@ -457,15 +457,19 @@ def test_health_check(self, test_client):
         """Test the basic health check endpoint."""
         response = test_client.get("/health")
         assert response.status_code == 200
-        assert response.json()["status"] == "healthy"
+        data = response.json()
+        assert data["status"] == "healthy"
+        assert "is_leader" in data
+        assert isinstance(data["is_leader"], bool)
 
     def test_ready_check(self, test_client):
         """Test the readiness check endpoint."""
         response = test_client.get("/ready")
         assert response.status_code == 200
         assert response.json()["status"] == "ready"
 
-    def test_health_check_db_error(self):
+    @pytest.mark.asyncio
+    async def test_health_check_db_error(self):
         """Test health check error path with rollback failure."""
         # First-Party
         from mcpgateway import main as mcpgateway_main
@@ -491,7 +495,7 @@ def close(self):
 
         session = DummySession()
         with patch("mcpgateway.main.SessionLocal", return_value=session):
-            response = mcpgateway_main.healthcheck()
+            response = await mcpgateway_main.healthcheck()
         assert response["status"] == "unhealthy"
         assert session.invalidate_called is True
 

tests/unit/mcpgateway/test_main_extended.py

@@ -9657,7 +9657,8 @@ def close(self):  # noqa: D401 - test helper
         assert sess.invalidated is True
         assert sess.closed is True
 
-    def test_healthcheck_invalidate_failure_is_best_effort(self, monkeypatch):
+    @pytest.mark.asyncio
+    async def test_healthcheck_invalidate_failure_is_best_effort(self, monkeypatch):
         # First-Party
         import mcpgateway.main as main_mod
 
@@ -9683,12 +9684,13 @@ def close(self):
         sess = FakeSession()
         monkeypatch.setattr(main_mod, "SessionLocal", lambda: sess)
 
-        result = main_mod.healthcheck()
+        result = await main_mod.healthcheck()
         assert result["status"] == "unhealthy"
         assert "db down" in result["error"]
         assert sess.closed is True
 
-    def test_healthcheck_reports_runtime_mode_and_headers(self, monkeypatch):
+    @pytest.mark.asyncio
+    async def test_healthcheck_reports_runtime_mode_and_headers(self, monkeypatch):
         # First-Party
         import mcpgateway.main as main_mod
 
@@ -9708,7 +9710,7 @@ def close(self):
         monkeypatch.setattr(main_mod.settings, "experimental_rust_mcp_runtime_enabled", False)
 
         response = FastAPIResponse()
-        result = main_mod.healthcheck(response)
+        result = await main_mod.healthcheck(response)
 
         assert result["status"] == "healthy"
         assert result["mcp_runtime"]["mode"] == "python-rust-built-disabled"
@@ -9863,7 +9865,8 @@ def test_runtime_status_payload_reports_python_mount_without_session_auth_reuse(
         assert payload["rust_affinity_core_enabled"] is False
         assert payload["session_auth_reuse_mode"] == "python"
 
-    def test_healthcheck_unhealthy_applies_runtime_headers(self, monkeypatch):
+    @pytest.mark.asyncio
+    async def test_healthcheck_unhealthy_applies_runtime_headers(self, monkeypatch):
         # First-Party
         import mcpgateway.main as main_mod
 
@@ -9881,7 +9884,7 @@ def close(self):
         monkeypatch.setattr(main_mod.settings, "experimental_rust_mcp_runtime_enabled", True)
 
         response = FastAPIResponse()
-        result = main_mod.healthcheck(response)
+        result = await main_mod.healthcheck(response)
 
         assert result["status"] == "unhealthy"
         assert response.headers["x-contextforge-mcp-runtime-mode"] == "rust-managed"