cluster_status
Check CockroachDB cluster health status including node count and live nodes for monitoring and troubleshooting.
Instructions
Get the health status of the CockroachDB cluster.
Returns:
Cluster health including node count and live nodes.Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- src/cockroachdb_mcp/server.py:49-59 (handler)Handler and registration for the 'cluster_status' tool. Thin wrapper that delegates to cluster.cluster_health() with error handling.
@mcp.tool() async def cluster_status() -> dict[str, Any]: """Get the health status of the CockroachDB cluster. Returns: Cluster health including node count and live nodes. """ try: return await cluster.cluster_health() except Exception as e: return {"status": "error", "error": str(e)} - Core helper function implementing cluster health check by querying cluster organization, version, total nodes, live nodes, and determining status.
async def cluster_health() -> dict[str, Any]: """Get the health status of the CockroachDB cluster. Returns: Cluster health information. """ conn = await connection_manager.ensure_connected() try: result: dict[str, Any] = {"status": "healthy"} async with conn.cursor() as cur: # Get cluster settings await cur.execute("SHOW CLUSTER SETTING cluster.organization") org_row = await cur.fetchone() result["organization"] = org_row.get("cluster.organization") if org_row else None # Get version await cur.execute("SELECT version()") version_row = await cur.fetchone() result["version"] = version_row.get("version") if version_row else None # Get node count await cur.execute("SELECT count(*) as node_count FROM crdb_internal.gossip_nodes") node_row = await cur.fetchone() result["node_count"] = node_row.get("node_count") if node_row else 0 # Get live node count await cur.execute(""" SELECT count(*) as live_nodes FROM crdb_internal.gossip_nodes WHERE is_live = true """) live_row = await cur.fetchone() result["live_nodes"] = live_row.get("live_nodes") if live_row else 0 # Check if cluster is healthy if result["live_nodes"] < result["node_count"]: result["status"] = "degraded" result["message"] = ( f"{result['node_count'] - result['live_nodes']} node(s) are not live" ) return result except Exception as e: return {"status": "error", "error": str(e)}