This is an automated email from the ASF dual-hosted git repository. FreeOnePlus pushed a commit to branch feat/v1-09-cluster-domain in repository https://gitbox.apache.org/repos/asf/doris-mcp-server.git
commit 8011aef0f41faf12e8665a31ef9802d3a14a4049 Author: FreeOnePlus <[email protected]> AuthorDate: Fri Jul 31 16:58:32 2026 +0800 feat: add Cluster domain runtime --- 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]
