From b85e03921c48c68467503eaf1536d5e789cb4483 Mon Sep 17 00:00:00 2001 From: Nikhil Sinha Date: Tue, 13 Oct 2026 14:09:38 +0530 Subject: [PATCH 1/3] add log, trace errors as tools --- src/tool_catalog.rs | 70 ++++++++++++++++- src/tool_executor.rs | 175 ++++++++++++++++++++++++++++++++++++++++++- 2 files changed, 238 insertions(+), 7 deletions(-) diff --git a/src/tool_catalog.rs b/src/tool_catalog.rs index 196893fb5..c75f5eca3 100644 --- a/src/tool_catalog.rs +++ b/src/tool_catalog.rs @@ -104,7 +104,9 @@ impl ToolSpec { | "explain_query" | "discover_datasets" | "get_traces" - | "get_trace" => ToolEffect::ExpensiveRead, + | "get_trace" + | "list_log_errors" + | "list_trace_errors" => ToolEffect::ExpensiveRead, "create_alert" | "enable_alert" | "disable_alert" @@ -402,6 +404,34 @@ pub fn oss_read_tool_specs() -> Vec { &["dataset", "pTimestamp"], ), ), + ToolSpec::new( + "list_log_errors", + "List log errors", + "List grouped error, fatal, and exception log records from one logs dataset in a bounded time range.", + object_schema( + json!({ + "dataset": string_property("One logs dataset."), + "startTime": string_property("RFC3339 range start."), + "endTime": string_property("RFC3339 range end."), + "limit": { "type": "integer", "minimum": 1, "maximum": 500, "default": 50 } + }), + &["dataset", "startTime", "endTime"], + ), + ), + ToolSpec::new( + "list_trace_errors", + "List trace errors", + "List failed and error spans from one traces dataset in a bounded time range.", + object_schema( + json!({ + "dataset": string_property("One traces dataset."), + "startTime": string_property("RFC3339 range start."), + "endTime": string_property("RFC3339 range end."), + "limit": { "type": "integer", "minimum": 1, "maximum": 500, "default": 50 } + }), + &["dataset", "startTime", "endTime"], + ), + ), ToolSpec::new( "query_sql", "Run SQL query", @@ -824,7 +854,7 @@ mod tests { let tools = oss_tool_specs(); let names = tools.iter().map(|tool| tool.name).collect::>(); - assert_eq!(tools.len(), 47); + assert_eq!(tools.len(), 49); assert_eq!(tools.len(), names.len()); } @@ -863,7 +893,7 @@ mod tests { let read_tools = oss_read_tool_specs(); let action_tools = oss_action_tool_specs(); - assert_eq!(read_tools.len(), 34); + assert_eq!(read_tools.len(), 36); assert_eq!(action_tools.len(), 13); assert!(read_tools.iter().all(|tool| matches!( tool.effect(), @@ -913,6 +943,40 @@ mod tests { ); } + #[actix_web::test] + async fn oss_catalog_exposes_list_log_errors() { + let tool = oss_read_tool_specs() + .into_iter() + .find(|tool| tool.name == "list_log_errors") + .expect("list_log_errors must be available in OSS"); + + assert_eq!(tool.effect(), ToolEffect::ExpensiveRead); + assert_eq!(tool.input_schema["properties"]["dataset"]["type"], "string"); + assert!( + tool.input_schema["required"] + .as_array() + .unwrap() + .contains(&serde_json::json!("dataset")) + ); + } + + #[actix_web::test] + async fn oss_catalog_exposes_list_trace_errors() { + let tool = oss_read_tool_specs() + .into_iter() + .find(|tool| tool.name == "list_trace_errors") + .expect("list_trace_errors must be available in OSS"); + + assert_eq!(tool.effect(), ToolEffect::ExpensiveRead); + assert_eq!(tool.input_schema["properties"]["dataset"]["type"], "string"); + assert!( + tool.input_schema["required"] + .as_array() + .unwrap() + .contains(&serde_json::json!("dataset")) + ); + } + #[actix_web::test] async fn tool_call_defaults_to_empty_arguments() { let call: ToolCallRequest = serde_json::from_value(serde_json::json!({ diff --git a/src/tool_executor.rs b/src/tool_executor.rs index a68bf88a5..f2fcd48af 100644 --- a/src/tool_executor.rs +++ b/src/tool_executor.rs @@ -409,6 +409,148 @@ async fn get_log_context(arguments: &Value, context: OssToolContext<'_>) -> Resu response_value(response).await } +fn first_present_field(fields: &[String], candidates: &[&str]) -> Option { + candidates + .iter() + .find(|candidate| fields.iter().any(|field| field == **candidate)) + .map(|field| (*field).to_owned()) +} + +pub fn list_log_errors_query( + dataset: &str, + fields: &[String], + limit: u64, +) -> Result { + let service = first_present_field(fields, &["service.name", "service_name"]) + .map(|field| { + format!( + "COALESCE(CAST({} AS VARCHAR), 'unknown')", + quote_identifier(&field) + ) + }) + .unwrap_or_else(|| "'unknown'".to_owned()); + let severity_text = first_present_field(fields, &["severity_text", "severity_description"]); + let exception_type = first_present_field( + fields, + &["event_exception.type", "exception.type", "error.type"], + ); + let error_type = exception_type + .as_ref() + .or(severity_text.as_ref()) + .map(|field| { + format!( + "COALESCE(CAST({} AS VARCHAR), 'error')", + quote_identifier(field) + ) + }) + .unwrap_or_else(|| "'error'".to_owned()); + let message = first_present_field( + fields, + &[ + "event_exception.message", + "exception.message", + "error.message", + "body", + "message", + "log", + ], + ); + let sample_message = message + .as_ref() + .map(|field| format!("MAX(CAST({} AS VARCHAR))", quote_identifier(field))) + .unwrap_or_else(|| "CAST(NULL AS VARCHAR)".to_owned()); + let last_seen = first_present_field(fields, &["p_timestamp", "time_unix_nano"]) + .map(|field| format!("MAX({})", quote_identifier(&field))) + .unwrap_or_else(|| "CAST(NULL AS VARCHAR)".to_owned()); + + let mut predicates = Vec::new(); + if fields.iter().any(|field| field == "severity_number") { + predicates.push("CAST(\"severity_number\" AS BIGINT) >= 17".to_owned()); + } + if let Some(field) = &severity_text { + predicates.push(format!( + "UPPER(CAST({} AS VARCHAR)) IN ('ERROR', 'FATAL', 'CRITICAL')", + quote_identifier(field) + )); + } + if let Some(field) = &exception_type { + predicates.push(format!( + "{} IS NOT NULL AND CAST({} AS VARCHAR) <> ''", + quote_identifier(field), + quote_identifier(field) + )); + } + if predicates.is_empty() + && let Some(field) = &message + { + let field = quote_identifier(field); + predicates.push(format!( + "(LOWER(CAST({field} AS VARCHAR)) LIKE '%error%' OR LOWER(CAST({field} AS VARCHAR)) LIKE '%exception%' OR LOWER(CAST({field} AS VARCHAR)) LIKE '%failed%' OR LOWER(CAST({field} AS VARCHAR)) LIKE '%fatal%' OR LOWER(CAST({field} AS VARCHAR)) LIKE '%panic%')" + )); + } + if predicates.is_empty() { + return Err( + "log dataset has no supported severity, exception, or message field".to_owned(), + ); + } + + Ok(format!( + "SELECT {service} AS \"serviceName\", {error_type} AS \"errorType\", {sample_message} AS \"sampleMessage\", COUNT(*) AS \"occurrences\", {last_seen} AS \"lastSeen\" FROM {} WHERE ({}) GROUP BY {service}, {error_type} ORDER BY \"occurrences\" DESC LIMIT {}", + quote_identifier(dataset), + predicates.join(" OR "), + limit.clamp(1, 500), + )) +} + +async fn list_log_errors(arguments: &Value, context: OssToolContext<'_>) -> Result { + let dataset = required_string(arguments, "dataset")?; + authorize(context.session_key, Action::Query, Some(dataset), None)?; + let info = serde_json::to_value( + get_prism_logstream_info(dataset, context.tenant_id) + .await + .map_err(|error| error.to_string())?, + ) + .map_err(|error| error.to_string())?; + if !nested_string(&info, "telemetryType").is_some_and(|value| value == "logs") { + return Err(format!( + "tool requires a logs dataset, but `{dataset}` is not logs telemetry" + )); + } + let schema = logstream::get_schema_internal(dataset, context.tenant_id) + .await + .map_err(|error| error.to_string())?; + let fields = schema + .fields() + .iter() + .map(|field| field.name().to_owned()) + .collect::>(); + let limit = bounded_integer(arguments, "limit", 50, 1, 500); + let query_text = bounded_select(&list_log_errors_query(dataset, &fields, limit)?)?; + let response = query::query_internal( + query::Query { + query: query_text, + start_time: required_string(arguments, "startTime")?.to_owned(), + end_time: required_string(arguments, "endTime")?.to_owned(), + send_null: true, + fields: false, + streaming: false, + filter_tags: None, + }, + context.session_key, + context.tenant_id, + ) + .await + .map_err(|error| error.to_string())?; + let records = response_value(response).await?; + let total = records.as_array().map_or(0, Vec::len); + Ok(json!({ + "total": total, + "offset": 0, + "limit": limit, + "records": records, + })) +} + async fn execute_sql_tool( name: &str, arguments: &Value, @@ -563,7 +705,8 @@ async fn execute_trace_tool( let datasets = [required_string(arguments, "dataset")?.to_owned()]; let start_time = required_string(arguments, "startTime")?.to_owned(); let end_time = required_string(arguments, "endTime")?.to_owned(); - let limit = bounded_integer(arguments, "limit", 100, 1, 500) as usize; + let default_limit = if name == "list_trace_errors" { 50 } else { 100 }; + let limit = bounded_integer(arguments, "limit", default_limit, 1, 500) as usize; let mut results = Vec::new(); let mut errors = Vec::new(); for dataset in datasets { @@ -596,7 +739,8 @@ async fn execute_trace_tool( end_time: end_time.clone(), sort_by: None, conditions: None, - options: None, + options: (name == "list_trace_errors") + .then_some(crate::handlers::http::traces::TraceListOption::ErrorSpans), limit: Some(limit), offset: Some(0), }, @@ -818,13 +962,16 @@ async fn execute_oss_tool_inner( } "sample_events" => sample_events(arguments, context).await, "get_log_context" => get_log_context(arguments, context).await, + "list_log_errors" => list_log_errors(arguments, context).await, "query_sql" | "explain_query" => execute_sql_tool(name, arguments, context).await, "list_alerts" | "get_alert" | "list_alert_tags" | "enable_alert" | "disable_alert" | "evaluate_alert" | "create_alert" => execute_alert_tool(name, arguments, context).await, "list_alert_targets" | "get_alert_target" | "create_alert_target" => { execute_alert_target_tool(name, arguments, context).await } - "get_traces" | "get_trace" => execute_trace_tool(name, arguments, context).await, + "get_traces" | "get_trace" | "list_trace_errors" => { + execute_trace_tool(name, arguments, context).await + } "ping" | "get_cluster_status" | "get_cluster_metrics" @@ -894,7 +1041,27 @@ mod tests { use chrono::{DateTime, Duration, Utc}; use serde_json::{Value, json}; - use super::{call_tool_registry, normalized_log_context_bounds}; + use super::{call_tool_registry, list_log_errors_query, normalized_log_context_bounds}; + + #[actix_web::test] + async fn log_error_query_uses_otel_severity_fields() { + let query = list_log_errors_query( + "app-logs", + &[ + "p_timestamp".to_owned(), + "service.name".to_owned(), + "severity_number".to_owned(), + "severity_text".to_owned(), + "body".to_owned(), + ], + 5, + ) + .unwrap(); + + assert!(query.contains("CAST(\"severity_number\" AS BIGINT) >= 17")); + assert!(query.contains("'ERROR', 'FATAL', 'CRITICAL'")); + assert!(query.contains("LIMIT 5")); + } #[actix_web::test] async fn log_context_bounds_normalize_relative_times() { From e0151dd994fdaf1f4af1f76d8b3bed651dba669d Mon Sep 17 00:00:00 2001 From: Nikhil Sinha Date: Tue, 13 Oct 2026 14:14:49 +0530 Subject: [PATCH 2/3] deesource error fix --- src/tool_executor.rs | 40 ++++++++++++++++++++++------------------ 1 file changed, 22 insertions(+), 18 deletions(-) diff --git a/src/tool_executor.rs b/src/tool_executor.rs index f2fcd48af..c832a5846 100644 --- a/src/tool_executor.rs +++ b/src/tool_executor.rs @@ -421,14 +421,15 @@ pub fn list_log_errors_query( fields: &[String], limit: u64, ) -> Result { - let service = first_present_field(fields, &["service.name", "service_name"]) - .map(|field| { + let service = first_present_field(fields, &["service.name", "service_name"]).map_or_else( + || "'unknown'".to_owned(), + |field| { format!( "COALESCE(CAST({} AS VARCHAR), 'unknown')", quote_identifier(&field) ) - }) - .unwrap_or_else(|| "'unknown'".to_owned()); + }, + ); let severity_text = first_present_field(fields, &["severity_text", "severity_description"]); let exception_type = first_present_field( fields, @@ -437,13 +438,15 @@ pub fn list_log_errors_query( let error_type = exception_type .as_ref() .or(severity_text.as_ref()) - .map(|field| { - format!( - "COALESCE(CAST({} AS VARCHAR), 'error')", - quote_identifier(field) - ) - }) - .unwrap_or_else(|| "'error'".to_owned()); + .map_or_else( + || "'error'".to_owned(), + |field| { + format!( + "COALESCE(CAST({} AS VARCHAR), 'error')", + quote_identifier(field) + ) + }, + ); let message = first_present_field( fields, &[ @@ -455,13 +458,14 @@ pub fn list_log_errors_query( "log", ], ); - let sample_message = message - .as_ref() - .map(|field| format!("MAX(CAST({} AS VARCHAR))", quote_identifier(field))) - .unwrap_or_else(|| "CAST(NULL AS VARCHAR)".to_owned()); - let last_seen = first_present_field(fields, &["p_timestamp", "time_unix_nano"]) - .map(|field| format!("MAX({})", quote_identifier(&field))) - .unwrap_or_else(|| "CAST(NULL AS VARCHAR)".to_owned()); + let sample_message = message.as_ref().map_or_else( + || "CAST(NULL AS VARCHAR)".to_owned(), + |field| format!("MAX(CAST({} AS VARCHAR))", quote_identifier(field)), + ); + let last_seen = first_present_field(fields, &["p_timestamp", "time_unix_nano"]).map_or_else( + || "CAST(NULL AS VARCHAR)".to_owned(), + |field| format!("MAX({})", quote_identifier(&field)), + ); let mut predicates = Vec::new(); if fields.iter().any(|field| field == "severity_number") { From 610d9c3073920b4e6d1a7d86cabe6d11f63d1f19 Mon Sep 17 00:00:00 2001 From: Nikhil Sinha Date: Tue, 13 Oct 2026 14:57:52 +0530 Subject: [PATCH 3/3] get_dataset_schema tool to return compact TSV response --- src/handlers/http/logstream.rs | 4 +++- src/tool_catalog.rs | 2 +- src/tool_executor.rs | 10 ++++------ 3 files changed, 8 insertions(+), 8 deletions(-) diff --git a/src/handlers/http/logstream.rs b/src/handlers/http/logstream.rs index fdfe0f84a..82c58bd70 100644 --- a/src/handlers/http/logstream.rs +++ b/src/handlers/http/logstream.rs @@ -205,7 +205,9 @@ fn schema_response(schema: &Schema, format: SchemaFormat) -> HttpResponse { } } -fn compact_schema(schema: &Schema) -> String { +/// Render the token-efficient schema representation used by the compact HTTP +/// response and by agent-facing schema tools. +pub fn compact_schema(schema: &Schema) -> String { let mut output = String::from("name\ttype\n"); for field in schema.fields() { append_escaped_compact_field(&mut output, field.name()); diff --git a/src/tool_catalog.rs b/src/tool_catalog.rs index c75f5eca3..70acf79f3 100644 --- a/src/tool_catalog.rs +++ b/src/tool_catalog.rs @@ -356,7 +356,7 @@ pub fn oss_read_tool_specs() -> Vec { ToolSpec::new( "get_dataset_schema", "Get dataset schema", - "Get columns and data types for a dataset.", + "Get all dataset columns and data types as compact tab-separated text with a name/type header.", dataset_schema(), ), ToolSpec::new( diff --git a/src/tool_executor.rs b/src/tool_executor.rs index c832a5846..c0eaeaf6d 100644 --- a/src/tool_executor.rs +++ b/src/tool_executor.rs @@ -272,12 +272,10 @@ async fn execute_dataset_metadata_tool( match name { "get_dataset_schema" => { authorize(context.session_key, Action::GetSchema, Some(dataset), None)?; - serde_json::to_value( - logstream::get_schema_internal(dataset, context.tenant_id) - .await - .map_err(|error| error.to_string())?, - ) - .map_err(|error| error.to_string()) + let schema = logstream::get_schema_internal(dataset, context.tenant_id) + .await + .map_err(|error| error.to_string())?; + Ok(Value::String(logstream::compact_schema(&schema))) } "get_dataset_info" => { authorize(