fn contains_ci(value: Option<&Value>, needle: &str) -> bool { value .and_then(Value::as_str) .is_some_and(|text| text.to_ascii_lowercase().contains(needle)) } fn record_matches(value: &Value, query: &FileLogQuery, search: &[FileLogSearchClause]) -> bool { let Some(object) = value.as_object() else { return false; }; if !record_matches_search(value, search) { return false; } if let Some(event) = query .event .as_deref() .filter(|v| !v.is_empty() && *v != "all") { if object.get("event").and_then(Value::as_str) != Some(event) { return false; } } if let Some(class) = query .status_class .as_deref() .filter(|v| !v.is_empty() && *v != "all") { let Some(status) = object.get("status").and_then(Value::as_u64) else { return false; }; let expected = class .as_bytes() .first() .and_then(|v| v.checked_sub(b'0')) .map(u64::from); if expected != Some(status / 100) { return false; } } if let Some(client) = query.client_ip.as_deref().filter(|v| !v.is_empty()) { if !contains_ci(object.get("client_ip"), &client.to_ascii_lowercase()) { return false; } } if let Some(vhost) = query.vhost.as_deref().filter(|v| !v.is_empty()) { if !contains_ci(object.get("vhost"), &vhost.to_ascii_lowercase()) { return false; } } true } #[derive(Debug)] enum FileLogSortKey { Number(serde_json::Number), Text(String), } fn stored_log_summary(value: &Value) -> String { let field = |name| { value .get(name) .and_then(Value::as_str) .filter(|text| !text.is_empty()) }; if let Some(reason) = field("reason").or_else(|| field("protocol_error")) { return reason.to_owned(); } if field("method").is_some() || field("path").is_some() { return format!( "{} {}{}", field("method").unwrap_or(""), field("host").or_else(|| field("authority")).unwrap_or(""), field("path").unwrap_or("") ) .trim() .to_owned(); } for key in ["error", "error_detail", "message", "raw", "outcome"] { if let Some(text) = field(key) { return text.to_owned(); } } String::new() } fn stored_log_sort_key(record: &Value, field: &str) -> Option { if field == "record" { return Some(FileLogSortKey::Text( stored_log_summary(record).to_lowercase(), )); } let value = if field == "time" { record .get("time_ms") .filter(|value| value.is_number()) .or_else(|| record.get("timestamp")) } else { file_log_field(record, field) }?; match value { Value::Null => None, Value::Number(number) => Some(FileLogSortKey::Number(number.clone())), Value::String(text) if !text.is_empty() => Some(FileLogSortKey::Text(text.to_lowercase())), Value::Bool(boolean) => Some(FileLogSortKey::Text(boolean.to_string())), Value::Array(_) | Value::Object(_) => { Some(FileLogSortKey::Text(value.to_string().to_lowercase())) } _ => None, } } fn file_log_timestamp_cmp_desc(left: &Value, right: &Value) -> std::cmp::Ordering { match ( left.get("time_ms").and_then(Value::as_u64), right.get("time_ms").and_then(Value::as_u64), ) { (Some(a), Some(b)) => b.cmp(&a), _ => right .get("timestamp") .and_then(Value::as_str) .unwrap_or("") .cmp(left.get("timestamp").and_then(Value::as_str).unwrap_or("")), } } fn sort_file_log_records(records: &mut Vec, field: &str, descending: bool) { use std::cmp::Ordering; let mut keyed = std::mem::take(records) .into_iter() .enumerate() .map(|(ordinal, record)| (stored_log_sort_key(&record, field), ordinal, record)) .collect::>(); keyed.sort_by(|left, right| { let order = match (&left.0, &right.0) { (None, None) => Ordering::Equal, (None, Some(_)) => return Ordering::Greater, (Some(_), None) => return Ordering::Less, (Some(FileLogSortKey::Number(a)), Some(FileLogSortKey::Number(b))) => { match (a.as_i64(), b.as_i64()) { (Some(left), Some(right)) => left.cmp(&right), _ => match (a.as_u64(), b.as_u64()) { (Some(left), Some(right)) => left.cmp(&right), _ => a .as_f64() .partial_cmp(&b.as_f64()) .unwrap_or(Ordering::Equal), }, } } (Some(FileLogSortKey::Text(a)), Some(FileLogSortKey::Text(b))) => a.cmp(b), (Some(FileLogSortKey::Number(_)), Some(FileLogSortKey::Text(_))) => Ordering::Less, (Some(FileLogSortKey::Text(_)), Some(FileLogSortKey::Number(_))) => Ordering::Greater, }; (if descending { order.reverse() } else { order }) .then_with(|| file_log_timestamp_cmp_desc(&left.2, &right.2)) .then_with(|| left.1.cmp(&right.1)) }); records.extend(keyed.into_iter().map(|(_, _, record)| record)); } fn validate_file_log_query(query: &FileLogQuery) -> Result<(), ApiError> { let text_ok = |value: &Option, max: usize| { value .as_ref() .is_none_or(|v| v.len() <= max && !v.chars().any(char::is_control)) }; if !text_ok(&query.source, 192) || !text_ok(&query.q, 512) || !text_ok(&query.client_ip, 128) || !text_ok(&query.vhost, 256) || query .aggregate_field .as_ref() .is_some_and(|field| !valid_raw_field_key(field)) || query .sort_by .as_ref() .is_some_and(|field| !valid_raw_field_key(field)) || query .aggregate_limit .is_some_and(|value| value == 0 || value > FILE_LOG_MAX_AGGREGATE_RESULTS) || query .aggregate_min_count .is_some_and(|value| !(2..=FILE_LOG_SCAN_LINES).contains(&value)) || query.offset.unwrap_or(0) > FILE_LOG_MAX_OFFSET || query .limit .is_some_and(|v| v == 0 || v > FILE_LOG_MAX_LIMIT) || query.event.as_ref().is_some_and(|v| { !matches!( v.as_str(), "" | "all" | "request" | "protocol_error" | "error" ) }) || query.status_class.as_ref().is_some_and(|v| { !matches!( v.as_str(), "" | "all" | "1xx" | "2xx" | "3xx" | "4xx" | "5xx" ) }) { return Err(ApiError( StatusCode::BAD_REQUEST, "invalid file log filter".into(), )); } Ok(()) } async fn file_logs_query( State(state): State>, Query(query): Query, ) -> Result, ApiError> { validate_file_log_query(&query)?; let search = parse_file_log_search(query.q.as_deref().unwrap_or(""))?; let config = state.config.load().envelope.config.clone(); tokio::task::spawn_blocking(move || { let mut sources = configured_file_log_sources(&config); let log_format = config.logging.format.clone(); let requested = query.source.as_deref().unwrap_or("all"); let mut truncated = false; if requested != "all" { let consolidate_vhosts = query.consolidate_vhosts.unwrap_or(false) && requested.starts_with("vhost:"); sources.retain(|source| { source.id == requested || (consolidate_vhosts && source.group_id == requested) }); if sources.is_empty() { return Err(ApiError( StatusCode::BAD_REQUEST, "unknown file log source".into(), )); } } else { sources.sort_by_key(|source| std::cmp::Reverse(source.modified_ms)); if sources.len() > FILE_LOG_MAX_SOURCES { truncated = true; sources.truncate(FILE_LOG_MAX_SOURCES); } } let offset = query.offset.unwrap_or(0); let limit = query.limit.unwrap_or(100).clamp(1, FILE_LOG_MAX_LIMIT); let mut remaining_bytes = FILE_LOG_SCAN_BYTES_TOTAL; let mut scanned_bytes = 0u64; let mut remaining_lines = FILE_LOG_SCAN_LINES; let mut scanned_lines = 0usize; let mut records = Vec::new(); let mut fields = std::collections::BTreeSet::new(); let aggregate_field = query .aggregate_field .as_deref() .map(str::to_ascii_lowercase); let aggregate_min_count = query.aggregate_min_count.unwrap_or(2); let aggregate_limit = query .aggregate_limit .unwrap_or(50) .clamp(1, FILE_LOG_MAX_AGGREGATE_RESULTS); let mut aggregate_counts = std::collections::HashMap::::new(); let mut aggregate_truncated = false; for (index, source) in sources.iter().enumerate() { if remaining_bytes == 0 || remaining_lines == 0 { truncated = true; break; } let sources_left = sources.len().saturating_sub(index).max(1); let line_budget = (remaining_lines / sources_left).max(1); match read_tail(source, remaining_bytes, line_budget) { Ok((lines, bytes, source_truncated)) => { remaining_bytes = remaining_bytes.saturating_sub(bytes); remaining_lines = remaining_lines.saturating_sub(lines.len()); scanned_bytes += bytes; scanned_lines += lines.len(); truncated |= source_truncated; for line in lines { if line.is_empty() { continue; } let value = normalized_record(source, &line, &log_format); if record_matches(&value, &query, &search) { collect_file_log_fields(&value, &mut fields); if let Some(field) = aggregate_field.as_deref() { if let Some(field_value) = file_log_field(&value, field) { let mut values = Vec::new(); file_log_aggregate_values(field_value, &mut values); for aggregate_value in values { if aggregate_counts.contains_key(&aggregate_value) || aggregate_counts.len() < FILE_LOG_MAX_AGGREGATE_VALUES { *aggregate_counts .entry(aggregate_value) .or_insert(0) += 1; } else { aggregate_truncated = true; } } } } records.push(value); } } } Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} Err(_) => truncated = true, } } sort_file_log_records( &mut records, query.sort_by.as_deref().unwrap_or("time"), query.sort_descending.unwrap_or(true), ); let mut aggregation = aggregate_counts .into_iter() .filter(|(_, count)| *count >= aggregate_min_count) .map(|(value, count)| json!({"value": value, "count": count})) .collect::>(); aggregation.sort_by(|left, right| { let left_count = left.get("count").and_then(Value::as_u64).unwrap_or(0); let right_count = right.get("count").and_then(Value::as_u64).unwrap_or(0); right_count.cmp(&left_count).then_with(|| { left.get("value") .and_then(Value::as_str) .unwrap_or("") .cmp(right.get("value").and_then(Value::as_str).unwrap_or("")) }) }); if aggregation.len() > aggregate_limit { aggregation.truncate(aggregate_limit); aggregate_truncated = true; } let total = records.len(); let page = records .into_iter() .skip(offset) .take(limit) .collect::>(); Ok(Json(json!({ "entries": page, "total": total, "offset": offset, "limit": limit, "truncated": truncated, "scanned_bytes": scanned_bytes, "scanned_lines": scanned_lines, "fields": fields, "sources": file_log_sources_json(&sources), "aggregation": aggregate_field.map(|field| json!({ "field": field, "min_count": aggregate_min_count, "entries": aggregation, "truncated": aggregate_truncated, })), }))) }) .await .map_err(|_| { ApiError( StatusCode::INTERNAL_SERVER_ERROR, "file log query worker failed".into(), ) })? }