martino/rmon
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Working on better plotting

Browse Source
This commit is contained in:
Martino Ferrari
2026-04-12 16:20:44 +02:00
parent 1a39b3a54e
commit 41410e4e3d
10 changed files with 2088 additions and 1064 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rmon-agent/Cargo.toml
+13 -28
View File
@@ -1,39 +1,24 @@
[package]
name = "rmon-agent"
version.workspace = true
edition.workspace = true
authors.workspace = true
license.workspace = true
version = "0.1.0"
edition = "2021"
[dependencies]
anyhow.workspace = true
rmon-common = { path = "../rmon-common" }
tokio.workspace = true
tokio-util.workspace = true
serde.workspace = true
serde_json.workspace = true
anyhow.workspace = true
toml.workspace = true
tracing.workspace = true
tracing-subscriber.workspace = true
serde.workspace = true
chrono.workspace = true
toml.workspace = true
bincode.workspace = true
async-trait.workspace = true
futures.workspace = true
tokio-util.workspace = true
regex.workspace = true
notify.workspace = true
clap.workspace = true
futures.workspace = true
bytes.workspace = true
itertools.workspace = true
byteorder.workspace = true
async-trait.workspace = true
rmon-common.workspace = true
[dev-dependencies]
tokio = { workspace = true, features = ["full"] }
futures.workspace = true
toml.workspace = true
anyhow.workspace = true
tempfile.workspace = true
[profile.release]
strip = true
lto = true
codegen-units = 1
panic = "abort"
byteorder.workspace = true
chrono.workspace = true
bytes.workspace = true
rmon-agent/src/server/session.rs
+26 -1
View File
@@ -117,12 +117,37 @@ pub async fn handle_session(
tracing::info!("[{}] Updating config...", addr);
if let Err(e) = std::fs::write(&state.config_path, &config_toml) {
tracing::error!("[{}] Config write failed: {}", addr, e);
framed.send(AgentMessage::Error { context: "UpdateConfig".to_string(), message: e.to_string() }).await?;
let _ = framed.send(AgentMessage::Error { context: "UpdateConfig".to_string(), message: e.to_string() }).await;
} else {
let _ = state.reload_tx.send(());
}
}
ClientMessage::QueryHistory { id, from_ns, to_ns, max_points } => {
tracing::info!("[{}] QueryHistory for {}", addr, id);
let source_name = id.split("://").nth(1).and_then(|s| s.split('/').next()).unwrap_or("unknown");
match crate::storage::ContinuousStorage::read_range(
state.config_path.parent().unwrap().join("data"), // Assuming data is in same dir as config
source_name,
&id,
from_ns,
to_ns,
max_points as usize
) {
Ok(samples) => {
framed.send(AgentMessage::HistoryChunk {
id,
samples,
done: true
}).await?;
}
Err(e) => {
tracing::error!("[{}] QueryHistory failed: {}", addr, e);
let _ = framed.send(AgentMessage::Error { context: "QueryHistory".to_string(), message: e.to_string() }).await;
}
}
}
ClientMessage::Subscribe { ids, .. } => {
tracing::info!("[{}] Subscribing to {:?}", addr, ids);
subscriptions.extend(ids);
}
rmon-agent/src/storage/continuous.rs
+58 -37
View File
@@ -1,32 +1,29 @@
use std::fs::{File, OpenOptions};
use std::io::{Write, BufWriter};
use std::path::PathBuf;
use anyhow::Result;
use rmon_common::signal::Sample;
use byteorder::{LittleEndian, WriteBytesExt};
use std::fs::File;
use std::io::{BufWriter, BufReader, Write};
use std::path::PathBuf;
use byteorder::{LittleEndian, WriteBytesExt, ReadBytesExt};
pub struct ContinuousStorage {
writer: BufWriter<File>,
_path: PathBuf,
}
impl ContinuousStorage {
pub fn create(data_dir: PathBuf, source_name: &str, signal_id: &str, session_start_ns: i64) -> Result<Self> {
pub fn create(data_dir: PathBuf, source_name: &str, signal_id: &str, session_start: i64) -> Result<Self> {
let mut path = data_dir;
path.push(source_name);
path.push(signal_id.replace(":", "_").replace("/", "_")); // Escape signal ID for filesystem
path.push(signal_id.replace(":", "_").replace("/", "_"));
std::fs::create_dir_all(&path)?;
path.push(format!("{}.bin", session_start_ns));
let file = OpenOptions::new()
let filename = format!("{}.bin", session_start);
let file = std::fs::OpenOptions::new()
.create(true)
.append(true)
.open(&path)?;
.open(path.join(filename))?;
Ok(Self {
writer: BufWriter::new(file),
_path: path,
})
}
@@ -40,43 +37,67 @@ impl ContinuousStorage {
self.writer.flush()?;
Ok(())
}
pub fn read_range(data_dir: PathBuf, source_name: &str, signal_id: &str, from_ns: i64, to_ns: i64, max_points: usize) -> Result<Vec<Sample>> {
let mut path = data_dir;
path.push(source_name);
path.push(signal_id.replace(":", "_").replace("/", "_"));
if !path.exists() { return Ok(vec![]); }
let mut all_samples = Vec::new();
if let Ok(entries) = std::fs::read_dir(path) {
for entry in entries {
let entry = entry?;
if entry.path().extension().and_then(|s| s.to_str()) == Some("bin") {
let file = File::open(entry.path())?;
let file_len = file.metadata()?.len();
let num_samples = (file_len / 16) as usize;
let mut reader = BufReader::new(file);
for _ in 0..num_samples {
let ts = reader.read_i64::<LittleEndian>()?;
let val = reader.read_f64::<LittleEndian>()?;
if ts >= from_ns && ts <= to_ns {
all_samples.push(Sample { timestamp: ts, value: val });
}
}
}
}
}
all_samples.sort_by_key(|s| s.timestamp);
if all_samples.len() > max_points && max_points > 0 {
let step = all_samples.len() / max_points;
Ok(all_samples.into_iter().step_by(step).take(max_points).collect())
} else {
Ok(all_samples)
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use tempfile::tempdir;
use std::fs;
#[test]
fn test_continuous_storage_creation() {
let dir = tempdir().unwrap();
let res = ContinuousStorage::create(
dir.path().to_path_buf(),
"src",
"sig1",
12345
);
assert!(res.is_ok());
let mut storage = ContinuousStorage::create(dir.path().to_path_buf(), "src", "sig", 123).unwrap();
storage.append(Sample { timestamp: 1, value: 1.0 }).unwrap();
storage.flush().unwrap();
let samples = ContinuousStorage::read_range(dir.path().to_path_buf(), "src", "sig", 0, 10, 100).unwrap();
assert_eq!(samples.len(), 1);
assert_eq!(samples[0].value, 1.0);
}
#[test]
fn test_continuous_append() {
let dir = tempdir().unwrap();
let mut storage = ContinuousStorage::create(
dir.path().to_path_buf(),
"src",
"sig1",
12345
).unwrap();
let sample = Sample { timestamp: 100, value: 42.0 };
storage.append(sample).unwrap();
storage.flush().unwrap();
// Verify file exists and has size 16 bytes (i64 + f64)
let path = dir.path().join("src/sig1/12345.bin");
assert!(path.exists());
assert_eq!(fs::metadata(path).unwrap().len(), 16);
let mut storage = ContinuousStorage::create(dir.path().to_path_buf(), "s", "i", 1).unwrap();
assert!(storage.append(Sample { timestamp: 1, value: 1.0 }).is_ok());
}
}