gru/RustProxy
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v0.26.2

Browse Source
This commit is contained in:
Mateusz Gruszczyński committed 2026-10-05 22:29:54 +02:00
1 parent b9d137eee1
commit 32e6875980
51 files changed
+1255 -452
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/admin/file_logs.rs
+42 -37
View File
@@ -4,7 +4,6 @@ const FILE_LOG_SCAN_LINES: usize = 20_000;
const FILE_LOG_MAX_SOURCES: usize = 32;
const FILE_LOG_MAX_LIMIT: usize = 500;
const FILE_LOG_MAX_OFFSET: usize = 10_000;
#[derive(Clone)]
struct FileLogSource {
id: String,
@@ -17,7 +16,6 @@ struct FileLogSource {
size: u64,
modified_ms: u64,
}
#[derive(Default, Deserialize)]
#[serde(default, deny_unknown_fields)]
struct FileLogQuery {
@@ -30,7 +28,6 @@ struct FileLogQuery {
offset: Option<usize>,
limit: Option<usize>,
}
fn file_modified_ms(metadata: &std::fs::Metadata) -> u64 {
metadata
.modified()
@@ -39,7 +36,6 @@ fn file_modified_ms(metadata: &std::fs::Metadata) -> u64 {
.map(|duration| duration.as_millis() as u64)
.unwrap_or(0)
}
fn source_from_path(
id: String,
label: String,
@@ -70,7 +66,6 @@ fn source_from_path(
.map_or(0, file_modified_ms),
}
}
fn configured_file_log_sources(config: &crate::config::Config) -> Vec<FileLogSource> {
let mut sources = Vec::new();
if config.logging.file.per_vhost {
@@ -134,7 +129,6 @@ fn configured_file_log_sources(config: &crate::config::Config) -> Vec<FileLogSou
));
sources
}
fn file_log_sources_json(sources: &[FileLogSource]) -> Value {
Value::Array(
sources
@@ -155,7 +149,6 @@ fn file_log_sources_json(sources: &[FileLogSource]) -> Value {
.collect(),
)
}
async fn file_logs(State(state): State<Arc<App>>) -> Result<Json<Value>, ApiError> {
let config = state.config.load().envelope.config.clone();
let sources = tokio::task::spawn_blocking(move || configured_file_log_sources(&config))
@@ -178,7 +171,6 @@ async fn file_logs(State(state): State<Arc<App>>) -> Result<Json<Value>, ApiErro
}
})))
}
fn open_file_log(path: &std::path::Path) -> std::io::Result<std::fs::File> {
let mut options = std::fs::OpenOptions::new();
options.read(true);
@@ -197,7 +189,6 @@ fn open_file_log(path: &std::path::Path) -> std::io::Result<std::fs::File> {
}
Ok(file)
}
fn read_tail(
source: &FileLogSource,
budget: u64,
@@ -228,7 +219,6 @@ fn read_tail(
.collect::<Vec<_>>();
Ok((lines, bytes, truncated || iter.next().is_some()))
}
fn copy_mapped_field(
object: &mut serde_json::Map<String, Value>,
format: &crate::logging::JsonFormat,
@@ -251,16 +241,35 @@ fn normalized_record(
line: &str,
format: &crate::logging::JsonFormat,
) -> Value {
let mut value = serde_json::from_str::<Value>(line).unwrap_or_else(|_| json!({"raw": line}));
let mut value = if source.format == "RAW" {
parse_raw_error_record(line)
} else {
serde_json::from_str::<Value>(line).unwrap_or_else(|_| json!({"raw": line}))
};
if !value.is_object() {
value = json!({"value": value});
}
let object = value.as_object_mut().expect("record normalized to object");
if source.kind == "access" {
for field in [
"event", "time_ms", "status", "client_ip", "vhost", "method", "host", "path",
"transport", "http_version", "protocol_error", "error_detail", "tls_sni", "reason",
"error", "sni", "authority", "matched_by",
"event",
"time_ms",
"status",
"client_ip",
"vhost",
"method",
"host",
"path",
"transport",
"http_version",
"protocol_error",
"error_detail",
"tls_sni",
"reason",
"error",
"sni",
"authority",
"matched_by",
] {
copy_mapped_field(object, format, field);
}
@@ -297,6 +306,7 @@ fn normalized_record(
});
object.insert("_source".into(), Value::String(source.id.clone()));
object.insert("_source_kind".into(), Value::String(source.kind.into()));
object.insert("_source_format".into(), Value::String(source.format.into()));
value
}
@@ -306,27 +316,12 @@ fn contains_ci(value: Option<&Value>, needle: &str) -> bool {
.is_some_and(|text| text.to_ascii_lowercase().contains(needle))
}
fn append_search_values(value: &Value, output: &mut String) {
match value {
Value::Null => {}
Value::Bool(value) => { output.push_str(if *value { " true" } else { " false" }); }
Value::Number(value) => { output.push(' '); output.push_str(&value.to_string()); }
Value::String(value) => { output.push(' '); output.push_str(&value.to_ascii_lowercase()); }
Value::Array(values) => values.iter().for_each(|value| append_search_values(value, output)),
Value::Object(values) => values.values().for_each(|value| append_search_values(value, output)),
}
}
fn record_matches(value: &Value, query: &FileLogQuery, q: &str) -> bool {
fn record_matches(value: &Value, query: &FileLogQuery, search: &[FileLogSearchClause]) -> bool {
let Some(object) = value.as_object() else {
return false;
};
if !q.is_empty() {
let mut searchable = String::new();
append_search_values(value, &mut searchable);
if !q.split_whitespace().all(|term| searchable.contains(term)) {
return false;
}
if !record_matches_search(value, search) {
return false;
}
if let Some(event) = query
.event
@@ -374,7 +369,7 @@ fn validate_file_log_query(query: &FileLogQuery) -> Result<(), ApiError> {
.is_none_or(|v| v.len() <= max && !v.chars().any(char::is_control))
};
if !text_ok(&query.source, 192)
|| !text_ok(&query.q, 256)
|| !text_ok(&query.q, 512)
|| !text_ok(&query.client_ip, 128)
|| !text_ok(&query.vhost, 256)
|| query.offset.unwrap_or(0) > FILE_LOG_MAX_OFFSET
@@ -407,6 +402,7 @@ async fn file_logs_query(
Query(query): Query<FileLogQuery>,
) -> Result<Json<Value>, 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);
@@ -428,7 +424,6 @@ async fn file_logs_query(
sources.truncate(FILE_LOG_MAX_SOURCES);
}
}
let q = query.q.clone().unwrap_or_default().to_ascii_lowercase();
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;
@@ -436,6 +431,7 @@ async fn file_logs_query(
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();
for (index, source) in sources.iter().enumerate() {
if remaining_bytes == 0 || remaining_lines == 0 {
truncated = true;
@@ -455,7 +451,8 @@ async fn file_logs_query(
continue;
}
let value = normalized_record(source, &line, &log_format);
if record_matches(&value, &query, &q) {
if record_matches(&value, &query, &search) {
collect_file_log_fields(&value, &mut fields);
records.push(value);
}
}
@@ -465,8 +462,15 @@ async fn file_logs_query(
}
}
records.sort_by(|a, b| {
let time = |value: &Value| value.get("time_ms").and_then(Value::as_u64).unwrap_or(0);
time(b).cmp(&time(a))
let numeric = |value: &Value| value.get("time_ms").and_then(Value::as_u64);
match (numeric(a), numeric(b)) {
(Some(left), Some(right)) => right.cmp(&left),
_ => b
.get("timestamp")
.and_then(Value::as_str)
.unwrap_or("")
.cmp(a.get("timestamp").and_then(Value::as_str).unwrap_or("")),
}
});
let total = records.len();
let page = records
@@ -482,6 +486,7 @@ async fn file_logs_query(
"truncated": truncated,
"scanned_bytes": scanned_bytes,
"scanned_lines": scanned_lines,
"fields": fields,
"sources": file_log_sources_json(&sources),
})))
})