Split src/ into an api/cli/engine/config module hierarchy
- src/api/: HTTP router + DTOs + the MCP server (the HTTP/MCP-facing surface): mod.rs (formerly api.rs), dto.rs, mcp_server.rs. - src/cli/: argument parsing and process entry (formerly inline in main.rs) plus session registry: mod.rs, sessions.rs. main.rs is now a thin entry point that just declares modules and calls cli::run(). - src/engine/: the actor + everything it depends on to parse/index/ query log lines: mod.rs (formerly engine.rs), alerts.rs, promql.rs, rate.rs, ingest.rs, logfmt.rs, filter_parser.rs/.pest, filter_datetime.rs, index_key.rs, data_value.rs, metric_primitive.rs, timestamp.rs. - src/config/: persisted configuration: mod.rs (formerly settings.rs), profile.rs. - src/consts.rs stays at the crate root, used from every module above. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01A1ZJ3SrayzLzVndhVq9BmS
This commit is contained in:
Hugh Kaznowski
co-authored by
Claude Sonnet 5
Claude Sonnet 5
parent
311713986a
commit
a810ecf396
@@ -0,0 +1,319 @@
|
||||
use std::collections::BTreeMap;
|
||||
use std::path::{Path, PathBuf};
|
||||
|
||||
use crate::consts::{METRICS_POLL_INTERVAL, STDIN_POLL_INTERVAL as POLL_INTERVAL};
|
||||
use crate::engine::{EngineHandle, EngineMessage, IngestedLine, MetricSample};
|
||||
use crate::engine::timestamp::Timestamp;
|
||||
use tokio::fs::File;
|
||||
use tokio::io::{AsyncBufReadExt, BufReader};
|
||||
|
||||
pub async fn run(engine: EngineHandle) {
|
||||
let stdin = tokio::io::stdin();
|
||||
let mut lines = BufReader::new(stdin).lines();
|
||||
let mut sequence: u64 = 0;
|
||||
|
||||
while let Ok(Some(text)) = lines.next_line().await {
|
||||
let line = IngestedLine {
|
||||
sequence,
|
||||
timestamp: Timestamp::now(),
|
||||
text,
|
||||
};
|
||||
sequence += 1;
|
||||
|
||||
if engine.tx.send(EngineMessage::Ingest(line)).await.is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(PartialEq, Eq, Clone, Copy)]
|
||||
struct FileId(#[cfg(unix)] u64, #[cfg(not(unix))] ());
|
||||
|
||||
fn file_id(meta: &std::fs::Metadata) -> FileId {
|
||||
#[cfg(unix)]
|
||||
{
|
||||
use std::os::unix::fs::MetadataExt;
|
||||
FileId(meta.ino())
|
||||
}
|
||||
#[cfg(not(unix))]
|
||||
{
|
||||
let _ = meta;
|
||||
FileId(())
|
||||
}
|
||||
}
|
||||
|
||||
async fn open(path: &Path) -> Option<(File, FileId)> {
|
||||
let file = File::open(path).await.ok()?;
|
||||
let meta = file.metadata().await.ok()?;
|
||||
Some((file, file_id(&meta)))
|
||||
}
|
||||
|
||||
pub async fn run_file(engine: EngineHandle, path: PathBuf) {
|
||||
let mut sequence: u64 = 0;
|
||||
|
||||
'reopen: loop {
|
||||
let (file, mut id) = loop {
|
||||
if let Some(opened) = open(&path).await {
|
||||
break opened;
|
||||
}
|
||||
tokio::time::sleep(POLL_INTERVAL).await;
|
||||
};
|
||||
|
||||
let mut reader = BufReader::new(file);
|
||||
let mut offset: u64 = 0;
|
||||
let mut pending = String::new();
|
||||
|
||||
loop {
|
||||
let mut chunk = String::new();
|
||||
match reader.read_line(&mut chunk).await {
|
||||
Ok(0) => {
|
||||
match tokio::fs::metadata(&path).await {
|
||||
Ok(meta) => {
|
||||
let new_id = file_id(&meta);
|
||||
if new_id != id || meta.len() < offset {
|
||||
continue 'reopen;
|
||||
}
|
||||
id = new_id;
|
||||
}
|
||||
Err(_) => {}
|
||||
}
|
||||
tokio::time::sleep(POLL_INTERVAL).await;
|
||||
}
|
||||
Ok(n) => {
|
||||
offset += n as u64;
|
||||
pending.push_str(&chunk);
|
||||
if pending.ends_with('\n') {
|
||||
let text = pending
|
||||
.trim_end_matches(['\n', '\r'])
|
||||
.to_string();
|
||||
pending.clear();
|
||||
|
||||
let line = IngestedLine {
|
||||
sequence,
|
||||
timestamp: Timestamp::now(),
|
||||
text,
|
||||
};
|
||||
sequence += 1;
|
||||
|
||||
if engine.tx.send(EngineMessage::Ingest(line)).await.is_err() {
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(_) => continue 'reopen,
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn run_metrics_poll(engine: EngineHandle, url: String) {
|
||||
let client = reqwest::Client::new();
|
||||
loop {
|
||||
match client.get(&url).send().await {
|
||||
Ok(response) => match response.error_for_status() {
|
||||
Ok(response) => match response.text().await {
|
||||
Ok(body) => {
|
||||
let now = Timestamp::now();
|
||||
for sample in parse_openmetrics(&body, now) {
|
||||
if engine.tx.send(EngineMessage::IngestMetricSample(sample)).await.is_err() {
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(err) => eprintln!("warning: failed to read metrics response from {url}: {err}"),
|
||||
},
|
||||
Err(err) => eprintln!("warning: metrics endpoint {url} returned an error: {err}"),
|
||||
},
|
||||
Err(err) => eprintln!("warning: failed to poll metrics endpoint {url}: {err}"),
|
||||
}
|
||||
tokio::time::sleep(METRICS_POLL_INTERVAL).await;
|
||||
}
|
||||
}
|
||||
|
||||
fn parse_openmetrics(body: &str, now: Timestamp) -> Vec<MetricSample> {
|
||||
body.lines().filter_map(|line| parse_openmetrics_line(line, now)).collect()
|
||||
}
|
||||
|
||||
fn parse_openmetrics_line(line: &str, now: Timestamp) -> Option<MetricSample> {
|
||||
let line = line.trim();
|
||||
if line.is_empty() || line.starts_with('#') {
|
||||
return None;
|
||||
}
|
||||
|
||||
let (name_and_labels, rest) = line.split_once(char::is_whitespace)?;
|
||||
let mut fields = rest.split_whitespace();
|
||||
let value: f64 = fields.next()?.parse().ok()?;
|
||||
let timestamp = fields
|
||||
.next()
|
||||
.and_then(|raw| raw.parse::<f64>().ok())
|
||||
.map(|ms| Timestamp::from_millis(ms as i64))
|
||||
.unwrap_or(now);
|
||||
|
||||
let (metric_name, labels) = match name_and_labels.find('{') {
|
||||
Some(open) => {
|
||||
let close = name_and_labels.rfind('}')?;
|
||||
(name_and_labels[..open].to_string(), parse_openmetrics_labels(&name_and_labels[open + 1..close]))
|
||||
}
|
||||
None => (name_and_labels.to_string(), BTreeMap::new()),
|
||||
};
|
||||
if metric_name.is_empty() {
|
||||
return None;
|
||||
}
|
||||
|
||||
Some(MetricSample { metric_name, labels, value, timestamp })
|
||||
}
|
||||
|
||||
fn parse_openmetrics_labels(raw: &str) -> BTreeMap<String, String> {
|
||||
raw.split(',')
|
||||
.filter_map(|pair| pair.split_once('='))
|
||||
.map(|(name, value)| (name.trim().to_string(), value.trim().trim_matches('"').to_string()))
|
||||
.collect()
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod openmetrics_tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn parses_bare_metric_without_labels_or_timestamp() {
|
||||
let now = Timestamp::from_millis(1000);
|
||||
let samples = parse_openmetrics("requests_total 42\n", now);
|
||||
assert_eq!(samples.len(), 1);
|
||||
assert_eq!(samples[0].metric_name, "requests_total");
|
||||
assert_eq!(samples[0].value, 42.0);
|
||||
assert!(samples[0].labels.is_empty());
|
||||
assert_eq!(samples[0].timestamp, now);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parses_labels_and_explicit_timestamp() {
|
||||
let now = Timestamp::from_millis(1000);
|
||||
let samples = parse_openmetrics(
|
||||
r#"http_requests_total{method="GET",path="/a"} 7 5000"#,
|
||||
now,
|
||||
);
|
||||
assert_eq!(samples.len(), 1);
|
||||
let sample = &samples[0];
|
||||
assert_eq!(sample.metric_name, "http_requests_total");
|
||||
assert_eq!(sample.value, 7.0);
|
||||
assert_eq!(sample.labels.get("method").map(String::as_str), Some("GET"));
|
||||
assert_eq!(sample.labels.get("path").map(String::as_str), Some("/a"));
|
||||
assert_eq!(sample.timestamp, Timestamp::from_millis(5000));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn skips_comments_blank_lines_and_malformed_lines() {
|
||||
let now = Timestamp::from_millis(0);
|
||||
let body = "# HELP requests_total docs\n# TYPE requests_total counter\n\nnot a valid line\nrequests_total 1\n";
|
||||
let samples = parse_openmetrics(body, now);
|
||||
assert_eq!(samples.len(), 1);
|
||||
assert_eq!(samples[0].metric_name, "requests_total");
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::time::Duration;
|
||||
use std::io::Write;
|
||||
use crate::engine::test_handle;
|
||||
use tokio::sync::mpsc;
|
||||
|
||||
async fn recv_text(rx: &mut mpsc::Receiver<EngineMessage>) -> String {
|
||||
match tokio::time::timeout(Duration::from_secs(2), rx.recv())
|
||||
.await
|
||||
.expect("timed out waiting for ingested line")
|
||||
.expect("engine channel closed")
|
||||
{
|
||||
EngineMessage::Ingest(line) => line.text,
|
||||
_ => panic!("expected Ingest"),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn tails_existing_and_appended_lines() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let path = dir.path().join("app.log");
|
||||
std::fs::write(&path, "first\n").unwrap();
|
||||
|
||||
let (tx, mut rx) = test_handle(16);
|
||||
tokio::spawn(run_file(tx, path.clone()));
|
||||
|
||||
assert_eq!(recv_text(&mut rx).await, "first");
|
||||
|
||||
let mut f = std::fs::OpenOptions::new().append(true).open(&path).unwrap();
|
||||
writeln!(f, "second").unwrap();
|
||||
assert_eq!(recv_text(&mut rx).await, "second");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn resumes_after_delete_and_recreate() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let path = dir.path().join("app.log");
|
||||
std::fs::write(&path, "before-restart\n").unwrap();
|
||||
|
||||
let (tx, mut rx) = test_handle(16);
|
||||
tokio::spawn(run_file(tx, path.clone()));
|
||||
|
||||
assert_eq!(recv_text(&mut rx).await, "before-restart");
|
||||
|
||||
std::fs::remove_file(&path).unwrap();
|
||||
std::fs::write(&path, "after-restart\n").unwrap();
|
||||
|
||||
assert_eq!(recv_text(&mut rx).await, "after-restart");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn resumes_after_truncation_in_place() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let path = dir.path().join("app.log");
|
||||
std::fs::write(&path, "long line before truncation\n").unwrap();
|
||||
|
||||
let (tx, mut rx) = test_handle(16);
|
||||
tokio::spawn(run_file(tx, path.clone()));
|
||||
|
||||
assert_eq!(recv_text(&mut rx).await, "long line before truncation");
|
||||
|
||||
let mut f = std::fs::OpenOptions::new()
|
||||
.write(true)
|
||||
.truncate(true)
|
||||
.open(&path)
|
||||
.unwrap();
|
||||
writeln!(f, "new").unwrap();
|
||||
|
||||
assert_eq!(recv_text(&mut rx).await, "new");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn sequence_keeps_incrementing_across_recreation() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let path = dir.path().join("app.log");
|
||||
std::fs::write(&path, "one\n").unwrap();
|
||||
|
||||
let (tx, mut rx) = test_handle(16);
|
||||
tokio::spawn(run_file(tx, path.clone()));
|
||||
|
||||
let first = match tokio::time::timeout(Duration::from_secs(2), rx.recv())
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap()
|
||||
{
|
||||
EngineMessage::Ingest(line) => line.sequence,
|
||||
_ => panic!("expected Ingest"),
|
||||
};
|
||||
assert_eq!(first, 0);
|
||||
|
||||
std::fs::remove_file(&path).unwrap();
|
||||
std::fs::write(&path, "two\n").unwrap();
|
||||
|
||||
let second = match tokio::time::timeout(Duration::from_secs(2), rx.recv())
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap()
|
||||
{
|
||||
EngineMessage::Ingest(line) => line.sequence,
|
||||
_ => panic!("expected Ingest"),
|
||||
};
|
||||
assert_eq!(second, 1);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user