This is an automated email from the ASF dual-hosted git repository.
FreeOnePlus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris-mcp-server.git
The following commit(s) were added to refs/heads/master by this push:
new 8948b3e feat: add Cluster domain runtime (#176)
8948b3e is described below
commit 8948b3e5430afc2b11269f48c91c9dbc4ecc824b
Author: Yijia Su <[email protected]>
AuthorDate: Fri Jul 31 17:37:02 2026 +0800
feat: add Cluster domain runtime (#176)
Add the read-only Cluster domain runtime, route-aware capability probes,
bounded evidence handling, and real Doris transport coverage.
---
CHANGELOG.md | 8 +
doris_mcp_server/tools/capability_detector.py | 428 ++++++++-
doris_mcp_server/tools/capability_registry.py | 24 +-
doris_mcp_server/tools/cluster_handlers.py | 198 ++++
doris_mcp_server/tools/domain_dispatcher.py | 47 +-
doris_mcp_server/tools/doris_feature_matrix.py | 51 +-
doris_mcp_server/tools/tools_manager.py | 139 +--
doris_mcp_server/utils/cluster_runtime.py | 1185 ++++++++++++++++++++++++
doris_mcp_server/utils/db.py | 17 +
test/integration/test_real_doris_transports.py | 143 +++
test/tools/test_capability_detector.py | 139 +++
test/tools/test_capability_registry.py | 70 ++
test/tools/test_domain_dispatcher.py | 70 +-
test/tools/test_domain_manifest.py | 10 +-
test/tools/test_manager_routing_boundaries.py | 139 +--
test/utils/test_cluster_runtime.py | 313 +++++++
test/utils/test_doris_user_pool_manager.py | 2 +-
17 files changed, 2697 insertions(+), 286 deletions(-)
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 3e8daf8..75cd49e 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -48,6 +48,9 @@ under **Unreleased** until a new version is selected and
published.
- Bounded query result streaming with deployment and absolute ceilings for
rows, serialized bytes, and execution time, plus cancellation-safe database
connection disposal.
+- A read-only Cluster domain with eleven capability-gated children for node
+ inventory, active tasks, metrics, memory, cache, compaction, workload and
+ compute groups, recorded resource growth, and sanitized runtime evidence.
- Real Doris process tests covering Streamable HTTP and stdio.
### Changed
@@ -87,6 +90,11 @@ under **Unreleased** until a new version is selected and
published.
### Fixed
+- Released failed capability-probe connections from their captured owner pools
+ so a single-connection route cannot starve subsequent domain calls.
+- Excluded explicitly dead Doris components from active version gating while
+ preserving them in node inventory, and kept live runtime manifests within
+ the 16 KiB domain budget.
- Preserved service availability after malformed requests, unknown methods,
header mismatches, unsupported versions, and missing capabilities.
- Kept static-token Doris pools usable after repeated query timeouts, and
diff --git a/doris_mcp_server/tools/capability_detector.py
b/doris_mcp_server/tools/capability_detector.py
index e1d5bd3..7d78779 100644
--- a/doris_mcp_server/tools/capability_detector.py
+++ b/doris_mcp_server/tools/capability_detector.py
@@ -34,6 +34,7 @@ from ..utils.doris_http_client import (
DorisHTTPClient,
DorisHTTPPolicyError,
DorisHTTPRequestError,
+ DorisHTTPResponse,
configured_fe_http_hosts,
database_config_for_request,
)
@@ -177,6 +178,44 @@ _DOMAIN_PROBES: Mapping[str, tuple[tuple[str, tuple[str,
...]], ...]] = {
("audit_log_readable",),
),
),
+ "doris_cluster": (
+ (
+ 'SHOW PROC "/current_queries"',
+ (
+ "legacy_task_views_readable",
+ "unified_task_progress_readable",
+ ),
+ ),
+ (
+ (
+ "SELECT BE_ID, METRIC_NAME "
+ "FROM information_schema.file_cache_statistics LIMIT 1"
+ ),
+ (
+ "information_schema.file_cache_statistics",
+ "file_cache_metrics_readable",
+ ),
+ ),
+ (
+ ("SELECT * FROM information_schema.doris_be_compaction_tasks LIMIT
1"),
+ (
+ "information_schema.doris_be_compaction_tasks",
+ "compaction_system_table_or_http_api",
+ ),
+ ),
+ (
+ "SHOW WORKLOAD GROUPS",
+ ("workload_group_metadata_and_metrics_readable",),
+ ),
+ (
+ "SHOW COMPUTE GROUPS",
+ ("compute_group_metadata_readable",),
+ ),
+ (
+ ("SELECT `time` FROM internal.__internal_schema.audit_log LIMIT
1"),
+ ("metrics_history_readable",),
+ ),
+ ),
}
@@ -298,6 +337,8 @@ class DorisCapabilityDetector:
probes["profile_or_audit_readable"] =
_combine_query_evidence_probe(
probes
)
+ elif domain_name == "doris_cluster":
+ probes.update(await
self._safe_probe_cluster_services(auth_context))
completed_route = self.route_identity(auth_context)
if completed_route.fingerprint != base.route.fingerprint:
raise CapabilityRouteChangedError(
@@ -716,6 +757,111 @@ class DorisCapabilityDetector:
endpoint_reason="ADBC_RUNTIME_PROBE_FAILED",
)
+ async def _safe_probe_cluster_services(
+ self,
+ auth_context: Any | None,
+ ) -> dict[str, CapabilityProbeEvidence]:
+ try:
+ async with asyncio.timeout(self._optional_probe_timeout()):
+ return await self._probe_cluster_services(auth_context)
+ except TimeoutError:
+ return _cluster_http_probe_failure(
+ status=CapabilityProbeStatus.DEGRADED,
+ reason_code="CLUSTER_HTTP_PROBE_TIMED_OUT",
+ )
+ except DorisHTTPPolicyError:
+ return _cluster_http_probe_failure(
+ status=CapabilityProbeStatus.MISCONFIGURED,
+ reason_code="CLUSTER_HTTP_ENDPOINT_MISCONFIGURED",
+ )
+ except DorisHTTPRequestError:
+ return _cluster_http_probe_failure(
+ status=CapabilityProbeStatus.DEGRADED,
+ reason_code="CLUSTER_HTTP_ENDPOINT_UNREACHABLE",
+ )
+ except Exception:
+ return _cluster_http_probe_failure(
+ status=CapabilityProbeStatus.UNKNOWN,
+ reason_code="CLUSTER_HTTP_PROBE_FAILED",
+ )
+
+ async def _probe_cluster_services(
+ self,
+ auth_context: Any | None,
+ ) -> dict[str, CapabilityProbeEvidence]:
+ if getattr(auth_context, "auth_method", "") == "doris_oauth":
+ return _cluster_http_probe_failure(
+ status=CapabilityProbeStatus.MISCONFIGURED,
+ reason_code="CLUSTER_HTTP_CREDENTIAL_ROUTE_UNAVAILABLE",
+ )
+ config_resolver = getattr(
+ self._connection_manager,
+ "get_database_config_for_auth_context",
+ None,
+ )
+ db_config = (
+ config_resolver(auth_context)
+ if callable(config_resolver)
+ else database_config_for_request(self._connection_manager)
+ )
+ client = DorisHTTPClient.from_database_config(db_config)
+ fe_hosts = configured_fe_http_hosts(db_config)
+ raw_be_hosts = getattr(db_config, "be_hosts", []) or []
+ if not isinstance(raw_be_hosts, list) or any(
+ not isinstance(host, str) for host in raw_be_hosts
+ ):
+ return _cluster_http_probe_failure(
+ status=CapabilityProbeStatus.MISCONFIGURED,
+ reason_code="BE_METRICS_ENDPOINT_MISCONFIGURED",
+ )
+ be_hosts = tuple(dict.fromkeys(raw_be_hosts))
+ fe_response, be_response = await asyncio.gather(
+ client.get_first_available(
+ role="fe",
+ hosts=fe_hosts,
+ port=db_config.fe_http_port,
+ path="/metrics",
+ headers={"Accept": "text/plain"},
+ ),
+ (
+ client.get_first_available(
+ role="be",
+ hosts=be_hosts,
+ port=db_config.be_webserver_port,
+ path="/metrics",
+ headers={"Accept": "text/plain"},
+ )
+ if be_hosts
+ else _missing_be_metrics_response()
+ ),
+ )
+ fe_status, fe_reason = _classify_metrics_response(
+ fe_response.status,
+ configured=True,
+ )
+ be_status, be_reason = _classify_metrics_response(
+ be_response.status,
+ configured=bool(be_hosts),
+ )
+ fe_names = (
+ _prometheus_metric_names(fe_response.text())
+ if fe_status is CapabilityProbeStatus.SUPPORTED
+ else frozenset()
+ )
+ be_names = (
+ _prometheus_metric_names(be_response.text())
+ if be_status is CapabilityProbeStatus.SUPPORTED
+ else frozenset()
+ )
+ return _cluster_http_probe_evidence(
+ fe_status=fe_status,
+ fe_reason=fe_reason,
+ be_status=be_status,
+ be_reason=be_reason,
+ fe_metric_names=fe_names,
+ be_metric_names=be_names,
+ )
+
def _optional_probe_timeout(self) -> float:
return max(
0.25,
@@ -748,6 +894,277 @@ def _classify_probe_error(
return CapabilityProbeStatus.UNKNOWN, "RUNTIME_PROBE_FAILED"
+async def _missing_be_metrics_response() -> DorisHTTPResponse:
+ return DorisHTTPResponse(
+ status=0,
+ headers={},
+ body=b"",
+ url="",
+ )
+
+
+def _classify_metrics_response(
+ status_code: int,
+ *,
+ configured: bool,
+) -> tuple[CapabilityProbeStatus, str]:
+ if not configured:
+ return (
+ CapabilityProbeStatus.MISCONFIGURED,
+ "METRICS_ENDPOINT_NOT_CONFIGURED",
+ )
+ if status_code == 200:
+ return CapabilityProbeStatus.SUPPORTED, "METRICS_ENDPOINT_READABLE"
+ if status_code in {401, 403}:
+ return CapabilityProbeStatus.UNKNOWN,
"METRICS_ENDPOINT_PERMISSION_DENIED"
+ if status_code in {502, 503, 504}:
+ return CapabilityProbeStatus.DEGRADED, "METRICS_ENDPOINT_UNREACHABLE"
+ return CapabilityProbeStatus.UNSUPPORTED, "METRICS_ENDPOINT_UNSUPPORTED"
+
+
+def _prometheus_metric_names(payload: str) -> frozenset[str]:
+ names: set[str] = set()
+ for raw_line in payload.splitlines():
+ line = raw_line.strip()
+ if not line or line.startswith("#"):
+ continue
+ token = line.split(maxsplit=1)[0]
+ name = token.split("{", 1)[0]
+ if name:
+ names.add(name)
+ if len(names) >= 10_000:
+ break
+ return frozenset(names)
+
+
+def _cluster_http_probe_failure(
+ *,
+ status: CapabilityProbeStatus,
+ reason_code: str,
+) -> dict[str, CapabilityProbeEvidence]:
+ probe_ids = (
+ "fe_metrics",
+ "be_metrics",
+ "metrics_endpoints_readable",
+ "memory_tracker_metrics_readable",
+ "legacy_compaction_status_readable",
+ "connection_limit_metrics",
+ "routine_load_metrics",
+ "file_cache_queue_metrics",
+ "enhanced_metrics_present",
+ "file_cache_queue_metrics_present",
+ "condition_cache",
+ "parquet_page_cache",
+ "advanced_cache_metrics_present",
+ )
+ return {
+ probe_id: CapabilityProbeEvidence(
+ probe_id=probe_id,
+ status=status,
+ reason_code=reason_code,
+ evidence_sources=("doris_http_metrics", "runtime_probe"),
+ )
+ for probe_id in probe_ids
+ }
+
+
+def _cluster_http_probe_evidence(
+ *,
+ fe_status: CapabilityProbeStatus,
+ fe_reason: str,
+ be_status: CapabilityProbeStatus,
+ be_reason: str,
+ fe_metric_names: frozenset[str],
+ be_metric_names: frozenset[str],
+) -> dict[str, CapabilityProbeEvidence]:
+ probes: dict[str, CapabilityProbeEvidence] = {
+ "fe_metrics": CapabilityProbeEvidence(
+ probe_id="fe_metrics",
+ status=fe_status,
+ reason_code=fe_reason,
+ evidence_sources=("doris_fe_metrics", "runtime_probe"),
+ ),
+ "be_metrics": CapabilityProbeEvidence(
+ probe_id="be_metrics",
+ status=be_status,
+ reason_code=be_reason,
+ evidence_sources=("doris_be_metrics", "runtime_probe"),
+ ),
+ }
+ combined_status, combined_reason = _combine_endpoint_status(
+ fe_status,
+ be_status,
+ )
+ probes["metrics_endpoints_readable"] = CapabilityProbeEvidence(
+ probe_id="metrics_endpoints_readable",
+ status=combined_status,
+ reason_code=combined_reason,
+ evidence_sources=(
+ "doris_fe_metrics",
+ "doris_be_metrics",
+ "runtime_probe",
+ ),
+ )
+
+ all_names = fe_metric_names | be_metric_names
+ memory_present = _has_metric_marker(
+ be_metric_names,
+ ("memory", "mem_tracker", "jemalloc"),
+ )
+ probes["memory_tracker_metrics_readable"] = _metric_presence_evidence(
+ "memory_tracker_metrics_readable",
+ endpoint_status=be_status,
+ present=memory_present,
+ present_reason="MEMORY_TRACKER_METRICS_READABLE",
+ absent_reason="MEMORY_TRACKER_METRICS_NOT_PRESENT",
+ source="doris_be_metrics",
+ )
+
+ compaction_present = _has_metric_marker(all_names, ("compaction",))
+ if compaction_present:
+ probes["legacy_compaction_status_readable"] = CapabilityProbeEvidence(
+ probe_id="legacy_compaction_status_readable",
+ status=CapabilityProbeStatus.DEGRADED,
+ reason_code="LEGACY_COMPACTION_SUMMARY_READABLE",
+ evidence_sources=("doris_metrics", "runtime_probe"),
+ )
+ else:
+ probes["legacy_compaction_status_readable"] =
_metric_presence_evidence(
+ "legacy_compaction_status_readable",
+ endpoint_status=combined_status,
+ present=False,
+ present_reason="LEGACY_COMPACTION_SUMMARY_READABLE",
+ absent_reason="LEGACY_COMPACTION_METRICS_NOT_PRESENT",
+ source="doris_metrics",
+ )
+
+ marker_contracts = {
+ "connection_limit_metrics": ("connection", "limit"),
+ "routine_load_metrics": ("routine_load",),
+ "file_cache_queue_metrics": ("file_cache", "queue"),
+ "condition_cache": ("condition_cache",),
+ "parquet_page_cache": ("parquet", "page_cache"),
+ }
+ for probe_id, markers in marker_contracts.items():
+ probes[probe_id] = _metric_presence_evidence(
+ probe_id,
+ endpoint_status=combined_status,
+ present=all(marker in " ".join(all_names) for marker in markers),
+ present_reason=f"{probe_id.upper()}_PRESENT",
+ absent_reason=f"{probe_id.upper()}_NOT_PRESENT",
+ source="doris_metrics",
+ )
+
+ queue_probe = probes["file_cache_queue_metrics"]
+ probes["file_cache_queue_metrics_present"] = CapabilityProbeEvidence(
+ probe_id="file_cache_queue_metrics_present",
+ status=queue_probe.status,
+ reason_code=queue_probe.reason_code,
+ evidence_sources=queue_probe.evidence_sources,
+ )
+ enhanced_candidates = (
+ probes["connection_limit_metrics"],
+ probes["routine_load_metrics"],
+ probes["file_cache_queue_metrics"],
+ )
+ probes["enhanced_metrics_present"] = _combine_metric_presence(
+ "enhanced_metrics_present",
+ enhanced_candidates,
+ supported_reason="ENHANCED_METRICS_PRESENT",
+ )
+ advanced_candidates = (
+ probes["condition_cache"],
+ probes["parquet_page_cache"],
+ )
+ probes["advanced_cache_metrics_present"] = _combine_metric_presence(
+ "advanced_cache_metrics_present",
+ advanced_candidates,
+ supported_reason="ADVANCED_CACHE_METRICS_PRESENT",
+ )
+ return probes
+
+
+def _combine_endpoint_status(
+ *statuses: CapabilityProbeStatus,
+) -> tuple[CapabilityProbeStatus, str]:
+ if all(status is CapabilityProbeStatus.SUPPORTED for status in statuses):
+ return (
+ CapabilityProbeStatus.SUPPORTED,
+ "FE_BE_METRICS_ENDPOINTS_READABLE",
+ )
+ for status in (
+ CapabilityProbeStatus.MISCONFIGURED,
+ CapabilityProbeStatus.DEGRADED,
+ CapabilityProbeStatus.UNKNOWN,
+ CapabilityProbeStatus.UNSUPPORTED,
+ ):
+ if status in statuses:
+ return status, f"FE_BE_METRICS_{status.value.upper()}"
+ return CapabilityProbeStatus.UNKNOWN, "FE_BE_METRICS_UNKNOWN"
+
+
+def _has_metric_marker(
+ names: frozenset[str],
+ markers: tuple[str, ...],
+) -> bool:
+ return any(any(marker in name.casefold() for marker in markers) for name
in names)
+
+
+def _metric_presence_evidence(
+ probe_id: str,
+ *,
+ endpoint_status: CapabilityProbeStatus,
+ present: bool,
+ present_reason: str,
+ absent_reason: str,
+ source: str,
+) -> CapabilityProbeEvidence:
+ if present:
+ status = CapabilityProbeStatus.SUPPORTED
+ reason = present_reason
+ elif endpoint_status is CapabilityProbeStatus.SUPPORTED:
+ status = CapabilityProbeStatus.UNSUPPORTED
+ reason = absent_reason
+ else:
+ status = endpoint_status
+ reason = absent_reason
+ return CapabilityProbeEvidence(
+ probe_id=probe_id,
+ status=status,
+ reason_code=reason,
+ evidence_sources=(source, "runtime_probe"),
+ )
+
+
+def _combine_metric_presence(
+ probe_id: str,
+ candidates: tuple[CapabilityProbeEvidence, ...],
+ *,
+ supported_reason: str,
+) -> CapabilityProbeEvidence:
+ if all(
+ candidate.status is CapabilityProbeStatus.SUPPORTED for candidate in
candidates
+ ):
+ status = CapabilityProbeStatus.SUPPORTED
+ reason = supported_reason
+ else:
+ status = next(
+ (
+ candidate.status
+ for candidate in candidates
+ if candidate.status is not CapabilityProbeStatus.SUPPORTED
+ ),
+ CapabilityProbeStatus.UNKNOWN,
+ )
+ reason = f"{probe_id.upper()}_INCOMPLETE"
+ return CapabilityProbeEvidence(
+ probe_id=probe_id,
+ status=status,
+ reason_code=reason,
+ evidence_sources=("doris_metrics", "runtime_probe"),
+ )
+
+
def _profile_api_code(payload: Mapping[str, Any] | None) -> int | None:
if payload is None or "code" not in payload:
return None
@@ -880,6 +1297,8 @@ def _frontend_versions(
) -> tuple[DorisVersion, tuple[DorisVersion, ...]]:
observed: list[tuple[DorisVersion, bool]] = []
for row in rows:
+ if not _component_is_active(row):
+ continue
version = _component_version(_row_value(row, "Version", "FeVersion"))
observed.append(
(
@@ -907,10 +1326,17 @@ def _backend_versions(
rows: Sequence[Mapping[str, Any]],
) -> tuple[DorisVersion, ...]:
return tuple(
- _component_version(_row_value(row, "Version", "BeVersion")) for row in
rows
+ _component_version(_row_value(row, "Version", "BeVersion"))
+ for row in rows
+ if _component_is_active(row)
)
+def _component_is_active(row: Mapping[str, Any]) -> bool:
+ alive = _row_value(row, "Alive")
+ return alive is None or str(alive).strip() == "" or _truthy(alive)
+
+
def _cluster_fingerprint(
route: DorisRouteIdentity,
frontend_rows: Sequence[Mapping[str, Any]],
diff --git a/doris_mcp_server/tools/capability_registry.py
b/doris_mcp_server/tools/capability_registry.py
index e495c49..df7ff8b 100644
--- a/doris_mcp_server/tools/capability_registry.py
+++ b/doris_mcp_server/tools/capability_registry.py
@@ -287,16 +287,10 @@ class CapabilityEvaluator:
limitations=limitations,
)
- if (
+ uncertified = (
version_result.certification_status
is not VersionCertificationStatus.CERTIFIED
- ):
- limitations = _ordered_unique(
- (
- *limitations,
- "The observed Doris patch is not certified by this
project.",
- )
- )
+ )
if snapshot.stale:
degraded = True
limitations = _ordered_unique(
@@ -313,9 +307,17 @@ class CapabilityEvaluator:
),
callable=True,
reason_code=(
- "CAPABILITY_VERIFIED_DEGRADED"
- if degraded
- else "CAPABILITY_VERIFIED"
+ "CAPABILITY_VERIFIED_DEGRADED_UNCERTIFIED"
+ if degraded and uncertified
+ else (
+ "CAPABILITY_VERIFIED_DEGRADED"
+ if degraded
+ else (
+ "CAPABILITY_VERIFIED_UNCERTIFIED"
+ if uncertified
+ else "CAPABILITY_VERIFIED"
+ )
+ )
),
detected_versions=detected_versions,
active_variant=variant.name,
diff --git a/doris_mcp_server/tools/cluster_handlers.py
b/doris_mcp_server/tools/cluster_handlers.py
new file mode 100644
index 0000000..c1bbd39
--- /dev/null
+++ b/doris_mcp_server/tools/cluster_handlers.py
@@ -0,0 +1,198 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Formal Cluster-domain handlers backed by one strict runtime."""
+
+from __future__ import annotations
+
+from typing import Any, Protocol, cast
+
+from ..utils.cluster_runtime import DorisClusterRuntime
+from ..utils.db import DorisConnectionManager
+from ..utils.monitoring_tools import DorisMonitoringTools
+from ..utils.security import get_current_auth_context
+from .capability_registry import CapabilityRegistry
+
+
+class _ClusterHandlerOwner(Protocol):
+ """State supplied by ``DorisToolsManager`` to the mixin."""
+
+ cluster_runtime: DorisClusterRuntime
+ _capability_registry: CapabilityRegistry | None
+
+
+class ClusterToolHandlersMixin:
+ """Route every Cluster child through the strict evidence runtime."""
+
+ def _initialize_cluster_handlers(
+ self: _ClusterHandlerOwner,
+ connection_manager: DorisConnectionManager,
+ monitoring_tools: DorisMonitoringTools,
+ ) -> None:
+ self.cluster_runtime = DorisClusterRuntime(
+ connection_manager,
+ monitoring_tools,
+ )
+
+ async def _formal_doris_cluster_get_cluster_overview_tool(
+ self: _ClusterHandlerOwner,
+ arguments: dict[str, Any],
+ ) -> dict[str, Any]:
+ return await self.cluster_runtime.get_cluster_overview(
+ include=cast(list[str] | None, arguments.get("include")),
+ )
+
+ async def _formal_doris_cluster_list_cluster_nodes_tool(
+ self: _ClusterHandlerOwner,
+ arguments: dict[str, Any],
+ ) -> dict[str, Any]:
+ return await self.cluster_runtime.list_cluster_nodes(
+ node_types=cast(list[str] | None, arguments.get("node_types")),
+ include_metrics=bool(arguments.get("include_metrics", False)),
+ )
+
+ async def _formal_doris_cluster_list_active_tasks_tool(
+ self: _ClusterHandlerOwner,
+ arguments: dict[str, Any],
+ ) -> dict[str, Any]:
+ return await self.cluster_runtime.list_active_tasks(
+ task_types=cast(list[str] | None, arguments.get("task_types")),
+ states=cast(list[str] | None, arguments.get("states")),
+ limit=cast(int | None, arguments.get("limit")),
+ )
+
+ async def _formal_doris_cluster_get_cache_status_tool(
+ self: _ClusterHandlerOwner,
+ arguments: dict[str, Any],
+ ) -> dict[str, Any]:
+ return await self.cluster_runtime.get_cache_status(
+ scope=cast(str | None, arguments.get("scope")),
+ node_ids=cast(list[str] | None, arguments.get("node_ids")),
+ include_queues=bool(arguments.get("include_queues", False)),
+ )
+
+ async def _formal_doris_cluster_get_compaction_status_tool(
+ self: _ClusterHandlerOwner,
+ arguments: dict[str, Any],
+ ) -> dict[str, Any]:
+ return await self.cluster_runtime.get_compaction_status(
+ database=cast(str | None, arguments.get("database")),
+ table=cast(str | None, arguments.get("table")),
+ state=cast(str | None, arguments.get("state")),
+ limit=cast(int | None, arguments.get("limit")),
+ )
+
+ async def _formal_doris_cluster_get_workload_group_status_tool(
+ self: _ClusterHandlerOwner,
+ arguments: dict[str, Any],
+ ) -> dict[str, Any]:
+ return await self.cluster_runtime.get_workload_group_status(
+ name=cast(str | None, arguments.get("name")),
+ include_usage=bool(arguments.get("include_usage", False)),
+ )
+
+ async def _formal_doris_cluster_get_compute_group_status_tool(
+ self: _ClusterHandlerOwner,
+ arguments: dict[str, Any],
+ ) -> dict[str, Any]:
+ return await self.cluster_runtime.get_compute_group_status(
+ name=cast(str | None, arguments.get("name")),
+ include_nodes=bool(arguments.get("include_nodes", False)),
+ include_usage=bool(arguments.get("include_usage", False)),
+ )
+
+ async def _formal_doris_cluster_get_runtime_capabilities_tool(
+ self: _ClusterHandlerOwner,
+ arguments: dict[str, Any],
+ ) -> dict[str, Any]:
+ registry = self._capability_registry
+ if registry is None:
+ return {
+ "status": "partial",
+ "data": {
+ "detail": arguments.get("detail", "summary"),
+ "capability_registry": "external",
+ },
+ "warnings": [
+ "Runtime capability evidence is supplied by an external "
+ "availability provider."
+ ],
+ "metadata": {"source": "availability_provider"},
+ "evidence": [],
+ }
+ snapshot = await registry.ensure_fresh(
+ "doris_cluster",
+ get_current_auth_context(),
+ )
+ detail = str(arguments.get("detail", "summary"))
+ versions = {
+ "master_fe": snapshot.version_vector.master_fe.normalized,
+ "follower_fes": [
+ version.normalized for version in
snapshot.version_vector.follower_fes
+ ],
+ "backends": [
+ version.normalized for version in
snapshot.version_vector.backends
+ ],
+ }
+ statuses: dict[str, int] = {}
+ for probe in snapshot.probes.values():
+ statuses[probe.status.value] = statuses.get(probe.status.value, 0)
+ 1
+ data: dict[str, Any] = {
+ "detail": detail,
+ "versions": versions,
+ "deployment_mode": snapshot.deployment_mode,
+ "mixed_versions": snapshot.mixed_versions,
+ "stale": snapshot.stale,
+ "probe_status_counts": statuses,
+ "probed_domains": sorted(snapshot.probed_domains),
+ }
+ if detail == "full":
+ data["probes"] = {
+ probe_id: {
+ "status": probe.status.value,
+ "reason_code": probe.reason_code,
+ "evidence_sources": list(probe.evidence_sources),
+ }
+ for probe_id, probe in sorted(snapshot.probes.items())
+ }
+ warnings = (
+ ["Capability evidence is stale and may be retried."]
+ if snapshot.stale
+ else []
+ )
+ return {
+ "status": "partial" if warnings else "success",
+ "data": data,
+ "warnings": warnings,
+ "metadata": {
+ "source": "route_private_capability_snapshot",
+ "created_at": snapshot.created_at.isoformat(),
+ "expires_at": snapshot.expires_at.isoformat(),
+ },
+ "evidence": [
+ {
+ "source": "SELECT @@version_comment",
+ "probe": "version_probe_completed",
+ },
+ {
+ "source": "SHOW FRONTENDS, SHOW BACKENDS",
+ "probe": "cluster_components_discoverable",
+ },
+ ],
+ }
+
+
+__all__ = ["ClusterToolHandlersMixin"]
diff --git a/doris_mcp_server/tools/domain_dispatcher.py
b/doris_mcp_server/tools/domain_dispatcher.py
index e9eb6bd..0d8b60b 100644
--- a/doris_mcp_server/tools/domain_dispatcher.py
+++ b/doris_mcp_server/tools/domain_dispatcher.py
@@ -43,6 +43,7 @@ from ..schema_validation import (
)
from ..state_handles import StateHandleCodec, StateHandleError
from ..utils.catalog_metadata import CatalogMetadataFailure
+from ..utils.cluster_runtime import ClusterRuntimeFailure
from ..utils.logger import get_audit_logger, get_logger
from ..utils.query_runtime import QueryRuntimeFailure
from . import domain_catalog as domain_catalog_module
@@ -640,6 +641,25 @@ class DomainDispatcher:
"status_code": exc.status_code,
},
)
+ except ClusterRuntimeFailure as exc:
+ self._audit(feature_id, arguments, "error", started)
+ return self._error(
+ domain.name,
+ (
+ DomainErrorCode.CHILD_ARGUMENTS_INVALID
+ if exc.reason_code == "CLUSTER_ARGUMENT_INVALID"
+ else DomainErrorCode.CHILD_EXECUTION_FAILED
+ ),
+ str(exc),
+ child_tool=child.name,
+ manifest_version=manifest.manifest_version,
+ retryable=exc.retryable,
+ details={
+ "rediscover": False,
+ "reason_code": exc.reason_code,
+ "status_code": exc.status_code,
+ },
+ )
except ToolOutputValidationError:
logger.exception("Formal child output validation failed for %s",
feature_id)
self._audit(
@@ -1131,24 +1151,9 @@ def _adapt_arguments(
prepared["single_replica"] = False
prepared["_formal_catalog"] = True
elif adapter_name == "adapt:monitoring_arguments":
- if prepared:
- warnings.append(
- "Metric and node filters are not supported by the active
handler."
- )
- prepared = {}
+ pass
elif adapter_name == "adapt:memory_arguments":
- detail = prepared.pop("detail", None)
- if prepared.pop("node_ids", None) is not None:
- warnings.append(
- "Memory node filtering is not supported by the active handler."
- )
- prepared["data_type"] = "realtime"
- prepared["tracker_type"] = {
- None: "overview",
- "summary": "overview",
- "trackers": "all",
- "top_consumers": "all",
- }.get(detail, "overview")
+ pass
elif adapter_name == "adapt:audit_filters":
window_minutes = prepared.pop("window_minutes", None)
if window_minutes is not None:
@@ -1214,13 +1219,7 @@ def _adapt_arguments(
1000,
)
elif adapter_name == "adapt:resource_growth_arguments":
- resource = prepared.pop("resource", None)
- prepared["days"] = prepared.pop("window_days", 30)
- prepared["resource_types"] = [resource] if resource else []
- if prepared.pop("granularity", None) is not None:
- warnings.append(
- "Growth granularity is not supported by the active handler."
- )
+ pass
elif adapter_name == "adapt:adbc_query_arguments":
timeout_ms = prepared.pop("timeout_ms", None)
if timeout_ms is not None:
diff --git a/doris_mcp_server/tools/doris_feature_matrix.py
b/doris_mcp_server/tools/doris_feature_matrix.py
index 3c9d9ae..1067e80 100644
--- a/doris_mcp_server/tools/doris_feature_matrix.py
+++ b/doris_mcp_server/tools/doris_feature_matrix.py
@@ -1059,28 +1059,22 @@ FEATURE_DEFINITIONS = (
"doris_cluster",
"list_active_tasks",
A,
- _variant(
- "legacy_task_views",
- probes=("legacy_task_views_readable",),
- callable_when_degraded=True,
- ),
_variant(
"unified_task_progress",
ranges=(">=4.1.1",),
- system_objects=("information_schema.active_tasks",),
probes=("unified_task_progress_readable",),
sources=("DORIS_RELEASE_4_1_1",),
),
+ _variant(
+ "legacy_task_views",
+ probes=("legacy_task_views_readable",),
+ callable_when_degraded=True,
+ ),
),
_feature(
"doris_cluster",
"get_monitoring_metrics",
A,
- _variant(
- "base_metrics",
- endpoints=("fe_metrics", "be_metrics"),
- probes=("metrics_endpoints_readable",),
- ),
_variant(
"observability_4_0_7",
ranges=(">=4.0.7,<4.1.0",),
@@ -1092,6 +1086,11 @@ FEATURE_DEFINITIONS = (
probes=("enhanced_metrics_present",),
sources=("DORIS_RELEASE_4_0_7",),
),
+ _variant(
+ "base_metrics",
+ endpoints=("fe_metrics", "be_metrics"),
+ probes=("metrics_endpoints_readable",),
+ ),
),
_feature(
"doris_cluster",
@@ -1109,9 +1108,11 @@ FEATURE_DEFINITIONS = (
"get_cache_status",
B,
_variant(
- "file_cache",
- endpoints=("be_metrics",),
- probes=("file_cache_metrics_readable",),
+ "advanced_cache_types",
+ ranges=(">=4.1.0",),
+ features=("condition_cache", "parquet_page_cache"),
+ probes=("advanced_cache_metrics_present",),
+ sources=("DORIS_RELEASE_4_1_0",),
),
_variant(
"file_cache_queue_metrics",
@@ -1120,31 +1121,29 @@ FEATURE_DEFINITIONS = (
sources=("DORIS_RELEASE_4_0_7",),
),
_variant(
- "advanced_cache_types",
- ranges=(">=4.1.0",),
- features=("condition_cache", "parquet_page_cache"),
- probes=("advanced_cache_metrics_present",),
- sources=("DORIS_RELEASE_4_1_0",),
+ "file_cache",
+ system_objects=("information_schema.file_cache_statistics",),
+ probes=("file_cache_metrics_readable",),
),
),
_feature(
"doris_cluster",
"get_compaction_status",
B,
- _variant(
- "legacy_compaction_summary",
- probes=("legacy_compaction_status_readable",),
- evidence_quality="summary",
- callable_when_degraded=True,
- ),
_variant(
"compaction_task_tracker",
ranges=(">=4.0.6,<4.1.0", ">=4.1.1"),
excluded_ranges=(">=4.1.0-alpha0,<4.1.1",),
- endpoints=("be_compaction_api",),
+ system_objects=("information_schema.doris_be_compaction_tasks",),
probes=("compaction_system_table_or_http_api",),
sources=("DORIS_RELEASE_4_0_6", "DORIS_RELEASE_4_1_1"),
),
+ _variant(
+ "legacy_compaction_summary",
+ probes=("legacy_compaction_status_readable",),
+ evidence_quality="summary",
+ callable_when_degraded=True,
+ ),
),
_feature(
"doris_cluster",
diff --git a/doris_mcp_server/tools/tools_manager.py
b/doris_mcp_server/tools/tools_manager.py
index e654f2a..246903c 100644
--- a/doris_mcp_server/tools/tools_manager.py
+++ b/doris_mcp_server/tools/tools_manager.py
@@ -29,7 +29,7 @@ from mcp.types import Tool
from ..auth.operation_policy import authorize_operation
from ..state_handles import StateHandleCodec
from ..utils.adbc_query_tools import DorisADBCQueryTools
-from ..utils.analysis_tools import MemoryTracker, SQLAnalyzer, TableAnalyzer
+from ..utils.analysis_tools import SQLAnalyzer, TableAnalyzer
from ..utils.data_exploration_tools import DataExplorationTools
from ..utils.data_governance_tools import DataGovernanceTools
from ..utils.data_quality_tools import DataQualityTools
@@ -37,7 +37,6 @@ from ..utils.db import DorisConnectionManager
from ..utils.dependency_analysis_tools import DependencyAnalysisTools
from ..utils.logger import get_audit_logger, get_logger
from ..utils.monitoring_tools import DorisMonitoringTools
-from ..utils.performance_analytics_tools import PerformanceAnalyticsTools
from ..utils.query_executor import DorisQueryExecutor
from ..utils.schema_extractor import MetadataExtractor
from ..utils.security import get_current_auth_context
@@ -49,6 +48,7 @@ from .capability_registry import (
CapabilityRegistry,
)
from .catalog_handlers import CatalogToolHandlersMixin
+from .cluster_handlers import ClusterToolHandlersMixin
from .domain_catalog import CURRENT_FLAT_TOOL_NAMES
from .domain_dispatcher import (
BoundHandlerAvailabilityProvider,
@@ -73,6 +73,7 @@ logger = get_logger(__name__)
class DorisToolsManager(
QueryToolHandlersMixin,
CatalogToolHandlersMixin,
+ ClusterToolHandlersMixin,
DomainManifestManagerMixin,
):
"""Apache Doris Tools Manager"""
@@ -96,7 +97,10 @@ class DorisToolsManager(
)
self._initialize_catalog_handlers(connection_manager)
self.monitoring_tools = DorisMonitoringTools(connection_manager)
- self.memory_tracker = MemoryTracker(connection_manager)
+ self._initialize_cluster_handlers(
+ connection_manager,
+ self.monitoring_tools,
+ )
# Initialize v0.5.0 advanced analytics tools
self.data_governance_tools = DataGovernanceTools(connection_manager)
@@ -106,7 +110,6 @@ class DorisToolsManager(
)
self.security_analytics_tools =
SecurityAnalyticsTools(connection_manager)
self.dependency_analysis_tools =
DependencyAnalysisTools(connection_manager)
- self.performance_analytics_tools =
PerformanceAnalyticsTools(connection_manager)
# Initialize ADBC query tools
self.adbc_query_tools = DorisADBCQueryTools(connection_manager)
@@ -311,98 +314,26 @@ class DorisToolsManager(
async def _get_monitoring_metrics_tool(
self, arguments: dict[str, Any]
) -> dict[str, Any]:
- """Unified monitoring metrics tool routing"""
- content_type = arguments.get("content_type", "data")
- role = arguments.get("role", "all")
- monitor_type = arguments.get("monitor_type", "all")
- priority = arguments.get("priority", "core")
- include_raw_metrics = arguments.get("include_raw_metrics", False)
-
- if content_type == "definitions":
- # Only get definitions
- return await self.monitoring_tools.get_monitoring_metrics(
- role, monitor_type, priority, info_only=True,
format_type="prometheus"
- )
- elif content_type == "data":
- # Only get data
- return await self.monitoring_tools.get_monitoring_metrics(
- role,
- monitor_type,
- priority,
- info_only=False,
- format_type="prometheus",
- include_raw_metrics=include_raw_metrics,
- )
- elif content_type == "both":
- # Get both definitions and data
- definitions = await self.monitoring_tools.get_monitoring_metrics(
- role, monitor_type, priority, info_only=True,
format_type="prometheus"
- )
- data = await self.monitoring_tools.get_monitoring_metrics(
- role,
- monitor_type,
- priority,
- info_only=False,
- format_type="prometheus",
- include_raw_metrics=include_raw_metrics,
- )
- return {
- "content_type": "both",
- "definitions": definitions,
- "data": data,
- "timestamp": data.get("timestamp"),
- "_execution_info": {
- "combined_response": True,
- "definitions_available": definitions.get("success", False),
- "data_available": data.get("success", False),
- },
- }
- else:
- return {
- "error": f"Invalid content_type: {content_type}. Must be
'definitions', 'data', or 'both'"
- }
+ """Route the migrated metric child through the strict Cluster
runtime."""
+ return await self.cluster_runtime.get_monitoring_metrics(
+ metric_names=arguments.get("metric_names"),
+ node_ids=arguments.get("node_ids"),
+ window=arguments.get("window"),
+ )
async def _get_memory_stats_tool(self, arguments: dict[str, Any]) ->
dict[str, Any]:
- """Unified memory statistics tool routing"""
- data_type = arguments.get("data_type", "realtime")
- tracker_type = arguments.get("tracker_type", "overview")
- tracker_names = arguments.get("tracker_names")
- time_range = arguments.get("time_range", "1h")
- include_details = arguments.get("include_details", True)
-
- if data_type == "realtime":
- # Only get real-time data
- return await self.memory_tracker.get_realtime_memory_stats(
- tracker_type, include_details
- )
- elif data_type == "historical":
- # Only get historical data
- return await self.memory_tracker.get_historical_memory_stats(
- tracker_names, time_range
- )
- elif data_type == "both":
- # Get both real-time and historical data
- realtime = await self.memory_tracker.get_realtime_memory_stats(
- tracker_type, include_details
- )
- historical = await self.memory_tracker.get_historical_memory_stats(
- tracker_names, time_range
- )
- return {
- "data_type": "both",
- "realtime": realtime,
- "historical": historical,
- "timestamp": realtime.get("timestamp"),
- "_execution_info": {
- "combined_response": True,
- "realtime_available": realtime.get("success", False),
- "historical_available": historical.get("success", False),
- },
- }
- else:
- return {
- "error": f"Invalid data_type: {data_type}. Must be 'realtime',
'historical', or 'both'"
- }
+ """Route the migrated memory child without placeholder values."""
+ detail = {
+ "overview": "summary",
+ "all": "trackers",
+ }.get(
+ str(arguments.get("tracker_type", "")),
+ arguments.get("detail", "summary"),
+ )
+ return await self.cluster_runtime.get_memory_stats(
+ node_ids=arguments.get("node_ids"),
+ detail=detail,
+ )
# Legacy tool methods (for backward compatibility)
async def _get_monitoring_metrics_info_tool(
@@ -618,15 +549,15 @@ class DorisToolsManager(
async def _analyze_resource_growth_curves_tool(
self, arguments: dict[str, Any]
) -> dict[str, Any]:
- """Resource growth curves analysis tool routing"""
- days = arguments.get("days", 30)
- resource_types = arguments.get(
- "resource_types", ["storage", "query_volume", "user_activity"]
+ """Route the migrated growth child through recorded Doris evidence."""
+ resource_types = arguments.get("resource_types")
+ resource = (
+ resource_types[0]
+ if isinstance(resource_types, list) and len(resource_types) == 1
+ else arguments.get("resource")
)
- include_predictions = arguments.get("include_predictions", False)
- detailed_response = arguments.get("detailed_response", False)
-
- # Delegate to performance analytics tools for processing
- return await
self.performance_analytics_tools.analyze_resource_growth_curves(
- days, resource_types, include_predictions, detailed_response
+ return await self.cluster_runtime.analyze_resource_growth(
+ resource=resource,
+ window_days=arguments.get("days", arguments.get("window_days")),
+ granularity=arguments.get("granularity"),
)
diff --git a/doris_mcp_server/utils/cluster_runtime.py
b/doris_mcp_server/utils/cluster_runtime.py
new file mode 100644
index 0000000..ce88fa2
--- /dev/null
+++ b/doris_mcp_server/utils/cluster_runtime.py
@@ -0,0 +1,1185 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Strict, evidence-bearing runtime for the read-only Cluster domain."""
+
+from __future__ import annotations
+
+import re
+import uuid
+from collections.abc import Mapping, Sequence
+from datetime import datetime
+from typing import Any
+
+from .db import DorisConnectionManager
+from .monitoring_tools import DorisMonitoringTools
+from .redaction import redact_sensitive_data
+from .security import get_current_auth_context
+
+_MAX_ROWS = 1_000
+_MAX_METRIC_NAMES = 100
+_MAX_METRICS_PER_NODE = 256
+_MEMORY_MARKERS = ("memory", "mem_tracker", "memtable", "jemalloc")
+_CACHE_MARKERS = ("cache", "file_cache")
+_COMPACTION_MARKERS = ("compaction",)
+_SNAKE_BOUNDARY = re.compile(r"(?<!^)(?=[A-Z])")
+_SAFE_KEY = re.compile(r"[^a-z0-9]+")
+_GROWTH_SQL = {
+ ("query_volume", "hour"): (
+ "SELECT DATE_FORMAT(`time`, '%%Y-%%m-%%dT%%H:00:00') AS bucket, "
+ "COUNT(*) AS value "
+ "FROM internal.__internal_schema.audit_log "
+ "WHERE `time` >= DATE_SUB(NOW(), INTERVAL %s DAY) "
+ "GROUP BY bucket ORDER BY bucket"
+ ),
+ ("query_volume", "day"): (
+ "SELECT DATE(`time`) AS bucket, COUNT(*) AS value "
+ "FROM internal.__internal_schema.audit_log "
+ "WHERE `time` >= DATE_SUB(NOW(), INTERVAL %s DAY) "
+ "GROUP BY bucket ORDER BY bucket"
+ ),
+ ("query_volume", "week"): (
+ "SELECT DATE_FORMAT("
+ "DATE_SUB(DATE(`time`), INTERVAL WEEKDAY(`time`) DAY), "
+ "'%%Y-%%m-%%d') AS bucket, COUNT(*) AS value "
+ "FROM internal.__internal_schema.audit_log "
+ "WHERE `time` >= DATE_SUB(NOW(), INTERVAL %s DAY) "
+ "GROUP BY bucket ORDER BY bucket"
+ ),
+ ("user_activity", "hour"): (
+ "SELECT DATE_FORMAT(`time`, '%%Y-%%m-%%dT%%H:00:00') AS bucket, "
+ "COUNT(DISTINCT `user`) AS value "
+ "FROM internal.__internal_schema.audit_log "
+ "WHERE `time` >= DATE_SUB(NOW(), INTERVAL %s DAY) "
+ "GROUP BY bucket ORDER BY bucket"
+ ),
+ ("user_activity", "day"): (
+ "SELECT DATE(`time`) AS bucket, COUNT(DISTINCT `user`) AS value "
+ "FROM internal.__internal_schema.audit_log "
+ "WHERE `time` >= DATE_SUB(NOW(), INTERVAL %s DAY) "
+ "GROUP BY bucket ORDER BY bucket"
+ ),
+ ("user_activity", "week"): (
+ "SELECT DATE_FORMAT("
+ "DATE_SUB(DATE(`time`), INTERVAL WEEKDAY(`time`) DAY), "
+ "'%%Y-%%m-%%d') AS bucket, COUNT(DISTINCT `user`) AS value "
+ "FROM internal.__internal_schema.audit_log "
+ "WHERE `time` >= DATE_SUB(NOW(), INTERVAL %s DAY) "
+ "GROUP BY bucket ORDER BY bucket"
+ ),
+ ("storage", "hour"): (
+ "SELECT DATE_FORMAT(CREATE_TIME, '%%Y-%%m-%%dT%%H:00:00') AS bucket, "
+ "SUM(COALESCE(DATA_LENGTH, 0) + COALESCE(INDEX_LENGTH, 0)) AS value "
+ "FROM information_schema.partitions "
+ "WHERE CREATE_TIME >= DATE_SUB(NOW(), INTERVAL %s DAY) "
+ "GROUP BY bucket ORDER BY bucket"
+ ),
+ ("storage", "day"): (
+ "SELECT DATE(CREATE_TIME) AS bucket, "
+ "SUM(COALESCE(DATA_LENGTH, 0) + COALESCE(INDEX_LENGTH, 0)) AS value "
+ "FROM information_schema.partitions "
+ "WHERE CREATE_TIME >= DATE_SUB(NOW(), INTERVAL %s DAY) "
+ "GROUP BY bucket ORDER BY bucket"
+ ),
+ ("storage", "week"): (
+ "SELECT DATE_FORMAT("
+ "DATE_SUB(DATE(CREATE_TIME), INTERVAL WEEKDAY(CREATE_TIME) DAY), "
+ "'%%Y-%%m-%%d') AS bucket, "
+ "SUM(COALESCE(DATA_LENGTH, 0) + COALESCE(INDEX_LENGTH, 0)) AS value "
+ "FROM information_schema.partitions "
+ "WHERE CREATE_TIME >= DATE_SUB(NOW(), INTERVAL %s DAY) "
+ "GROUP BY bucket ORDER BY bucket"
+ ),
+}
+
+
+class ClusterRuntimeFailure(RuntimeError):
+ """Sanitized Cluster-domain failure with a stable reason code."""
+
+ def __init__(
+ self,
+ message: str,
+ *,
+ reason_code: str,
+ status_code: int,
+ retryable: bool = False,
+ ) -> None:
+ super().__init__(message)
+ self.reason_code = reason_code
+ self.status_code = status_code
+ self.retryable = retryable
+
+
+class DorisClusterRuntime:
+ """Read bounded cluster metadata and monitoring evidence without
invention."""
+
+ def __init__(
+ self,
+ connection_manager: DorisConnectionManager,
+ monitoring_tools: DorisMonitoringTools,
+ ) -> None:
+ self._connection_manager = connection_manager
+ self._monitoring_tools = monitoring_tools
+ self._session_prefix = f"formal_cluster_{uuid.uuid4().hex[:8]}"
+
+ async def get_cluster_overview(
+ self,
+ *,
+ include: Sequence[str] | None,
+ ) -> dict[str, Any]:
+ sections = tuple(
+ dict.fromkeys(
+ include
+ or (
+ "nodes",
+ "tasks",
+ "metrics",
+ "memory",
+ "cache",
+ "compaction",
+ )
+ )
+ )
+ data: dict[str, Any] = {}
+ evidence: list[dict[str, Any]] = []
+ warnings: list[str] = []
+ operations = {
+ "nodes": lambda: self.list_cluster_nodes(
+ node_types=None,
+ include_metrics=False,
+ ),
+ "tasks": lambda: self.list_active_tasks(
+ task_types=None,
+ states=None,
+ limit=100,
+ ),
+ "metrics": lambda: self.get_monitoring_metrics(
+ metric_names=None,
+ node_ids=None,
+ window=None,
+ ),
+ "memory": lambda: self.get_memory_stats(
+ node_ids=None,
+ detail="summary",
+ ),
+ "cache": lambda: self.get_cache_status(
+ scope=None,
+ node_ids=None,
+ include_queues=False,
+ ),
+ "compaction": lambda: self.get_compaction_status(
+ database=None,
+ table=None,
+ state=None,
+ limit=100,
+ ),
+ }
+ for section in sections:
+ operation = operations[section]
+ try:
+ result = await operation()
+ except ClusterRuntimeFailure as exc:
+ data[section] = {
+ "status": "unavailable",
+ "reason_code": exc.reason_code,
+ }
+ warnings.append(
+ f"{section} evidence is unavailable ({exc.reason_code})."
+ )
+ evidence.append(
+ {
+ "section": section,
+ "success": False,
+ "reason_code": exc.reason_code,
+ }
+ )
+ continue
+ data[section] = {
+ "status": result["status"],
+ "data": result["data"],
+ }
+ warnings.extend(result["warnings"])
+ evidence.append(
+ {
+ "section": section,
+ "success": True,
+ "source": result["metadata"].get("source"),
+ }
+ )
+ if not any(item["success"] for item in evidence):
+ raise ClusterRuntimeFailure(
+ "Doris cluster evidence is unavailable.",
+ reason_code="CLUSTER_EVIDENCE_UNAVAILABLE",
+ status_code=503,
+ retryable=True,
+ )
+ return _result(
+ data,
+ source="cluster_composite",
+ warnings=warnings,
+ evidence=evidence,
+ metadata={"sections": list(sections)},
+ )
+
+ async def list_cluster_nodes(
+ self,
+ *,
+ node_types: Sequence[str] | None,
+ include_metrics: bool,
+ ) -> dict[str, Any]:
+ requested = set(node_types or ("fe", "be", "broker"))
+ statements = {
+ "fe": "SHOW FRONTENDS",
+ "be": "SHOW BACKENDS",
+ "broker": "SHOW BROKER",
+ }
+ items: list[dict[str, Any]] = []
+ warnings: list[str] = []
+ successful_sources: list[str] = []
+ for node_type in ("fe", "be", "broker"):
+ if node_type not in requested:
+ continue
+ try:
+ rows = await self._execute(statements[node_type], max_rows=512)
+ except ClusterRuntimeFailure as exc:
+ if node_type in {"fe", "be"}:
+ warnings.append(
+ f"{node_type.upper()} metadata is unavailable "
+ f"({exc.reason_code})."
+ )
+ continue
+ successful_sources.append(statements[node_type])
+ items.extend(_node_item(node_type, row) for row in rows)
+ if not successful_sources:
+ raise ClusterRuntimeFailure(
+ "Doris node metadata is unavailable.",
+ reason_code="CLUSTER_NODE_METADATA_UNAVAILABLE",
+ status_code=503,
+ retryable=True,
+ )
+
+ metric_nodes: dict[str, dict[str, Any]] = {}
+ if include_metrics:
+ try:
+ monitoring_result = await self.get_monitoring_metrics(
+ metric_names=None,
+ node_ids=None,
+ window=None,
+ )
+ except ClusterRuntimeFailure as exc:
+ warnings.append(f"Node metrics are unavailable
({exc.reason_code}).")
+ else:
+ for node in monitoring_result["data"].get("nodes", []):
+ for identity in (
+ node.get("node_id"),
+ node.get("host"),
+ ):
+ if identity:
+ metric_nodes[str(identity)] = node.get("metrics",
{})
+ for item in items:
+ node_metrics: dict[str, Any] | None = None
+ for identity in (item.get("node_id"), item.get("host")):
+ if identity is not None and str(identity) in metric_nodes:
+ node_metrics = metric_nodes[str(identity)]
+ break
+ if node_metrics is not None:
+ item["metrics"] = node_metrics
+ items.sort(
+ key=lambda item: (
+ str(item.get("node_type", "")),
+ str(item.get("node_id", "")),
+ str(item.get("host", "")),
+ )
+ )
+ return _collection(
+ items,
+ source=",".join(successful_sources),
+ warnings=warnings,
+ )
+
+ async def list_active_tasks(
+ self,
+ *,
+ task_types: Sequence[str] | None,
+ states: Sequence[str] | None,
+ limit: int | None,
+ ) -> dict[str, Any]:
+ requested_types = {value.casefold() for value in (task_types or
("query",))}
+ max_rows = _bounded_limit(limit, default=100)
+ items: list[dict[str, Any]] = []
+ warnings: list[str] = []
+ sources: list[str] = []
+
+ if "query" in requested_types:
+ query_rows: list[Mapping[str, Any]] | None = None
+ for statement in (
+ 'SHOW PROC "/current_queries"',
+ "SELECT * FROM information_schema.active_queries",
+ "SHOW FULL PROCESSLIST",
+ ):
+ try:
+ query_rows = await self._execute(
+ statement,
+ max_rows=max_rows + 1,
+ )
+ except ClusterRuntimeFailure:
+ continue
+ sources.append(statement)
+ break
+ if query_rows is None:
+ warnings.append("Active query tasks are unavailable.")
+ else:
+ for row in query_rows:
+ item = _normalized_row(row)
+ command = str(item.get("command", "")).casefold()
+ if command == "sleep":
+ continue
+ item["task_type"] = "query"
+ items.append(item)
+
+ if "compaction" in requested_types:
+ try:
+ rows = await self._execute(
+ "SELECT * FROM
information_schema.doris_be_compaction_tasks",
+ max_rows=max_rows + 1,
+ )
+ except ClusterRuntimeFailure:
+ warnings.append(
+ "Compaction task detail is unavailable; use "
+ "get_compaction_status for a summary fallback."
+ )
+ else:
+ sources.append("information_schema.doris_be_compaction_tasks")
+ for row in rows:
+ item = _normalized_row(row)
+ item["task_type"] = "compaction"
+ items.append(item)
+
+ unsupported = requested_types - {"query", "compaction"}
+ if unsupported:
+ warnings.append(
+ "No cluster-wide read-only source is available for requested "
+ f"task types: {', '.join(sorted(unsupported))}."
+ )
+
+ state_filter = {value.casefold() for value in (states or ())}
+ if state_filter:
+ items = [
+ item for item in items if _task_state(item).casefold() in
state_filter
+ ]
+ items.sort(
+ key=lambda item: (
+ str(item.get("task_type", "")),
+ str(
+ item.get("query_id") or item.get("task_id") or
item.get("id") or ""
+ ),
+ )
+ )
+ truncated = len(items) > max_rows
+ items = items[:max_rows]
+ if not sources and not items:
+ raise ClusterRuntimeFailure(
+ "Doris active task metadata is unavailable.",
+ reason_code="CLUSTER_TASK_METADATA_UNAVAILABLE",
+ status_code=503,
+ retryable=True,
+ )
+ return _collection(
+ items,
+ source=",".join(sources) or "cluster_task_views",
+ warnings=warnings,
+ truncated=truncated,
+ )
+
+ async def get_monitoring_metrics(
+ self,
+ *,
+ metric_names: Sequence[str] | None,
+ node_ids: Sequence[str] | None,
+ window: str | None,
+ ) -> dict[str, Any]:
+ requested_names = tuple(dict.fromkeys(metric_names or ()))
+ if len(requested_names) > _MAX_METRIC_NAMES:
+ raise _argument_failure("At most 100 metric names may be
requested.")
+ raw = await self._monitoring_tools.get_monitoring_metrics(
+ role="all",
+ monitor_type="all",
+ priority="all" if requested_names else "p0",
+ info_only=False,
+ include_raw_metrics=bool(requested_names),
+ format_type="prometheus",
+ )
+ if raw.get("success") is False:
+ raise ClusterRuntimeFailure(
+ "Doris monitoring metrics are unavailable.",
+ reason_code="CLUSTER_METRICS_UNAVAILABLE",
+ status_code=503,
+ retryable=True,
+ )
+ nodes, warnings = _monitoring_nodes(
+ raw,
+ metric_names=requested_names,
+ node_ids=node_ids,
+ )
+ if not nodes:
+ raise ClusterRuntimeFailure(
+ "No readable Doris monitoring endpoint returned metrics.",
+ reason_code="CLUSTER_METRICS_UNAVAILABLE",
+ status_code=503,
+ retryable=True,
+ )
+ if window:
+ warnings.append(
+ "The Doris metrics endpoint returns a point-in-time snapshot; "
+ "the requested window was not synthesized."
+ )
+ return _result(
+ {"nodes": nodes},
+ source="doris_http_metrics",
+ warnings=warnings,
+ metadata={
+ "snapshot_at": raw.get("timestamp"),
+ "requested_metric_names": list(requested_names),
+ },
+ )
+
+ async def get_memory_stats(
+ self,
+ *,
+ node_ids: Sequence[str] | None,
+ detail: str | None,
+ ) -> dict[str, Any]:
+ nodes, warnings = await self._metric_family(
+ markers=_MEMORY_MARKERS,
+ node_ids=node_ids,
+ )
+ if not nodes:
+ raise ClusterRuntimeFailure(
+ "No real Doris memory metrics were observed.",
+ reason_code="MEMORY_METRICS_UNAVAILABLE",
+ status_code=503,
+ retryable=True,
+ )
+ effective_detail = detail or "summary"
+ if effective_detail == "top_consumers":
+ for node in nodes:
+ metrics = node["metrics"]
+ node["metrics"] = dict(
+ sorted(
+ metrics.items(),
+ key=lambda item: _metric_total(item[1]),
+ reverse=True,
+ )[:20]
+ )
+ return _result(
+ {"detail": effective_detail, "nodes": nodes},
+ source="doris_be_metrics",
+ warnings=warnings,
+ metadata={"invented_values": False},
+ )
+
+ async def get_cache_status(
+ self,
+ *,
+ scope: str | None,
+ node_ids: Sequence[str] | None,
+ include_queues: bool,
+ ) -> dict[str, Any]:
+ warnings: list[str] = []
+ try:
+ rows = await self._execute(
+ "SELECT * FROM information_schema.file_cache_statistics",
+ max_rows=_MAX_ROWS,
+ )
+ except ClusterRuntimeFailure as exc:
+ warnings.append(
+ "File-cache system table is unavailable; using BE metrics "
+ f"({exc.reason_code})."
+ )
+ else:
+ items = [_normalized_row(row) for row in rows]
+ if node_ids:
+ allowed = {str(value) for value in node_ids}
+ items = [
+ item
+ for item in items
+ if str(item.get("be_id", "")) in allowed
+ or str(item.get("be_ip", "")) in allowed
+ ]
+ if scope:
+ scope_folded = scope.casefold()
+ items = [
+ item
+ for item in items
+ if scope_folded
+ in " ".join(str(value) for value in
item.values()).casefold()
+ ]
+ if not include_queues:
+ items = [
+ item
+ for item in items
+ if "queue" not in str(item.get("metric_name",
"")).casefold()
+ ]
+ return _result(
+ {"mode": "system_table", "items": items},
+ source="information_schema.file_cache_statistics",
+ warnings=warnings,
+ metadata={"row_count": len(items)},
+ )
+
+ nodes, metric_warnings = await self._metric_family(
+ markers=_CACHE_MARKERS,
+ node_ids=node_ids,
+ )
+ warnings.extend(metric_warnings)
+ if not include_queues:
+ for node in nodes:
+ node["metrics"] = {
+ name: value
+ for name, value in node["metrics"].items()
+ if "queue" not in name.casefold()
+ }
+ if scope:
+ scope_folded = scope.casefold()
+ for node in nodes:
+ node["metrics"] = {
+ name: value
+ for name, value in node["metrics"].items()
+ if scope_folded in name.casefold()
+ }
+ if not any(node["metrics"] for node in nodes):
+ raise ClusterRuntimeFailure(
+ "No real Doris cache metrics were observed.",
+ reason_code="CACHE_METRICS_UNAVAILABLE",
+ status_code=503,
+ retryable=True,
+ )
+ return _result(
+ {"mode": "metrics", "nodes": nodes},
+ source="doris_be_metrics",
+ warnings=warnings,
+ )
+
+ async def get_compaction_status(
+ self,
+ *,
+ database: str | None,
+ table: str | None,
+ state: str | None,
+ limit: int | None,
+ ) -> dict[str, Any]:
+ max_rows = _bounded_limit(limit, default=100)
+ try:
+ rows = await self._execute(
+ "SELECT * FROM information_schema.doris_be_compaction_tasks",
+ max_rows=max_rows + 1,
+ )
+ except ClusterRuntimeFailure as exc:
+ native_failure = exc
+ else:
+ items = [_normalized_row(row) for row in rows]
+ items = _filter_rows(
+ items,
+ database=database,
+ table=table,
+ state=state,
+ )
+ truncated = len(items) > max_rows
+ return _result(
+ {
+ "mode": "native_task_tracker",
+ "items": items[:max_rows],
+ "truncated": truncated,
+ },
+ source="information_schema.doris_be_compaction_tasks",
+ metadata={"native_tracker": True},
+ )
+
+ nodes, warnings = await self._metric_family(
+ markers=_COMPACTION_MARKERS,
+ node_ids=None,
+ )
+ if not any(node["metrics"] for node in nodes):
+ raise ClusterRuntimeFailure(
+ "Doris compaction evidence is unavailable.",
+ reason_code=native_failure.reason_code,
+ status_code=native_failure.status_code,
+ retryable=native_failure.retryable,
+ )
+ warnings.insert(
+ 0,
+ "Native CompactionTaskTracker is unavailable; returning aggregate "
+ "metrics without database, table, state, or task-level detail.",
+ )
+ return _result(
+ {
+ "mode": "legacy_summary",
+ "nodes": nodes,
+ "native_tracker": False,
+ },
+ source="doris_metrics_compaction_summary",
+ warnings=warnings,
+ metadata={"native_tracker": False},
+ )
+
+ async def get_workload_group_status(
+ self,
+ *,
+ name: str | None,
+ include_usage: bool,
+ ) -> dict[str, Any]:
+ rows = await self._execute("SHOW WORKLOAD GROUPS", max_rows=_MAX_ROWS)
+ items = [_normalized_row(row) for row in rows]
+ if name:
+ items = [
+ item
+ for item in items
+ if str(item.get("name", "")).casefold() == name.casefold()
+ ]
+ if not include_usage:
+ usage_markers = (
+ "current_",
+ "running_",
+ "waiting_",
+ "queue_",
+ "usage",
+ )
+ items = [
+ {
+ key: value
+ for key, value in item.items()
+ if not key.startswith(usage_markers)
+ }
+ for item in items
+ ]
+ return _result(
+ {"items": items},
+ source="SHOW WORKLOAD GROUPS",
+ metadata={"row_count": len(items)},
+ )
+
+ async def get_compute_group_status(
+ self,
+ *,
+ name: str | None,
+ include_nodes: bool,
+ include_usage: bool,
+ ) -> dict[str, Any]:
+ rows = await self._execute("SHOW COMPUTE GROUPS", max_rows=_MAX_ROWS)
+ items = [_normalized_row(row) for row in rows]
+ if name:
+ items = [
+ item
+ for item in items
+ if str(
+ item.get("name") or item.get("compute_group_name") or ""
+ ).casefold()
+ == name.casefold()
+ ]
+ if not include_nodes:
+ for item in items:
+ for key in tuple(item):
+ if "node" in key or "backend" in key:
+ item.pop(key, None)
+ if not include_usage:
+ for item in items:
+ for key in tuple(item):
+ if any(
+ marker in key
+ for marker in ("usage", "cpu", "memory", "running",
"queue")
+ ):
+ item.pop(key, None)
+ return _result(
+ {"items": items},
+ source="SHOW COMPUTE GROUPS",
+ metadata={"row_count": len(items)},
+ )
+
+ async def analyze_resource_growth(
+ self,
+ *,
+ resource: str | None,
+ window_days: int | None,
+ granularity: str | None,
+ ) -> dict[str, Any]:
+ days = _bounded_limit(window_days, default=30, maximum=3650)
+ bucket = granularity or "day"
+ requested = (
+ (resource,) if resource else ("storage", "query_volume",
"user_activity")
+ )
+ series: dict[str, list[dict[str, Any]]] = {}
+ evidence: list[dict[str, Any]] = []
+ warnings: list[str] = []
+ for resource_name in requested:
+ sql, params = _growth_sql(resource_name, days, bucket)
+ try:
+ rows = await self._execute(
+ sql,
+ params=params,
+ max_rows=_MAX_ROWS,
+ )
+ except ClusterRuntimeFailure as exc:
+ warnings.append(
+ f"{resource_name} history is unavailable
({exc.reason_code})."
+ )
+ evidence.append(
+ {
+ "resource": resource_name,
+ "success": False,
+ "reason_code": exc.reason_code,
+ }
+ )
+ continue
+ points = [
+ {
+ "bucket": _row_lookup(row, "bucket"),
+ "value": _number(_row_lookup(row, "value")),
+ }
+ for row in rows
+ ]
+ series[resource_name] = points
+ evidence.append(
+ {
+ "resource": resource_name,
+ "success": True,
+ "source": _growth_source(resource_name),
+ "points": len(points),
+ }
+ )
+ if resource_name == "storage":
+ warnings.append(
+ "Storage growth uses partition creation evidence inside
the "
+ "window, not synthesized historical cluster snapshots."
+ )
+ if not series:
+ raise ClusterRuntimeFailure(
+ "Recorded Doris resource history is unavailable.",
+ reason_code="RESOURCE_HISTORY_UNAVAILABLE",
+ status_code=503,
+ retryable=True,
+ )
+ summaries = {name: _growth_summary(points) for name, points in
series.items()}
+ return _result(
+ {
+ "window_days": days,
+ "granularity": bucket,
+ "series": series,
+ "summaries": summaries,
+ },
+ source="recorded_doris_metadata",
+ warnings=warnings,
+ evidence=evidence,
+ metadata={"invented_history": False},
+ )
+
+ async def _metric_family(
+ self,
+ *,
+ markers: Sequence[str],
+ node_ids: Sequence[str] | None,
+ ) -> tuple[list[dict[str, Any]], list[str]]:
+ raw = await self._monitoring_tools.get_monitoring_metrics(
+ role="be",
+ monitor_type="all",
+ priority="all",
+ info_only=False,
+ include_raw_metrics=True,
+ format_type="prometheus",
+ )
+ if raw.get("success") is False:
+ return [], ["BE monitoring metrics are unavailable."]
+ nodes, warnings = _monitoring_nodes(
+ raw,
+ metric_names=None,
+ node_ids=node_ids,
+ markers=markers,
+ )
+ return nodes, warnings
+
+ async def _execute(
+ self,
+ sql: str,
+ *,
+ params: Mapping[str, Any] | tuple[Any, ...] | None = None,
+ max_rows: int,
+ ) -> list[Mapping[str, Any]]:
+ auth_context = get_current_auth_context()
+ session_id = f"{self._session_prefix}:{uuid.uuid4().hex[:8]}"
+ try:
+ async with (
+
self._connection_manager.get_connection_context_for_auth_context(
+ session_id,
+ auth_context,
+ ) as connection
+ ):
+ result = await connection.execute(
+ sql,
+ params=params,
+ auth_context=auth_context,
+ mask_result=False,
+ max_rows=max_rows,
+ max_bytes=2 * 1024 * 1024,
+ )
+ except Exception as exc:
+ raise _classify_failure(exc) from exc
+ rows: list[Mapping[str, Any]] = [
+ row for row in (result.data or ()) if isinstance(row, Mapping)
+ ]
+ return rows[:max_rows]
+
+
+def _result(
+ data: Mapping[str, Any],
+ *,
+ source: str,
+ warnings: Sequence[str] = (),
+ metadata: Mapping[str, Any] | None = None,
+ evidence: Sequence[Mapping[str, Any]] | None = None,
+) -> dict[str, Any]:
+ unique_warnings = list(dict.fromkeys(str(item) for item in warnings))
+ response: dict[str, Any] = {
+ "status": "partial" if unique_warnings else "success",
+ "data": redact_sensitive_data(dict(data)),
+ "warnings": unique_warnings,
+ "metadata": {
+ "source": source,
+ **dict(metadata or {}),
+ },
+ }
+ if evidence is not None:
+ response["evidence"] = [redact_sensitive_data(dict(item)) for item in
evidence]
+ return response
+
+
+def _collection(
+ items: Sequence[Mapping[str, Any]],
+ *,
+ source: str,
+ warnings: Sequence[str] = (),
+ truncated: bool = False,
+) -> dict[str, Any]:
+ return _result(
+ {
+ "items": [dict(item) for item in items],
+ "next_cursor": None,
+ "truncated": truncated,
+ },
+ source=source,
+ warnings=warnings,
+ metadata={"returned_items": len(items)},
+ )
+
+
+def _node_item(node_type: str, row: Mapping[str, Any]) -> dict[str, Any]:
+ normalized = _normalized_row(row)
+ node_id = next(
+ (
+ normalized.get(key)
+ for key in (
+ "backend_id",
+ "frontend_id",
+ "broker_id",
+ "name",
+ "host",
+ )
+ if normalized.get(key) not in (None, "")
+ ),
+ None,
+ )
+ host = next(
+ (
+ normalized.get(key)
+ for key in ("host", "ip", "host_name", "hostname")
+ if normalized.get(key) not in (None, "")
+ ),
+ None,
+ )
+ return {
+ "node_type": node_type,
+ "node_id": None if node_id is None else str(node_id),
+ "host": None if host is None else str(host),
+ **normalized,
+ }
+
+
+def _normalized_row(row: Mapping[str, Any]) -> dict[str, Any]:
+ return {_normalize_key(str(key)): _safe_value(value) for key, value in
row.items()}
+
+
+def _normalize_key(value: str) -> str:
+ snake = (
+ value.casefold()
+ if value.upper() == value
+ else _SNAKE_BOUNDARY.sub("_", value).casefold()
+ )
+ return _SAFE_KEY.sub("_", snake).strip("_")
+
+
+def _safe_value(value: Any) -> Any:
+ if isinstance(value, datetime):
+ return value.isoformat()
+ if isinstance(value, bytes):
+ return value.decode("utf-8", errors="replace")
+ return value
+
+
+def _monitoring_nodes(
+ raw: Mapping[str, Any],
+ *,
+ metric_names: Sequence[str] | None,
+ node_ids: Sequence[str] | None,
+ markers: Sequence[str] | None = None,
+) -> tuple[list[dict[str, Any]], list[str]]:
+ data = raw.get("data")
+ if not isinstance(data, Mapping):
+ return [], ["Monitoring response did not contain metric data."]
+ allowed_nodes = {str(value) for value in (node_ids or ())}
+ requested_names = set(metric_names or ())
+ nodes: list[dict[str, Any]] = []
+ warnings: list[str] = []
+ be_data = data.get("be")
+ be_candidates = be_data if isinstance(be_data, list) else []
+ role_values: tuple[tuple[str, list[Any]], ...] = (
+ ("fe", [data.get("fe")]),
+ ("be", be_candidates),
+ )
+ for role, candidates in role_values:
+ for candidate in candidates:
+ if not isinstance(candidate, Mapping):
+ continue
+ if candidate.get("success") is False:
+ warnings.append(f"{role.upper()} metrics endpoint was
unavailable.")
+ continue
+ node_info = candidate.get("node_info")
+ node_info = node_info if isinstance(node_info, Mapping) else {}
+ node_id = (
+ node_info.get("backend_id")
+ or node_info.get("name")
+ or node_info.get("host")
+ or role
+ )
+ host = node_info.get("host")
+ if (
+ allowed_nodes
+ and not {
+ str(node_id),
+ str(host),
+ }
+ & allowed_nodes
+ ):
+ continue
+ metrics = candidate.get("raw_metrics")
+ if not isinstance(metrics, Mapping):
+ metrics = candidate.get("metrics")
+ if not isinstance(metrics, Mapping):
+ metrics = {}
+ selected: dict[str, Any] = {}
+ for name, value in metrics.items():
+ metric_name = str(name)
+ if requested_names and metric_name not in requested_names:
+ continue
+ if markers and not any(
+ marker in metric_name.casefold() for marker in markers
+ ):
+ continue
+ selected[metric_name] = _safe_value(value)
+ if len(selected) >= _MAX_METRICS_PER_NODE:
+ warnings.append(
+ f"{role.upper()} metric output was truncated to "
+ f"{_MAX_METRICS_PER_NODE} names per node."
+ )
+ break
+ nodes.append(
+ {
+ "role": role,
+ "node_id": str(node_id),
+ "host": None if host is None else str(host),
+ "metrics": selected,
+ }
+ )
+ return nodes, list(dict.fromkeys(warnings))
+
+
+def _metric_total(value: Any) -> float:
+ if isinstance(value, int | float):
+ return float(value)
+ if isinstance(value, list):
+ return sum(
+ _metric_total(item.get("value"))
+ for item in value
+ if isinstance(item, Mapping)
+ )
+ return 0.0
+
+
+def _task_state(item: Mapping[str, Any]) -> str:
+ return str(
+ item.get("query_status")
+ or item.get("state")
+ or item.get("status")
+ or item.get("command")
+ or ""
+ )
+
+
+def _filter_rows(
+ rows: Sequence[dict[str, Any]],
+ *,
+ database: str | None,
+ table: str | None,
+ state: str | None,
+) -> list[dict[str, Any]]:
+ filters = tuple(value.casefold() for value in (database, table, state) if
value)
+ if not filters:
+ return list(rows)
+ return [
+ row
+ for row in rows
+ if all(
+ expected in " ".join(str(value) for value in
row.values()).casefold()
+ for expected in filters
+ )
+ ]
+
+
+def _growth_sql(
+ resource: str,
+ days: int,
+ granularity: str,
+) -> tuple[str, tuple[int]]:
+ try:
+ return _GROWTH_SQL[(resource, granularity)], (days,)
+ except KeyError:
+ if resource not in {"storage", "query_volume", "user_activity"}:
+ raise _argument_failure(
+ "resource must be storage, query_volume, or user_activity."
+ ) from None
+ raise _argument_failure("granularity must be hour, day, or week.")
from None
+
+
+def _growth_source(resource: str) -> str:
+ if resource == "storage":
+ return "information_schema.partitions"
+ return "internal.__internal_schema.audit_log"
+
+
+def _growth_summary(points: Sequence[Mapping[str, Any]]) -> dict[str, Any]:
+ values = [
+ float(value)
+ for point in points
+ if isinstance((value := point.get("value")), int | float)
+ ]
+ if not values:
+ return {"point_count": 0, "growth_percent": None}
+ growth = None
+ if len(values) > 1 and values[0] != 0:
+ growth = round((values[-1] - values[0]) / values[0] * 100, 4)
+ return {
+ "point_count": len(values),
+ "first_value": values[0],
+ "last_value": values[-1],
+ "growth_percent": growth,
+ }
+
+
+def _row_lookup(row: Mapping[str, Any], name: str) -> Any:
+ expected = name.casefold()
+ return next(
+ (value for key, value in row.items() if str(key).casefold() ==
expected),
+ None,
+ )
+
+
+def _number(value: Any) -> int | float | None:
+ if value is None:
+ return None
+ if isinstance(value, int | float):
+ return value
+ try:
+ parsed = float(value)
+ except (TypeError, ValueError):
+ return None
+ return int(parsed) if parsed.is_integer() else parsed
+
+
+def _bounded_limit(
+ value: int | None,
+ *,
+ default: int,
+ maximum: int = _MAX_ROWS,
+) -> int:
+ if value is None:
+ return default
+ if isinstance(value, bool) or not isinstance(value, int) or value < 1:
+ raise _argument_failure("limit must be a positive integer.")
+ return min(value, maximum)
+
+
+def _argument_failure(message: str) -> ClusterRuntimeFailure:
+ return ClusterRuntimeFailure(
+ message,
+ reason_code="CLUSTER_ARGUMENT_INVALID",
+ status_code=400,
+ )
+
+
+def _classify_failure(exc: Exception) -> ClusterRuntimeFailure:
+ if isinstance(exc, ClusterRuntimeFailure):
+ return exc
+ numeric_code = next(
+ (value for value in getattr(exc, "args", ()) if isinstance(value,
int)),
+ None,
+ )
+ message = str(exc).casefold()
+ if numeric_code in {1044, 1045, 1142, 1227} or any(
+ marker in message
+ for marker in ("access denied", "permission denied", "privilege")
+ ):
+ return ClusterRuntimeFailure(
+ "Doris denied access to cluster metadata.",
+ reason_code="CLUSTER_PERMISSION_DENIED",
+ status_code=403,
+ )
+ if numeric_code in {1064, 1109, 1146} or any(
+ marker in message
+ for marker in (
+ "doesn't exist",
+ "does not exist",
+ "not supported",
+ "unsupported",
+ "unknown table",
+ )
+ ):
+ return ClusterRuntimeFailure(
+ "The requested Doris cluster evidence source is unsupported.",
+ reason_code="CLUSTER_SOURCE_UNSUPPORTED",
+ status_code=501,
+ )
+ if isinstance(exc, TimeoutError | ConnectionError | OSError):
+ return ClusterRuntimeFailure(
+ "Doris cluster evidence is temporarily unavailable.",
+ reason_code="CLUSTER_BACKEND_UNAVAILABLE",
+ status_code=503,
+ retryable=True,
+ )
+ return ClusterRuntimeFailure(
+ "Doris cluster metadata execution failed.",
+ reason_code="CLUSTER_EXECUTION_FAILED",
+ status_code=502,
+ )
+
+
+__all__ = ["ClusterRuntimeFailure", "DorisClusterRuntime"]
diff --git a/doris_mcp_server/utils/db.py b/doris_mcp_server/utils/db.py
index 4cb03a0..934cb4a 100644
--- a/doris_mcp_server/utils/db.py
+++ b/doris_mcp_server/utils/db.py
@@ -2618,6 +2618,23 @@ class DorisConnectionManager:
raw_connection,
"unhealthy routed connection",
)
+ if owner_pool is not None:
+ try:
+ owner_pool.release(raw_connection)
+ self.logger.debug(
+ "Discarded unhealthy %s connection for route=%s
owner=%s",
+ getattr(connection, "pool_kind", ""),
+ getattr(connection, "route_key", ""),
+ getattr(connection, "owner_id", ""),
+ )
+ except Exception as release_error:
+ self.logger.warning(
+ "Failed to discard unhealthy connection for route=%s "
+ "owner=%s: %s",
+ getattr(connection, "route_key", ""),
+ getattr(connection, "owner_id", ""),
+ release_error,
+ )
return
if owner_pool is None:
self.logger.warning(
diff --git a/test/integration/test_real_doris_transports.py
b/test/integration/test_real_doris_transports.py
index 48e1236..eb9f40f 100644
--- a/test/integration/test_real_doris_transports.py
+++ b/test/integration/test_real_doris_transports.py
@@ -71,6 +71,19 @@ pytestmark = [
PROJECT_ROOT = Path(__file__).resolve().parents[2]
SAFE_IDENTIFIER = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
+CLUSTER_CHILD_NAMES = (
+ "get_cluster_overview",
+ "list_cluster_nodes",
+ "list_active_tasks",
+ "get_monitoring_metrics",
+ "get_memory_stats",
+ "get_cache_status",
+ "get_compaction_status",
+ "get_workload_group_status",
+ "get_compute_group_status",
+ "analyze_resource_growth",
+ "get_runtime_capabilities",
+)
@dataclass(frozen=True)
@@ -419,6 +432,35 @@ async def _exec_query(
return result, result.structured_content
+async def _call_domain_child(
+ client: Client,
+ *,
+ domain: str,
+ child_tool: str,
+ arguments: dict[str, Any],
+ manifest_version: str,
+) -> dict[str, Any]:
+ result = await client.call_tool(
+ domain,
+ {
+ "child_tool": child_tool,
+ "arguments": arguments,
+ "manifest_version": manifest_version,
+ },
+ )
+ assert result.is_error is False, json.dumps(
+ result.model_dump(by_alias=True, mode="json"),
+ ensure_ascii=False,
+ )
+ assert isinstance(result.structured_content, dict)
+ payload = result.structured_content
+ assert payload["mode"] == "result"
+ assert payload["domain"] == domain
+ assert payload["child_tool"] == child_tool
+ assert isinstance(payload["data"], dict)
+ return payload["data"]
+
+
@pytest.mark.parametrize("transport", ["http", "stdio"])
async def test_real_doris_result_boundaries_and_cancellation(
transport: str,
@@ -1045,6 +1087,107 @@ async def test_real_doris_tool_regression_paths(
assert recovered_payload["data"][0]["recovered"] == 1
[email protected](
+ os.getenv("DORIS_REAL_HTTP_INTEGRATION") != "1",
+ reason="set DORIS_REAL_HTTP_INTEGRATION=1 with independent FE/BE HTTP
endpoints",
+)
[email protected]("transport", ["http", "stdio"])
+async def test_real_doris_hierarchical_cluster_domain_is_read_only_and_live(
+ transport: str,
+) -> None:
+ settings = _real_doris_settings()
+ environment = _server_environment(
+ settings,
+ user=settings.user,
+ password=settings.password,
+ )
+ environment["MCP_TOOL_EXPOSURE_MODE"] = "hierarchical"
+ assert environment.get("DORIS_FE_HTTP_HOST")
+ assert environment.get("DORIS_BE_HOSTS")
+
+ async with _transport_client(
+ transport,
+ environment,
+ read_timeout_seconds=60,
+ ) as client:
+ tools = {
+ tool.name: tool
+ for tool in (await client.list_tools(cache_mode="bypass")).tools
+ }
+ assert "doris_catalog" in tools
+ assert "doris_cluster" in tools
+
+ catalog_result = await client.call_tool("doris_catalog", {})
+ assert catalog_result.is_error is False
+ assert isinstance(catalog_result.structured_content, dict)
+ assert catalog_result.structured_content["mode"] == "manifest"
+ assert catalog_result.structured_content["domain"] == "doris_catalog"
+
+ cluster_result = await client.call_tool("doris_cluster", {})
+ assert cluster_result.is_error is False, json.dumps(
+ cluster_result.model_dump(by_alias=True, mode="json"),
+ ensure_ascii=False,
+ )
+ assert isinstance(cluster_result.structured_content, dict)
+ cluster_manifest = cluster_result.structured_content
+ assert cluster_manifest["mode"] == "manifest"
+ assert cluster_manifest["domain"] == "doris_cluster"
+ children = {
+ child["name"]: child
+ for child in cluster_manifest["children"]
+ }
+ assert tuple(children) == CLUSTER_CHILD_NAMES
+ manifest_version = cluster_manifest["manifest_version"]
+
+ runtime = await _call_domain_child(
+ client,
+ domain="doris_cluster",
+ child_tool="get_runtime_capabilities",
+ arguments={"detail": "full"},
+ manifest_version=manifest_version,
+ )
+ master_fe_version = runtime["data"]["versions"]["master_fe"]
+ assert master_fe_version
+
+ nodes = await _call_domain_child(
+ client,
+ domain="doris_cluster",
+ child_tool="list_cluster_nodes",
+ arguments={"node_types": ["fe", "be"], "include_metrics": False},
+ manifest_version=manifest_version,
+ )
+ assert nodes["data"]["items"]
+ assert {"fe", "be"}.issubset(
+ {item["node_type"] for item in nodes["data"]["items"]}
+ )
+
+ memory = await _call_domain_child(
+ client,
+ domain="doris_cluster",
+ child_tool="get_memory_stats",
+ arguments={"detail": "summary"},
+ manifest_version=manifest_version,
+ )
+ assert memory["metadata"]["invented_values"] is False
+
+ compaction_availability =
children["get_compaction_status"]["availability"]
+ if master_fe_version.startswith("4.0.5"):
+ assert compaction_availability["callable"] is True
+ assert (
+ compaction_availability["active_variant"]
+ == "legacy_compaction_summary"
+ )
+ compaction = await _call_domain_child(
+ client,
+ domain="doris_cluster",
+ child_tool="get_compaction_status",
+ arguments={"limit": 20},
+ manifest_version=manifest_version,
+ )
+ assert compaction["data"]["mode"] == "legacy_summary"
+ assert compaction["data"]["native_tracker"] is False
+
+
@pytest.mark.skipif(
os.getenv("DORIS_REAL_HTTP_INTEGRATION") != "1",
reason="set DORIS_REAL_HTTP_INTEGRATION=1 with independent FE/BE HTTP
endpoints",
diff --git a/test/tools/test_capability_detector.py
b/test/tools/test_capability_detector.py
index c452ffa..6124029 100644
--- a/test/tools/test_capability_detector.py
+++ b/test/tools/test_capability_detector.py
@@ -34,6 +34,7 @@ from doris_mcp_server.tools.capability_detector import (
)
from doris_mcp_server.tools.doris_feature_matrix import DORIS_FEATURE_MATRIX
from doris_mcp_server.utils.db import DorisRouteIdentity
+from doris_mcp_server.utils.doris_http_client import DorisHTTPResponse
from doris_mcp_server.utils.security import AuthContext
@@ -475,6 +476,42 @@ async def
test_detector_retains_visible_backend_with_unknown_version() -> None:
assert evaluation.reason_code == "DORIS_VERSION_UNKNOWN"
[email protected]
+async def test_detector_ignores_explicitly_dead_backend_for_version_gating()
-> None:
+ connection = _ProbeConnection()
+ connection.row_overrides["SHOW BACKENDS"] = [
+ {
+ "BackendId": "1",
+ "Alive": "true",
+ "Version": "doris-4.0.5-rc01-59de8c4c524",
+ },
+ {
+ "BackendId": "2",
+ "Alive": "false",
+ "Version": "",
+ },
+ ]
+ detector = DorisCapabilityDetector( # type: ignore[arg-type]
+ _ProbeConnectionManager(connection)
+ )
+
+ snapshot = await detector.detect_base(
+ None,
+ capability_generation=1,
+ provider_generation="provider.a",
+ )
+ evaluation = DORIS_FEATURE_MATRIX.evaluate(
+ domain="doris_cluster",
+ child_name="list_cluster_nodes",
+ versions=snapshot.version_vector,
+ )
+
+ assert tuple(
+ version.normalized for version in snapshot.version_vector.backends
+ ) == ("4.0.5rc1",)
+ assert evaluation.compatible is True
+
+
@pytest.mark.asyncio
async def
test_detector_uses_fallback_for_unknown_master_and_retains_follower() -> None:
connection = _ProbeConnection()
@@ -640,3 +677,105 @@ async def
test_detector_wraps_domain_connection_failures() -> None:
match="probe failed for doris_query",
):
await detector.detect_domain(base, "doris_query", None)
+
+
[email protected]
+async def test_cluster_probe_keeps_405_compaction_on_real_legacy_evidence(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ connection = _ProbeConnection()
+ manager = _ProbeConnectionManager(connection)
+ database_config = SimpleNamespace(
+ user="root",
+ password="secret",
+ host="fe-1",
+ hosts=["fe-1"],
+ fe_http_host="fe-1",
+ fe_http_hosts=["fe-1"],
+ fe_http_port=8030,
+ be_hosts=["be-1"],
+ be_webserver_port=8040,
+ )
+ manager.config.database = database_config
+ manager.selected_database_config = database_config
+
+ native_compaction = (
+ "SELECT * "
+ "FROM information_schema.doris_be_compaction_tasks LIMIT 1"
+ )
+ connection.failures[native_compaction] = RuntimeError(
+ 1146,
+ "table does not exist",
+ )
+ connection.row_overrides.update(
+ {
+ 'SHOW PROC "/current_queries"': [],
+ (
+ "SELECT BE_ID, METRIC_NAME "
+ "FROM information_schema.file_cache_statistics LIMIT 1"
+ ): [{"BE_ID": 1, "METRIC_NAME": "hits_ratio"}],
+ "SHOW WORKLOAD GROUPS": [],
+ "SHOW COMPUTE GROUPS": [],
+ (
+ "SELECT `time` "
+ "FROM internal.__internal_schema.audit_log LIMIT 1"
+ ): [],
+ }
+ )
+
+ class _HTTPClient:
+ async def get_first_available(
+ self,
+ *,
+ role: str,
+ **_kwargs: Any,
+ ) -> DorisHTTPResponse:
+ metrics = (
+ "doris_fe_tablet_max_compaction_score 4\n"
+ if role == "fe"
+ else (
+ "doris_be_memory_allocated_bytes 1024\n"
+ "doris_be_file_cache_hits_ratio 0.8\n"
+ "doris_be_tablet_base_max_compaction_score 7\n"
+ )
+ )
+ return DorisHTTPResponse(
+ status=200,
+ headers={},
+ body=metrics.encode(),
+ url=f"http://{role}/metrics";,
+ )
+
+ monkeypatch.setattr(
+ "doris_mcp_server.tools.capability_detector."
+ "DorisHTTPClient.from_database_config",
+ lambda _config: _HTTPClient(),
+ )
+ detector = DorisCapabilityDetector(manager) # type: ignore[arg-type]
+ base = await detector.detect_base(
+ None,
+ capability_generation=1,
+ provider_generation="provider.a",
+ )
+
+ snapshot = await detector.detect_domain(base, "doris_cluster", None)
+
+ assert snapshot.version_vector.master_fe.normalized == "4.0.5rc1"
+ assert (
+ snapshot.probe("information_schema.doris_be_compaction_tasks").status
+ is CapabilityProbeStatus.UNSUPPORTED
+ )
+ assert (
+ snapshot.probe("compaction_system_table_or_http_api").status
+ is CapabilityProbeStatus.UNSUPPORTED
+ )
+ legacy = snapshot.probe("legacy_compaction_status_readable")
+ assert legacy is not None
+ assert legacy.status is CapabilityProbeStatus.DEGRADED
+ assert legacy.reason_code == "LEGACY_COMPACTION_SUMMARY_READABLE"
+ assert snapshot.probe("fe_metrics").status is
CapabilityProbeStatus.SUPPORTED
+ assert snapshot.probe("be_metrics").status is
CapabilityProbeStatus.SUPPORTED
+ assert (
+ snapshot.probe("file_cache_metrics_readable").status
+ is CapabilityProbeStatus.SUPPORTED
+ )
diff --git a/test/tools/test_capability_registry.py
b/test/tools/test_capability_registry.py
index d5fac09..ff3a6e0 100644
--- a/test/tools/test_capability_registry.py
+++ b/test/tools/test_capability_registry.py
@@ -184,6 +184,8 @@ def
test_evaluator_requires_version_probes_handler_and_call_permission() -> None
assert available.status is AvailabilityStatus.AVAILABLE
assert available.callable is True
assert available.active_variant == "mysql_read_only"
+ assert available.reason_code == "CAPABILITY_VERIFIED_UNCERTIFIED"
+ assert available.limitations == ()
assert missing_probe.status is AvailabilityStatus.UNKNOWN
assert missing_probe.reason_code == "CAPABILITY_PROBE_PENDING"
assert mixed_backends.status is AvailabilityStatus.AVAILABLE
@@ -193,6 +195,74 @@ def
test_evaluator_requires_version_probes_handler_and_call_permission() -> None
assert allowed.callable is True
+def test_compaction_prefers_native_tracker_and_uses_legacy_on_405() -> None:
+ evaluator = CapabilityEvaluator(
+ matrix=DORIS_FEATURE_MATRIX,
+ bound_handlers=_BoundHandlers(
+ "doris_cluster.get_compaction_status"
+ ), # type: ignore[arg-type]
+ )
+ domain = DORIS_DOMAIN_CATALOG.resolve_domain("doris_cluster")
+ child = DORIS_DOMAIN_CATALOG.resolve_child(
+ "doris_cluster",
+ "get_compaction_status",
+ )
+ probes = {
+ "information_schema.doris_be_compaction_tasks": (
+ CapabilityProbeEvidence(
+ probe_id="information_schema.doris_be_compaction_tasks",
+ status=CapabilityProbeStatus.SUPPORTED,
+ reason_code="TEST_NATIVE_TRACKER",
+ )
+ ),
+ "compaction_system_table_or_http_api": CapabilityProbeEvidence(
+ probe_id="compaction_system_table_or_http_api",
+ status=CapabilityProbeStatus.SUPPORTED,
+ reason_code="TEST_NATIVE_TRACKER",
+ ),
+ "legacy_compaction_status_readable": CapabilityProbeEvidence(
+ probe_id="legacy_compaction_status_readable",
+ status=CapabilityProbeStatus.DEGRADED,
+ reason_code="TEST_LEGACY_SUMMARY",
+ ),
+ }
+ providers = CapabilityProviderRegistry({}).snapshot()
+ snapshot_405 = _snapshot(probes=probes)
+ snapshot_406 = replace(
+ snapshot_405,
+ version_vector=DorisClusterVersionVector.from_comments(
+ master_fe="Doris version doris-4.0.6",
+ follower_fes=("Doris version doris-4.0.6",),
+ backends=("Doris version doris-4.0.6",),
+ ),
+ )
+
+ native = evaluator.evaluate(
+ snapshot=snapshot_406,
+ providers=providers,
+ domain=domain,
+ child=child,
+ auth_context=None,
+ )
+ legacy = evaluator.evaluate(
+ snapshot=snapshot_405,
+ providers=providers,
+ domain=domain,
+ child=child,
+ auth_context=None,
+ )
+
+ assert native.callable is True
+ assert native.status is AvailabilityStatus.AVAILABLE
+ assert native.active_variant == "compaction_task_tracker"
+ assert legacy.callable is True
+ assert legacy.status is AvailabilityStatus.DEGRADED
+ assert legacy.active_variant == "legacy_compaction_summary"
+ assert legacy.reason_code == (
+ "CAPABILITY_VERIFIED_DEGRADED_UNCERTIFIED"
+ )
+
+
def test_evaluator_normalizes_system_object_probe_evidence_for_manifest() ->
None:
evaluator = CapabilityEvaluator(
matrix=DORIS_FEATURE_MATRIX,
diff --git a/test/tools/test_domain_dispatcher.py
b/test/tools/test_domain_dispatcher.py
index 9466998..83b7738 100644
--- a/test/tools/test_domain_dispatcher.py
+++ b/test/tools/test_domain_dispatcher.py
@@ -31,6 +31,7 @@ from doris_mcp_server.tools.domain_catalog import (
DORIS_DOMAIN_CATALOG,
)
from doris_mcp_server.tools.domain_dispatcher import (
+ BoundHandlerAvailabilityProvider,
ChildArgumentsAdapterError,
DomainDispatcher,
ToolExposureMode,
@@ -137,6 +138,52 @@ def
test_dispatcher_registers_exactly_47_formal_flat_names() -> None:
assert not manager.domain_dispatcher.handles_flat("exec_query")
+def test_cluster_domain_binds_all_eleven_children() -> None:
+ manager = _manager()
+ bound = BoundHandlerAvailabilityProvider(manager)
+ cluster = DORIS_DOMAIN_CATALOG.resolve_domain("doris_cluster")
+
+ assert len(cluster.children) == 11
+ assert all(
+ bound.is_bound(cluster.name, child.name)
+ for child in cluster.children
+ )
+
+
[email protected]
+async def test_cluster_metric_filters_reach_strict_runtime_unchanged() -> None:
+ manager = _manager("doris_cluster.get_monitoring_metrics")
+ manager.cluster_runtime.get_monitoring_metrics = AsyncMock(
+ return_value={
+ "status": "success",
+ "data": {"nodes": []},
+ "warnings": [],
+ "metadata": {"source": "test_metrics"},
+ }
+ )
+
+ response = json.loads(
+ await manager.call_tool(
+ "doris_cluster",
+ {
+ "child_tool": "get_monitoring_metrics",
+ "arguments": {
+ "metric_names": ["doris_be_cpu"],
+ "node_ids": ["be-1"],
+ "window": "instant",
+ },
+ },
+ )
+ )
+
+ assert response["mode"] == "result"
+ manager.cluster_runtime.get_monitoring_metrics.assert_awaited_once_with(
+ metric_names=["doris_be_cpu"],
+ node_ids=["be-1"],
+ window="instant",
+ )
+
+
@pytest.mark.asyncio
async def test_default_manager_fails_closed_without_capability_snapshot() ->
None:
manager = DorisToolsManager(_connection_manager())
@@ -476,12 +523,12 @@ async def
test_manifest_version_unavailable_and_unbound_fail_closed() -> None:
},
)
)
- unbound_manager = _manager("doris_cluster.list_cluster_nodes")
+ unbound_manager = _manager("doris_pipeline.get_ingestion_status")
unbound = json.loads(
await unbound_manager.call_tool(
- "doris_cluster",
+ "doris_pipeline",
{
- "child_tool": "list_cluster_nodes",
+ "child_tool": "get_ingestion_status",
"arguments": {},
},
)
@@ -1558,14 +1605,14 @@ def
test_detail_result_normalization_adds_formal_evidence() -> None:
(
"adapt:monitoring_arguments",
{"metric_names": ["qps"]},
- {},
- 1,
+ {"metric_names": ["qps"]},
+ 0,
),
(
"adapt:memory_arguments",
{"detail": "trackers", "node_ids": ["be-1"]},
- {"data_type": "realtime", "tracker_type": "all"},
- 1,
+ {"detail": "trackers", "node_ids": ["be-1"]},
+ 0,
),
(
"adapt:audit_filters",
@@ -1683,15 +1730,16 @@ def
test_detail_result_normalization_adds_formal_evidence() -> None:
(
"adapt:resource_growth_arguments",
{
- "resource": "memory",
+ "resource": "storage",
"window_days": 9,
"granularity": "day",
},
{
- "days": 9,
- "resource_types": ["memory"],
+ "resource": "storage",
+ "window_days": 9,
+ "granularity": "day",
},
- 1,
+ 0,
),
(
"adapt:adbc_query_arguments",
diff --git a/test/tools/test_domain_manifest.py
b/test/tools/test_domain_manifest.py
index d53836a..9cef6f8 100644
--- a/test/tools/test_domain_manifest.py
+++ b/test/tools/test_domain_manifest.py
@@ -304,15 +304,19 @@ async def
test_detected_version_has_priority_in_dynamic_description() -> None:
async def test_realistic_runtime_evidence_stays_within_manifest_budget() ->
None:
provider = StaticAvailabilityProvider(
_availability(
+ status=AvailabilityStatus.DEGRADED,
+ reason_code="CAPABILITY_VERIFIED_DEGRADED_UNCERTIFIED",
+ active_variant="legacy_compaction_summary",
detected_versions={
- "master_fe": ("4.1.3",),
- "fe": ("4.1.3",),
- "be": ("4.1.3",),
+ "master_fe": ("4.0.5rc1",),
+ "be": ("4.0.5rc1",),
},
evidence_sources=(
"version_probe",
"sql_probe",
"provider_config",
+ "show_frontends",
+ "show_backends",
),
)
)
diff --git a/test/tools/test_manager_routing_boundaries.py
b/test/tools/test_manager_routing_boundaries.py
index 75d7695..f8131fa 100644
--- a/test/tools/test_manager_routing_boundaries.py
+++ b/test/tools/test_manager_routing_boundaries.py
@@ -18,7 +18,7 @@
"""Branch coverage for the manager's thin routing helpers."""
from types import SimpleNamespace
-from unittest.mock import AsyncMock, call
+from unittest.mock import AsyncMock
import pytest
@@ -35,15 +35,22 @@ def _manager() -> DorisToolsManager:
]
)
)
- manager.memory_tracker = SimpleNamespace(
- get_realtime_memory_stats=AsyncMock(
+ manager.cluster_runtime = SimpleNamespace(
+ get_monitoring_metrics=AsyncMock(
return_value={
- "success": True,
- "timestamp": "2026-07-30T00:00:00Z",
+ "status": "success",
+ "data": {"nodes": []},
+ "warnings": [],
+ "metadata": {"source": "test"},
}
),
- get_historical_memory_stats=AsyncMock(
- return_value={"success": True, "points": []}
+ get_memory_stats=AsyncMock(
+ return_value={
+ "status": "success",
+ "data": {"nodes": []},
+ "warnings": [],
+ "metadata": {"source": "test"},
+ }
),
)
manager.data_governance_tools = SimpleNamespace(
@@ -59,119 +66,41 @@ def _manager() -> DorisToolsManager:
@pytest.mark.asyncio
-async def test_monitoring_metrics_routes_definitions_data_and_both():
- definitions_manager = _manager()
- definitions_manager.monitoring_tools.get_monitoring_metrics.side_effect =
None
- definitions_manager.monitoring_tools.get_monitoring_metrics.return_value =
{
- "success": True
- }
- definitions = await definitions_manager._get_monitoring_metrics_tool(
- {
- "content_type": "definitions",
- "role": "fe",
- "monitor_type": "system",
- "priority": "all",
- }
- )
+async def test_monitoring_metrics_routes_formal_filters_to_cluster_runtime():
+ manager = _manager()
- data_manager = _manager()
- data_manager.monitoring_tools.get_monitoring_metrics.side_effect = None
- data_manager.monitoring_tools.get_monitoring_metrics.return_value = {
- "success": True
- }
- data = await data_manager._get_monitoring_metrics_tool(
+ result = await manager._get_monitoring_metrics_tool(
{
- "content_type": "data",
- "role": "be",
- "monitor_type": "query",
- "priority": "core",
- "include_raw_metrics": True,
+ "metric_names": ["doris_be_cpu"],
+ "node_ids": ["be-1"],
+ "window": "instant",
}
)
- both_manager = _manager()
- both = await both_manager._get_monitoring_metrics_tool({"content_type":
"both"})
-
- assert definitions == {"success": True}
-
definitions_manager.monitoring_tools.get_monitoring_metrics.assert_awaited_once_with(
- "fe", "system", "all", info_only=True, format_type="prometheus"
- )
- assert data == {"success": True}
-
data_manager.monitoring_tools.get_monitoring_metrics.assert_awaited_once_with(
- "be",
- "query",
- "core",
- info_only=False,
- format_type="prometheus",
- include_raw_metrics=True,
+ assert result["status"] == "success"
+ manager.cluster_runtime.get_monitoring_metrics.assert_awaited_once_with(
+ metric_names=["doris_be_cpu"],
+ node_ids=["be-1"],
+ window="instant",
)
- assert both["content_type"] == "both"
- assert both["timestamp"] == "2026-07-30T00:00:00Z"
- assert both["_execution_info"] == {
- "combined_response": True,
- "definitions_available": True,
- "data_available": True,
- }
@pytest.mark.asyncio
-async def test_monitoring_metrics_rejects_unknown_content_type():
+async def test_memory_stats_routes_formal_detail_to_cluster_runtime():
manager = _manager()
- result = await manager._get_monitoring_metrics_tool({"content_type":
"unknown"})
-
- assert "Invalid content_type" in result["error"]
- manager.monitoring_tools.get_monitoring_metrics.assert_not_awaited()
-
-
[email protected]
-async def test_memory_stats_routes_realtime_historical_and_both():
- manager = _manager()
-
- realtime = await manager._get_memory_stats_tool(
- {
- "data_type": "realtime",
- "tracker_type": "process",
- "include_details": False,
- }
- )
- historical = await manager._get_memory_stats_tool(
+ result = await manager._get_memory_stats_tool(
{
- "data_type": "historical",
- "tracker_names": ["query"],
- "time_range": "24h",
+ "detail": "top_consumers",
+ "node_ids": ["be-1"],
}
)
- both = await manager._get_memory_stats_tool({"data_type": "both"})
-
- assert realtime["success"] is True
- assert historical == {"success": True, "points": []}
- assert both["data_type"] == "both"
- assert both["timestamp"] == "2026-07-30T00:00:00Z"
- assert both["_execution_info"] == {
- "combined_response": True,
- "realtime_available": True,
- "historical_available": True,
- }
- assert manager.memory_tracker.get_realtime_memory_stats.await_args_list ==
[
- call("process", False),
- call("overview", True),
- ]
- assert manager.memory_tracker.get_historical_memory_stats.await_args_list
== [
- call(["query"], "24h"),
- call(None, "1h"),
- ]
-
[email protected]
-async def test_memory_stats_rejects_unknown_data_type():
- manager = _manager()
-
- result = await manager._get_memory_stats_tool({"data_type": "unknown"})
-
- assert "Invalid data_type" in result["error"]
- manager.memory_tracker.get_realtime_memory_stats.assert_not_awaited()
- manager.memory_tracker.get_historical_memory_stats.assert_not_awaited()
+ assert result["status"] == "success"
+ manager.cluster_runtime.get_memory_stats.assert_awaited_once_with(
+ node_ids=["be-1"],
+ detail="top_consumers",
+ )
@pytest.mark.asyncio
diff --git a/test/utils/test_cluster_runtime.py
b/test/utils/test_cluster_runtime.py
new file mode 100644
index 0000000..08cd3b1
--- /dev/null
+++ b/test/utils/test_cluster_runtime.py
@@ -0,0 +1,313 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Production evidence contracts for the formal Cluster runtime."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from contextlib import asynccontextmanager
+from types import SimpleNamespace
+from typing import Any
+
+import pytest
+
+from doris_mcp_server.utils.cluster_runtime import (
+ ClusterRuntimeFailure,
+ DorisClusterRuntime,
+)
+
+
+class _ConnectionManager:
+ def __init__(
+ self,
+ rows: dict[str, list[dict[str, Any]]] | None = None,
+ failures: dict[str, Exception] | None = None,
+ ) -> None:
+ self.rows = rows or {}
+ self.failures = failures or {}
+ self.calls: list[str] = []
+ self.params: list[Mapping[str, Any] | tuple[Any, ...] | None] = []
+
+ @asynccontextmanager
+ async def get_connection_context_for_auth_context(
+ self,
+ _session_id: str,
+ _auth_context: Any,
+ ):
+ manager = self
+
+ class _Connection:
+ async def execute(
+ self,
+ sql: str,
+ params: Mapping[str, Any] | tuple[Any, ...] | None = None,
+ **_kwargs: Any,
+ ) -> SimpleNamespace:
+ manager.calls.append(sql)
+ manager.params.append(params)
+ if sql in manager.failures:
+ raise manager.failures[sql]
+ return SimpleNamespace(data=manager.rows.get(sql, []))
+
+ yield _Connection()
+
+
+class _MonitoringTools:
+ def __init__(self, response: dict[str, Any]) -> None:
+ self.response = response
+ self.calls: list[dict[str, Any]] = []
+
+ async def get_monitoring_metrics(self, **kwargs: Any) -> dict[str, Any]:
+ self.calls.append(kwargs)
+ return self.response
+
+
+def _metrics_response() -> dict[str, Any]:
+ return {
+ "success": True,
+ "timestamp": "2026-07-31T00:00:00Z",
+ "data": {
+ "fe": {
+ "success": True,
+ "node_info": {"host": "fe-1"},
+ "metrics": {"doris_fe_query_total": 7},
+ "raw_metrics": {
+ "doris_fe_query_total": 7,
+ "doris_fe_tablet_max_compaction_score": 3,
+ },
+ },
+ "be": [
+ {
+ "success": True,
+ "node_info": {
+ "backend_id": "1",
+ "host": "be-1",
+ },
+ "metrics": {
+ "memory_allocated_bytes": 1024,
+ },
+ "raw_metrics": {
+ "doris_be_memory_allocated_bytes": 1024,
+ "doris_be_memory_jemalloc_active_bytes": 2048,
+ "doris_be_file_cache_hits_ratio": 0.75,
+ "doris_be_tablet_base_max_compaction_score": 4,
+ },
+ }
+ ],
+ },
+ }
+
+
+def _runtime(
+ *,
+ rows: dict[str, list[dict[str, Any]]] | None = None,
+ failures: dict[str, Exception] | None = None,
+ metrics: dict[str, Any] | None = None,
+) -> tuple[DorisClusterRuntime, _ConnectionManager, _MonitoringTools]:
+ manager = _ConnectionManager(rows, failures)
+ monitoring = _MonitoringTools(metrics or _metrics_response())
+ return (
+ DorisClusterRuntime(manager, monitoring), # type: ignore[arg-type]
+ manager,
+ monitoring, # type: ignore[return-value]
+ )
+
+
[email protected]
+async def test_list_cluster_nodes_normalizes_real_fe_and_be_rows() -> None:
+ runtime, manager, _ = _runtime(
+ rows={
+ "SHOW FRONTENDS": [
+ {
+ "Name": "fe-1",
+ "Host": "10.0.0.1",
+ "IsMaster": "true",
+ "Version": "doris-4.0.5-rc01",
+ }
+ ],
+ "SHOW BACKENDS": [
+ {
+ "BackendId": 10001,
+ "Host": "10.0.0.2",
+ "Alive": "true",
+ "Version": "doris-4.0.5-rc01",
+ }
+ ],
+ "SHOW BROKER": [],
+ }
+ )
+
+ result = await runtime.list_cluster_nodes(
+ node_types=["fe", "be"],
+ include_metrics=False,
+ )
+
+ assert result["status"] == "success"
+ assert result["data"]["items"] == [
+ {
+ "node_type": "be",
+ "node_id": "10001",
+ "host": "10.0.0.2",
+ "backend_id": 10001,
+ "alive": "true",
+ "version": "doris-4.0.5-rc01",
+ },
+ {
+ "node_type": "fe",
+ "node_id": "fe-1",
+ "host": "10.0.0.1",
+ "name": "fe-1",
+ "is_master": "true",
+ "version": "doris-4.0.5-rc01",
+ },
+ ]
+ assert manager.calls == ["SHOW FRONTENDS", "SHOW BACKENDS"]
+
+
[email protected]
+async def test_memory_stats_only_returns_observed_metrics() -> None:
+ runtime, _, _ = _runtime()
+
+ result = await runtime.get_memory_stats(
+ node_ids=["1"],
+ detail="top_consumers",
+ )
+
+ assert result["metadata"]["invented_values"] is False
+ metrics = result["data"]["nodes"][0]["metrics"]
+ assert metrics == {
+ "doris_be_memory_jemalloc_active_bytes": 2048,
+ "doris_be_memory_allocated_bytes": 1024,
+ }
+ assert "8 GB" not in str(result)
+ assert "4.5 GB" not in str(result)
+
+
[email protected]
+async def test_cache_status_prefers_supported_system_table() -> None:
+ statement = "SELECT * FROM information_schema.file_cache_statistics"
+ runtime, manager, monitoring = _runtime(
+ rows={
+ statement: [
+ {
+ "BE_ID": 1,
+ "BE_IP": "be-1",
+ "CACHE_PATH": "/cache",
+ "METRIC_NAME": "hits_ratio",
+ "METRIC_VALUE": 0.9,
+ },
+ {
+ "BE_ID": 1,
+ "BE_IP": "be-1",
+ "CACHE_PATH": "/cache",
+ "METRIC_NAME": "normal_queue_curr_size",
+ "METRIC_VALUE": 10,
+ },
+ ]
+ }
+ )
+
+ result = await runtime.get_cache_status(
+ scope=None,
+ node_ids=None,
+ include_queues=False,
+ )
+
+ assert result["data"]["mode"] == "system_table"
+ assert [item["metric_name"] for item in result["data"]["items"]] == [
+ "hits_ratio"
+ ]
+ assert manager.calls == [statement]
+ assert monitoring.calls == []
+
+
[email protected]
+async def test_compaction_falls_back_to_real_metrics_without_task_detail() ->
None:
+ native = "SELECT * FROM information_schema.doris_be_compaction_tasks"
+ runtime, _, _ = _runtime(
+ failures={native: RuntimeError(1146, "table does not exist")}
+ )
+
+ result = await runtime.get_compaction_status(
+ database="analytics",
+ table="orders",
+ state="running",
+ limit=10,
+ )
+
+ assert result["status"] == "partial"
+ assert result["data"]["mode"] == "legacy_summary"
+ assert result["data"]["native_tracker"] is False
+ assert "task-level detail" in result["warnings"][0]
+ be_node = next(
+ node for node in result["data"]["nodes"] if node["role"] == "be"
+ )
+ assert be_node["metrics"] == {
+ "doris_be_tablet_base_max_compaction_score": 4
+ }
+
+
[email protected]
+async def
test_resource_growth_uses_recorded_rows_and_reports_partial_sources() -> None:
+ query_sql = (
+ "SELECT DATE(`time`) AS bucket, COUNT(*) AS value "
+ "FROM internal.__internal_schema.audit_log "
+ "WHERE `time` >= DATE_SUB(NOW(), INTERVAL %s DAY) "
+ "GROUP BY bucket ORDER BY bucket"
+ )
+ runtime, manager, _ = _runtime(
+ rows={
+ query_sql: [
+ {"bucket": "2026-07-30", "value": 10},
+ {"bucket": "2026-07-31", "value": 15},
+ ]
+ }
+ )
+
+ result = await runtime.analyze_resource_growth(
+ resource="query_volume",
+ window_days=7,
+ granularity="day",
+ )
+
+ assert manager.calls == [query_sql]
+ assert manager.params == [(7,)]
+ assert result["metadata"]["invented_history"] is False
+ assert result["data"]["summaries"]["query_volume"] == {
+ "point_count": 2,
+ "first_value": 10.0,
+ "last_value": 15.0,
+ "growth_percent": 50.0,
+ }
+ assert result["evidence"][0]["source"] == (
+ "internal.__internal_schema.audit_log"
+ )
+
+
[email protected]
+async def test_resource_growth_rejects_unknown_resource_before_sql() -> None:
+ runtime, manager, _ = _runtime()
+
+ with pytest.raises(ClusterRuntimeFailure) as error:
+ await runtime.analyze_resource_growth(
+ resource="forecast",
+ window_days=7,
+ granularity="day",
+ )
+
+ assert error.value.reason_code == "CLUSTER_ARGUMENT_INVALID"
+ assert manager.calls == []
diff --git a/test/utils/test_doris_user_pool_manager.py
b/test/utils/test_doris_user_pool_manager.py
index 53827b7..f453892 100644
--- a/test/utils/test_doris_user_pool_manager.py
+++ b/test/utils/test_doris_user_pool_manager.py
@@ -455,7 +455,7 @@ async def
test_release_routed_connection_force_closes_unhealthy_raw(manager):
await manager.release_routed_connection(connection)
- assert owner.release_calls == []
+ assert owner.release_calls == [raw_connection]
assert raw_connection.ensure_closed_calls == 1
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
