This is an automated email from the ASF dual-hosted git repository.
FreeOnePlus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris-mcp-server.git
The following commit(s) were added to refs/heads/master by this push:
new 97ebfac fix(cluster): require recorded history evidence (#212)
97ebfac is described below
commit 97ebfacca0b23058e7bbb35420823a767a768f12
Author: Yijia Su <[email protected]>
AuthorDate: Thu Aug 13 19:36:59 2026 +0800
fix(cluster): require recorded history evidence (#212)
---
CHANGELOG.md | 3 +
doris_mcp_server/tools/capability_detector.py | 109 +++++++++++++++++++++++---
doris_mcp_server/tools/domain_catalog.py | 3 +-
doris_mcp_server/utils/cluster_runtime.py | 15 ++++
test/tools/test_capability_detector.py | 54 ++++++++++++-
test/utils/test_cluster_runtime.py | 30 +++++++
6 files changed, 198 insertions(+), 16 deletions(-)
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 @@ _DOMAIN_PROBES: Mapping[str, tuple[tuple[str, tuple[str,
...]], ...]] = {
"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 @@ class DorisCapabilityDetector:
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 @@ class DorisCapabilityDetector:
)
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 @@ DOMAIN_DEFINITIONS = (
"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 @@ class DorisClusterRuntime:
"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 @@ _SEARCH_TARGET_DISCOVERY_SQL = (
"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 @@ class _ProbeConnection:
"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"
[email protected]
+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 == []
+
+
[email protected]
[email protected](
+ "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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
