Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion src/handlers/http/logstream.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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());
Expand Down
72 changes: 68 additions & 4 deletions src/tool_catalog.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -354,7 +356,7 @@ pub fn oss_read_tool_specs() -> Vec<ToolSpec> {
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(
Expand Down Expand Up @@ -402,6 +404,34 @@ pub fn oss_read_tool_specs() -> Vec<ToolSpec> {
&["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",
Expand Down Expand Up @@ -824,7 +854,7 @@ mod tests {
let tools = oss_tool_specs();
let names = tools.iter().map(|tool| tool.name).collect::<HashSet<_>>();

assert_eq!(tools.len(), 47);
assert_eq!(tools.len(), 49);
assert_eq!(tools.len(), names.len());
}

Expand Down Expand Up @@ -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(),
Expand Down Expand Up @@ -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!({
Expand Down
189 changes: 179 additions & 10 deletions src/tool_executor.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down Expand Up @@ -409,6 +407,152 @@ async fn get_log_context(arguments: &Value, context: OssToolContext<'_>) -> Resu
response_value(response).await
}

fn first_present_field(fields: &[String], candidates: &[&str]) -> Option<String> {
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<String, String> {
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)
)
},
);
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_or_else(
|| "'error'".to_owned(),
|field| {
format!(
"COALESCE(CAST({} AS VARCHAR), 'error')",
quote_identifier(field)
)
},
);
let message = first_present_field(
fields,
&[
"event_exception.message",
"exception.message",
"error.message",
"body",
"message",
"log",
],
);
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") {
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<Value, String> {
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::<Vec<_>>();
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,
Expand Down Expand Up @@ -563,7 +707,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 {
Expand Down Expand Up @@ -596,7 +741,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),
},
Expand Down Expand Up @@ -818,13 +964,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"
Expand Down Expand Up @@ -894,7 +1043,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() {
Expand Down
Loading