feat: add infra diagnostics tool to the Moose Dev MCP

refactor

Author: callicles

apps/framework-cli/src/mcp/server.rs

@@ -14,7 +14,9 @@ use rmcp::{
 use std::sync::Arc;
 
 use super::embedded_docs;
-use super::tools::{create_error_result, get_source, infra_map, logs, query_olap, sample_stream};
+use super::tools::{
+    create_error_result, get_source, infra_issues, infra_map, logs, query_olap, sample_stream,
+};
 use crate::cli::processing_coordinator::ProcessingCoordinator;
 use crate::infrastructure::olap::clickhouse::config::ClickHouseConfig;
 use crate::infrastructure::redis::redis_client::RedisClient;
@@ -84,6 +86,7 @@ impl ServerHandler for MooseMcpHandler {
             tools: vec![
                 logs::tool_definition(),
                 infra_map::tool_definition(),
+                infra_issues::tool_definition(),
                 query_olap::tool_definition(),
                 sample_stream::tool_definition(),
                 get_source::tool_definition(),
@@ -108,6 +111,12 @@ impl ServerHandler for MooseMcpHandler {
                 self.redis_client.clone(),
             )
             .await),
+            "diagnose_infrastructure" => Ok(infra_issues::handle_call(
+                param.arguments.as_ref(),
+                self.redis_client.clone(),
+                &self.clickhouse_config,
+            )
+            .await),
             "query_olap" => Ok(query_olap::handle_call(
                 &self.clickhouse_config,
                 param.arguments.as_ref(),

apps/framework-cli/src/mcp/tools/infra_issues/errors.rs

@@ -0,0 +1,124 @@
+//! Diagnostic provider for checking system-wide errors
+
+use log::debug;
+use serde_json::{json, Map, Value};
+
+use super::{Component, DiagnoseError, DiagnosticProvider, Issue, Severity};
+use crate::infrastructure::olap::clickhouse::client::ClickHouseClient;
+use crate::infrastructure::olap::clickhouse::config::ClickHouseConfig;
+use crate::infrastructure::olap::clickhouse::queries::ClickhouseEngine;
+
+/// Query timeout for diagnostic checks (30 seconds)
+const DIAGNOSTIC_QUERY_TIMEOUT_SECS: u64 = 30;
+
+/// Diagnostic provider for checking system-wide errors
+pub struct ErrorStatsDiagnostic;
+
+#[async_trait::async_trait]
+impl DiagnosticProvider for ErrorStatsDiagnostic {
+    fn name(&self) -> &str {
+        "ErrorStatsDiagnostic"
+    }
+
+    fn applicable_to(&self, _component: &Component, _engine: Option<&ClickhouseEngine>) -> bool {
+        // Error stats are system-wide, check only once (for first component)
+        true
+    }
+
+    async fn diagnose(
+        &self,
+        component: &Component,
+        _engine: Option<&ClickhouseEngine>,
+        config: &ClickHouseConfig,
+        _since: Option<&str>,
+    ) -> Result<Vec<Issue>, DiagnoseError> {
+        let client = ClickHouseClient::new(config)
+            .map_err(|e| DiagnoseError::ClickHouseConnection(format!("{}", e)))?;
+
+        // Get recent errors with significant counts
+        let query = "SELECT
+                name,
+                value,
+                last_error_time,
+                last_error_message
+             FROM system.errors
+             WHERE value > 0
+             ORDER BY value DESC
+             LIMIT 10
+             FORMAT JSON";
+
+        debug!("Executing errors query: {}", query);
+
+        let result = tokio::time::timeout(
+            std::time::Duration::from_secs(DIAGNOSTIC_QUERY_TIMEOUT_SECS),
+            client.execute_sql(query),
+        )
+        .await
+        .map_err(|_| DiagnoseError::QueryTimeout(DIAGNOSTIC_QUERY_TIMEOUT_SECS))?
+        .map_err(|e| DiagnoseError::QueryFailed(format!("{}", e)))?;
+
+        let json_response: Value = serde_json::from_str(&result)
+            .map_err(|e| DiagnoseError::ParseError(format!("{}", e)))?;
+
+        let data = json_response
+            .get("data")
+            .and_then(|v| v.as_array())
+            .ok_or_else(|| {
+                DiagnoseError::ParseError("Missing 'data' field in response".to_string())
+            })?;
+
+        let mut issues = Vec::new();
+
+        for row in data {
+            let name = row
+                .get("name")
+                .and_then(|v| v.as_str())
+                .unwrap_or("UNKNOWN");
+            let value = row.get("value").and_then(|v| v.as_u64()).unwrap_or(0);
+            let last_error_message = row
+                .get("last_error_message")
+                .and_then(|v| v.as_str())
+                .unwrap_or("");
+
+            // Skip if no occurrences
+            if value == 0 {
+                continue;
+            }
+
+            let severity = if value > 100 {
+                Severity::Error
+            } else if value > 10 {
+                Severity::Warning
+            } else {
+                Severity::Info
+            };
+
+            let mut details = Map::new();
+            details.insert("error_name".to_string(), json!(name));
+            details.insert("occurrence_count".to_string(), json!(value));
+            details.insert(
+                "last_error_time".to_string(),
+                row.get("last_error_time").cloned().unwrap_or(json!("")),
+            );
+            if !last_error_message.is_empty() {
+                details.insert("last_error_message".to_string(), json!(last_error_message));
+            }
+
+            issues.push(Issue {
+                severity,
+                source: "system.errors".to_string(),
+                component: component.clone(),
+                error_type: "system_error".to_string(),
+                message: format!("Error '{}' occurred {} times. Last: {}", name, value, last_error_message),
+                details,
+                suggested_action: "Review error pattern and recent query logs. Check ClickHouse server logs for more details.".to_string(),
+                related_queries: vec![
+                    "SELECT * FROM system.errors WHERE value > 0 ORDER BY value DESC".to_string(),
+                    format!("SELECT * FROM system.query_log WHERE exception LIKE '%{}%' ORDER BY event_time DESC LIMIT 10", name),
+                ],
+            });
+        }
+
+        Ok(issues)
+    }
+}

apps/framework-cli/src/mcp/tools/infra_issues/merge_failures.rs

@@ -0,0 +1,100 @@
+//! Diagnostic provider for checking merge failures from system.metrics
+
+use log::debug;
+use serde_json::{json, Map, Value};
+
+use super::{Component, DiagnoseError, DiagnosticProvider, Issue, Severity};
+use crate::infrastructure::olap::clickhouse::client::ClickHouseClient;
+use crate::infrastructure::olap::clickhouse::config::ClickHouseConfig;
+use crate::infrastructure::olap::clickhouse::queries::ClickhouseEngine;
+
+/// Query timeout for diagnostic checks (30 seconds)
+const DIAGNOSTIC_QUERY_TIMEOUT_SECS: u64 = 30;
+
+/// Diagnostic provider for checking merge failures from system.metrics
+pub struct MergeFailureDiagnostic;
+
+#[async_trait::async_trait]
+impl DiagnosticProvider for MergeFailureDiagnostic {
+    fn name(&self) -> &str {
+        "merge_failures"
+    }
+
+    fn applicable_to(&self, _component: &Component, _engine: Option<&ClickhouseEngine>) -> bool {
+        // Merge failures are system-wide but we report per-table for context
+        true
+    }
+
+    async fn diagnose(
+        &self,
+        component: &Component,
+        _engine: Option<&ClickhouseEngine>,
+        config: &ClickHouseConfig,
+        _since: Option<&str>,
+    ) -> Result<Vec<Issue>, DiagnoseError> {
+        let client = ClickHouseClient::new(config)
+            .map_err(|e| DiagnoseError::ClickHouseConnection(format!("{}", e)))?;
+
+        let mut issues = Vec::new();
+
+        // Check system.metrics for background merge failures
+        // Note: This is a system-wide metric, not per-table, but we report it per-table for context
+        let metrics_query =
+            "SELECT value FROM system.metrics WHERE metric = 'FailedBackgroundMerges' FORMAT JSON";
+
+        debug!("Executing merge failure metrics query: {}", metrics_query);
+
+        let result = tokio::time::timeout(
+            std::time::Duration::from_secs(DIAGNOSTIC_QUERY_TIMEOUT_SECS),
+            client.execute_sql(metrics_query),
+        )
+        .await
+        .map_err(|_| DiagnoseError::QueryTimeout(DIAGNOSTIC_QUERY_TIMEOUT_SECS))?
+        .map_err(|e| DiagnoseError::QueryFailed(format!("{}", e)))?;
+
+        let json_response: Value = serde_json::from_str(&result)
+            .map_err(|e| DiagnoseError::ParseError(format!("{}", e)))?;
+
+        let failed_merges = json_response
+            .get("data")
+            .and_then(|v| v.as_array())
+            .and_then(|arr| arr.first())
+            .and_then(|row| row.get("value"))
+            .and_then(|v| v.as_u64())
+            .unwrap_or(0);
+
+        if failed_merges > 0 {
+            let severity = if failed_merges > 10 {
+                Severity::Error
+            } else {
+                Severity::Warning
+            };
+
+            let mut details = Map::new();
+            details.insert("failed_merges".to_string(), json!(failed_merges));
+
+            issues.push(Issue {
+                severity,
+                source: "system.metrics".to_string(),
+                component: component.clone(),
+                error_type: "merge_failures".to_string(),
+                message: format!(
+                    "Background merge failures detected: {} failed merges currently in system. This may affect table maintenance.",
+                    failed_merges
+                ),
+                details,
+                suggested_action: "Check system.errors for merge failure details. Review disk space and memory availability. Consider increasing merge-related settings if failures persist.".to_string(),
+                related_queries: vec![
+                    "SELECT * FROM system.errors WHERE name LIKE '%Merge%' ORDER BY last_error_time DESC LIMIT 10".to_string(),
+                    "SELECT * FROM system.metrics WHERE metric LIKE '%Merge%'".to_string(),
+                    format!(
+                        "SELECT * FROM system.merges WHERE database = '{}' AND table = '{}'",
+                        config.db_name, component.name
+                    ),
+                ],
+            });
+        }
+
+        Ok(issues)
+    }
+}

apps/framework-cli/src/mcp/tools/infra_issues/merges.rs

@@ -0,0 +1,121 @@
+//! Diagnostic provider for checking stuck background merges
+
+use log::debug;
+use serde_json::{json, Map, Value};
+
+use super::{Component, DiagnoseError, DiagnosticProvider, Issue, Severity};
+use crate::infrastructure::olap::clickhouse::client::ClickHouseClient;
+use crate::infrastructure::olap::clickhouse::config::ClickHouseConfig;
+use crate::infrastructure::olap::clickhouse::queries::ClickhouseEngine;
+
+/// Query timeout for diagnostic checks (30 seconds)
+const DIAGNOSTIC_QUERY_TIMEOUT_SECS: u64 = 30;
+
+/// Diagnostic provider for checking stuck background merges
+pub struct MergeDiagnostic;
+
+#[async_trait::async_trait]
+impl DiagnosticProvider for MergeDiagnostic {
+    fn name(&self) -> &str {
+        "MergeDiagnostic"
+    }
+
+    fn applicable_to(&self, _component: &Component, _engine: Option<&ClickhouseEngine>) -> bool {
+        true
+    }
+
+    async fn diagnose(
+        &self,
+        component: &Component,
+        _engine: Option<&ClickhouseEngine>,
+        config: &ClickHouseConfig,
+        _since: Option<&str>,
+    ) -> Result<Vec<Issue>, DiagnoseError> {
+        let client = ClickHouseClient::new(config)
+            .map_err(|e| DiagnoseError::ClickHouseConnection(format!("{}", e)))?;
+
+        // Check for long-running merges
+        let query = format!(
+            "SELECT
+                elapsed,
+                progress,
+                num_parts,
+                result_part_name,
+                total_size_bytes_compressed
+             FROM system.merges
+             WHERE database = '{}' AND table = '{}'
+             AND elapsed > 300
+             ORDER BY elapsed DESC
+             FORMAT JSON",
+            config.db_name, component.name
+        );
+
+        debug!("Executing merges query: {}", query);
+
+        let result = tokio::time::timeout(
+            std::time::Duration::from_secs(DIAGNOSTIC_QUERY_TIMEOUT_SECS),
+            client.execute_sql(&query),
+        )
+        .await
+        .map_err(|_| DiagnoseError::QueryTimeout(DIAGNOSTIC_QUERY_TIMEOUT_SECS))?
+        .map_err(|e| DiagnoseError::QueryFailed(format!("{}", e)))?;
+
+        let json_response: Value = serde_json::from_str(&result)
+            .map_err(|e| DiagnoseError::ParseError(format!("{}", e)))?;
+
+        let data = json_response
+            .get("data")
+            .and_then(|v| v.as_array())
+            .ok_or_else(|| {
+                DiagnoseError::ParseError("Missing 'data' field in response".to_string())
+            })?;
+
+        let mut issues = Vec::new();
+
+        for row in data {
+            let elapsed = row.get("elapsed").and_then(|v| v.as_f64()).unwrap_or(0.0);
+            let progress = row.get("progress").and_then(|v| v.as_f64()).unwrap_or(0.0);
+
+            let severity = if elapsed > 1800.0 {
+                // 30 minutes
+                Severity::Error
+            } else {
+                Severity::Warning
+            };
+
+            let mut details = Map::new();
+            details.insert("elapsed_seconds".to_string(), json!(elapsed));
+            details.insert("progress".to_string(), json!(progress));
+            details.insert(
+                "num_parts".to_string(),
+                row.get("num_parts").cloned().unwrap_or(json!(0)),
+            );
+            details.insert(
+                "result_part_name".to_string(),
+                row.get("result_part_name").cloned().unwrap_or(json!("")),
+            );
+
+            issues.push(Issue {
+                severity,
+                source: "system.merges".to_string(),
+                component: component.clone(),
+                error_type: "slow_merge".to_string(),
+                message: format!(
+                    "Background merge running for {:.1} seconds ({:.1}% complete)",
+                    elapsed, progress * 100.0
+                ),
+                details,
+                suggested_action: "Monitor merge progress. If stuck, check server resources (CPU, disk I/O, memory). Consider stopping merge with SYSTEM STOP MERGES if necessary.".to_string(),
+                related_queries: vec![
+                    format!(
+                        "SELECT * FROM system.merges WHERE database = '{}' AND table = '{}'",
+                        config.db_name, component.name
+                    ),
+                    format!("SYSTEM STOP MERGES {}.{}", config.db_name, component.name),
+                ],
+            });
+        }
+
+        Ok(issues)
+    }
+}