diff --git a/CHANGELOG.md b/CHANGELOG.md index e58126b..58ccee2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -65,6 +65,9 @@ under **Unreleased** until a new version is selected and published. ### Fixed +- Required at least two recorded time buckets before exposing or executing + resource-growth analysis, so readable-but-empty audit and partition metadata + no longer masquerade as historical evidence. - Aligned `doris_cluster.list_active_tasks` capability detection with its read-only execution fallbacks, so restricted Doris accounts can use the `information_schema.active_queries` or process-list source when diff --git a/doris_mcp_server/tools/capability_detector.py b/doris_mcp_server/tools/capability_detector.py index e9342f4..6d115c4 100644 --- a/doris_mcp_server/tools/capability_detector.py +++ b/doris_mcp_server/tools/capability_detector.py @@ -244,17 +244,6 @@ class CapabilityRouteChangedError(CapabilityDetectionError): "SHOW COMPUTE GROUPS", ("compute_group_metadata_readable",), ), - ( - ("SELECT `time` FROM internal.__internal_schema.audit_log LIMIT 1"), - ("metrics_history_readable",), - ), - ( - ( - "SELECT CREATE_TIME, DATA_LENGTH, INDEX_LENGTH " - "FROM information_schema.partitions LIMIT 1" - ), - ("resource_storage_history_readable",), - ), ), "doris_pipeline": ( ( @@ -516,6 +505,9 @@ async def detect_domain( probes ) elif domain_name == "doris_cluster": + probes.update( + await self._probe_cluster_history_sources(auth_context) + ) probes.update(await self._safe_probe_cluster_services(auth_context)) probes.update(_combine_cluster_evidence_probes(probes)) elif domain_name == "doris_pipeline": @@ -753,6 +745,76 @@ async def _probe_query_services( ) return probes + async def _probe_cluster_history_sources( + self, + auth_context: Any | None, + ) -> dict[str, CapabilityProbeEvidence]: + """Require recorded history, not merely readable metadata objects.""" + contracts = ( + ( + "metrics_history_readable", + "SELECT COUNT(DISTINCT DATE(`time`)) AS evidence_bucket_count " + "FROM internal.__internal_schema.audit_log " + "WHERE `time` >= DATE_SUB(NOW(), INTERVAL 3650 DAY)", + "AUDIT_HISTORY_RECORDED", + "AUDIT_HISTORY_INSUFFICIENT", + ), + ( + "resource_storage_history_readable", + "SELECT COUNT(DISTINCT DATE(CREATE_TIME)) " + "AS evidence_bucket_count " + "FROM information_schema.partitions " + "WHERE CREATE_TIME >= DATE_SUB(NOW(), INTERVAL 3650 DAY)", + "PARTITION_CREATION_HISTORY_RECORDED", + "PARTITION_CREATION_HISTORY_INSUFFICIENT", + ), + ) + evidence: dict[str, CapabilityProbeEvidence] = {} + route = self.route_identity(auth_context) + for index, ( + probe_id, + statement, + supported_reason, + insufficient_reason, + ) in enumerate(contracts): + session_id = ( + f"capability-cluster:history:{index}:" + f"{route.fingerprint[:12]}" + ) + async with ( + self._connection_manager.get_connection_context_for_auth_context( + session_id, + auth_context, + ) as connection + ): + rows, probe = await self._probe_rows( + connection, + statement, + probe_id, + ) + if probe.status is CapabilityProbeStatus.SUPPORTED: + bucket_count = _nonnegative_int( + _row_value(rows[0], "evidence_bucket_count") + if rows + else None + ) + probe = CapabilityProbeEvidence( + probe_id=probe_id, + status=( + CapabilityProbeStatus.SUPPORTED + if bucket_count >= 2 + else CapabilityProbeStatus.UNKNOWN + ), + reason_code=( + supported_reason + if bucket_count >= 2 + else insufficient_reason + ), + evidence_sources=("recorded_history_probe",), + ) + evidence[probe_id] = probe + return evidence + async def _probe_search_match_syntax( self, auth_context: Any | None, @@ -1886,6 +1948,16 @@ def _combine_cluster_evidence_probes( evidence_sources=("runtime_probe",), ) ) + if audit is not None and storage is not None and not any( + source.status is CapabilityProbeStatus.SUPPORTED + for source in (audit, storage) + ): + full = CapabilityProbeEvidence( + probe_id=full.probe_id, + status=full.status, + reason_code="RESOURCE_HISTORY_UNAVAILABLE", + evidence_sources=full.evidence_sources, + ) def partial_source( probe_id: str, @@ -2349,6 +2421,21 @@ def _row_value( return None +def _nonnegative_int(value: Any | None) -> int: + if isinstance(value, bool): + return 0 + if isinstance(value, int): + parsed = value + elif isinstance(value, str): + try: + parsed = int(value.strip()) + except ValueError: + return 0 + else: + return 0 + return max(0, parsed) + + def _component_version(value: Any) -> DorisVersion: raw = "" if value is None else str(value).strip() if not raw: diff --git a/doris_mcp_server/tools/domain_catalog.py b/doris_mcp_server/tools/domain_catalog.py index 56267b4..df024b5 100644 --- a/doris_mcp_server/tools/domain_catalog.py +++ b/doris_mcp_server/tools/domain_catalog.py @@ -1513,7 +1513,8 @@ def _domain( "analyze_resource_growth", "Analyze resource growth", "Analyze recorded resource-growth evidence without inventing " - "missing history. For an unqualified cluster-history request, " + "missing history. A series requires at least two recorded time " + "buckets. For an unqualified cluster-history request, " "omit resource so every currently usable recorded series is " "attempted and partial evidence is preserved.", _input_schema( diff --git a/doris_mcp_server/utils/cluster_runtime.py b/doris_mcp_server/utils/cluster_runtime.py index ce88fa2..28b54c5 100644 --- a/doris_mcp_server/utils/cluster_runtime.py +++ b/doris_mcp_server/utils/cluster_runtime.py @@ -749,7 +749,22 @@ async def analyze_resource_growth( "value": _number(_row_lookup(row, "value")), } for row in rows + if _row_lookup(row, "bucket") is not None ] + if len(points) < 2: + warnings.append( + f"{resource_name} history has fewer than two recorded " + "time buckets." + ) + evidence.append( + { + "resource": resource_name, + "success": False, + "reason_code": "RESOURCE_HISTORY_INSUFFICIENT", + "points": len(points), + } + ) + continue series[resource_name] = points evidence.append( { diff --git a/test/tools/test_capability_detector.py b/test/tools/test_capability_detector.py index 187a1ec..8066cd6 100644 --- a/test/tools/test_capability_detector.py +++ b/test/tools/test_capability_detector.py @@ -44,6 +44,16 @@ "AND TABLE_SCHEMA NOT IN ('information_schema', 'mysql') " "ORDER BY TABLE_SCHEMA, TABLE_NAME, ORDINAL_POSITION LIMIT 8" ) +_AUDIT_HISTORY_PROBE_SQL = ( + "SELECT COUNT(DISTINCT DATE(`time`)) AS evidence_bucket_count " + "FROM internal.__internal_schema.audit_log " + "WHERE `time` >= DATE_SUB(NOW(), INTERVAL 3650 DAY)" +) +_STORAGE_HISTORY_PROBE_SQL = ( + "SELECT COUNT(DISTINCT DATE(CREATE_TIME)) AS evidence_bucket_count " + "FROM information_schema.partitions " + "WHERE CREATE_TIME >= DATE_SUB(NOW(), INTERVAL 3650 DAY)" +) class _ProbeConnection: @@ -113,6 +123,8 @@ async def execute( "DATA_LENGTH": 8, } ], + _AUDIT_HISTORY_PROBE_SQL: [{"evidence_bucket_count": 3}], + _STORAGE_HISTORY_PROBE_SQL: [{"evidence_bucket_count": 2}], } return SimpleNamespace(data=self.row_overrides.get(sql, rows.get(sql, []))) @@ -222,10 +234,7 @@ async def test_detector_builds_version_vector_and_extends_domains_lazily() -> No @pytest.mark.asyncio async def test_cluster_history_keeps_storage_fallback_without_audit_access() -> None: connection = _ProbeConnection() - audit_probe = ( - "SELECT `time` FROM internal.__internal_schema.audit_log LIMIT 1" - ) - connection.failures[audit_probe] = RuntimeError( + connection.failures[_AUDIT_HISTORY_PROBE_SQL] = RuntimeError( "Access denied; user lacks SELECT privilege" ) manager = _ProbeConnectionManager(connection) @@ -251,6 +260,43 @@ async def test_cluster_history_keeps_storage_fallback_without_audit_access() -> assert storage.reason_code == "PARTITION_CREATION_HISTORY_ONLY" +@pytest.mark.asyncio +async def test_cluster_history_requires_two_recorded_time_buckets() -> None: + connection = _ProbeConnection() + connection.row_overrides[_AUDIT_HISTORY_PROBE_SQL] = [ + {"evidence_bucket_count": 1} + ] + connection.row_overrides[_STORAGE_HISTORY_PROBE_SQL] = [ + {"evidence_bucket_count": 0} + ] + manager = _ProbeConnectionManager(connection) + detector = DorisCapabilityDetector(manager) # type: ignore[arg-type] + base = await detector.detect_base( + None, + capability_generation=1, + provider_generation="provider.cluster", + ) + + cluster = await detector.detect_domain(base, "doris_cluster", None) + + audit = cluster.probe("metrics_history_readable") + storage = cluster.probe("resource_storage_history_readable") + assert audit is not None + assert audit.status is CapabilityProbeStatus.UNKNOWN + assert audit.reason_code == "AUDIT_HISTORY_INSUFFICIENT" + assert storage is not None + assert storage.status is CapabilityProbeStatus.UNKNOWN + assert storage.reason_code == "PARTITION_CREATION_HISTORY_INSUFFICIENT" + assert ( + cluster.probe("resource_history_all_sources_readable").status + is CapabilityProbeStatus.UNKNOWN + ) + assert ( + cluster.probe("resource_history_all_sources_readable").reason_code + == "RESOURCE_HISTORY_UNAVAILABLE" + ) + + @pytest.mark.asyncio async def test_cluster_active_tasks_accepts_read_only_query_view_fallback() -> None: connection = _ProbeConnection() diff --git a/test/utils/test_cluster_runtime.py b/test/utils/test_cluster_runtime.py index b6b8e81..9e7c613 100644 --- a/test/utils/test_cluster_runtime.py +++ b/test/utils/test_cluster_runtime.py @@ -346,3 +346,33 @@ async def test_resource_growth_rejects_unknown_resource_before_sql() -> None: assert error.value.reason_code == "CLUSTER_ARGUMENT_INVALID" assert manager.calls == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "rows", + [ + [], + [{"bucket": "2026-07-31", "value": 15}], + ], +) +async def test_resource_growth_rejects_insufficient_recorded_history( + rows: list[dict[str, Any]], +) -> 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: rows}) + + with pytest.raises(ClusterRuntimeFailure) as error: + await runtime.analyze_resource_growth( + resource="query_volume", + window_days=30, + granularity="day", + ) + + assert error.value.reason_code == "RESOURCE_HISTORY_UNAVAILABLE" + assert manager.calls == [query_sql]