From d67d0ca063ae19909ac801650fa3de8f0354d206 Mon Sep 17 00:00:00 2001 From: stefankumaraaisnghe Date: Sat, 18 Apr 2026 01:01:00 +1000 Subject: [PATCH 01/19] (feat) add parquet archive engine with replay and archive modes --- Cargo.toml | 5 + crates/host-collectors/Cargo.toml | 3 + crates/host-collectors/src/lib.rs | 820 +++++++++++++++++++++++++++- crates/host-collectors/src/tests.rs | 85 ++- src/bin/archive-replay.rs | 735 +++++++++++++++++++++++++ src/config.rs | 10 +- src/tests/config_tests.rs | 4 + src/tests/otel_tests.rs | 4 + 8 files changed, 1639 insertions(+), 27 deletions(-) create mode 100644 src/bin/archive-replay.rs diff --git a/Cargo.toml b/Cargo.toml index c55b8d5..cde16db 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -2,6 +2,7 @@ name = "ojo" version = "0.0.3" edition = "2021" +default-run = "ojo" [workspace] resolver = "2" @@ -27,6 +28,10 @@ ctrlc = "3" serde = { version = "1", features = ["derive"] } serde_yaml = "0.9" serde_json = "1" +arrow-array = "58" +parquet = { version = "58", default-features = false, features = ["arrow"] } +prost = "0.13" +snap = "1" tokio = { version = "1", features = [ "macros", "rt-multi-thread", diff --git a/crates/host-collectors/Cargo.toml b/crates/host-collectors/Cargo.toml index f2c1cfa..736089e 100644 --- a/crates/host-collectors/Cargo.toml +++ b/crates/host-collectors/Cargo.toml @@ -7,6 +7,9 @@ edition = "2021" anyhow = "1" gethostname = "1" serde_json = "1" +arrow-array = "58" +arrow-schema = "58" +parquet = { version = "58", default-features = false, features = ["arrow", "zstd"] } opentelemetry = { version = "0.31", features = ["metrics", "trace"] } opentelemetry_sdk = { version = "0.31", features = ["metrics", "trace", "rt-tokio", "experimental_trace_batch_span_processor_with_async_runtime"] } opentelemetry-otlp = { version = "0.31", features = [ diff --git a/crates/host-collectors/src/lib.rs b/crates/host-collectors/src/lib.rs index bd605a4..8ad39ed 100644 --- a/crates/host-collectors/src/lib.rs +++ b/crates/host-collectors/src/lib.rs @@ -8,15 +8,99 @@ use opentelemetry_sdk::{ resource::Resource, trace::{BatchConfigBuilder, BatchSpanProcessor, SdkTracerProvider, SpanData, SpanExporter}, }; +use parquet::arrow::arrow_writer::ArrowWriter; +use parquet::basic::{Compression, ZstdLevel}; +use parquet::file::properties::WriterProperties; use serde_json::Value; -use std::collections::BTreeMap; -use std::fs::{self, OpenOptions}; -use std::io::Write; +use std::collections::{BTreeMap, HashMap}; +use std::fs::{self, File}; +use std::hash::{Hash, Hasher}; use std::path::Path; use std::sync::atomic::{AtomicBool, Ordering}; -use std::time::Duration; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use arrow_array::{ArrayRef, Float64Array, Int64Array, RecordBatch, StringArray, UInt64Array}; +use arrow_schema::{DataType, Field, Schema}; +use std::sync::Arc; pub const METRIC_PREFIX_SYSTEM: &str = "system."; +const ARCHIVE_SCHEMA_VERSION: &str = "v1"; +const DEFAULT_ARCHIVE_WINDOW_SECS: u64 = 60; +const DEFAULT_TOP_K_LOG_SIGNATURES: usize = 3; + +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub enum ArchiveFormat { + #[default] + Parquet, +} + +impl ArchiveFormat { + pub fn parse(value: Option<&str>) -> Self { + match value.map(|v| v.trim().to_ascii_lowercase()) { + Some(v) if v == "parquet" || v.is_empty() => Self::Parquet, + _ => Self::Parquet, + } + } + + pub fn as_str(&self) -> &'static str { + match self { + Self::Parquet => "parquet", + } + } +} + +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub enum ArchiveMode { + #[default] + Trend, + Forensic, + Lossless, +} + +impl ArchiveMode { + pub fn parse(value: Option<&str>) -> Self { + match value.map(|v| v.trim().to_ascii_lowercase()) { + Some(v) if v == "forensic" => Self::Forensic, + Some(v) if v == "lossless" => Self::Lossless, + _ => Self::Trend, + } + } + + pub fn as_str(&self) -> &'static str { + match self { + Self::Trend => "trend", + Self::Forensic => "forensic", + Self::Lossless => "lossless", + } + } +} + +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub enum ArchiveCompression { + #[default] + Zstd, +} + +impl ArchiveCompression { + pub fn parse(value: Option<&str>) -> Self { + match value.map(|v| v.trim().to_ascii_lowercase()) { + Some(v) if v == "zstd" || v.is_empty() => Self::Zstd, + _ => Self::Zstd, + } + } + + fn to_parquet(&self) -> Compression { + match self { + Self::Zstd => Compression::ZSTD(ZstdLevel::try_new(3).unwrap_or_default()), + } + } + + pub fn as_str(&self) -> &'static str { + match self { + Self::Zstd => "zstd", + } + } +} #[derive(Clone, Debug)] pub struct ArchiveStorageConfig { @@ -25,6 +109,10 @@ pub struct ArchiveStorageConfig { pub max_file_bytes: u64, pub retain_files: usize, pub file_stem: String, + pub format: ArchiveFormat, + pub mode: ArchiveMode, + pub window_secs: u64, + pub compression: ArchiveCompression, } impl ArchiveStorageConfig { @@ -35,24 +123,129 @@ impl ArchiveStorageConfig { max_file_bytes: 0, retain_files: 0, file_stem: default_stem.to_string(), + format: ArchiveFormat::Parquet, + mode: ArchiveMode::Trend, + window_secs: DEFAULT_ARCHIVE_WINDOW_SECS, + compression: ArchiveCompression::Zstd, + } + } +} + +pub trait ArchiveWriter { + fn write_snapshot(&mut self, value: &Value); + fn write_log_batch(&mut self, values: &[Value]); + fn is_healthy(&self) -> bool; + fn total_records(&self) -> u64; + fn total_bytes(&self) -> u64; + fn last_error(&self) -> Option<&str>; +} + +#[derive(Clone, Debug)] +struct TrendStats { + min: f64, + max: f64, + sum: f64, + count: u64, + first: f64, + last: f64, +} + +impl TrendStats { + fn new(value: f64) -> Self { + Self { + min: value, + max: value, + sum: value, + count: 1, + first: value, + last: value, + } + } + + fn update(&mut self, value: f64) { + if value < self.min { + self.min = value; } + if value > self.max { + self.max = value; + } + self.sum += value; + self.count = self.count.saturating_add(1); + self.last = value; + } + + fn avg(&self) -> f64 { + if self.count == 0 { + 0.0 + } else { + self.sum / self.count as f64 + } + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +struct TrendKey { + service_name: String, + instance_id: String, + metric_key: String, +} + +impl Hash for TrendKey { + fn hash(&self, state: &mut H) { + self.service_name.hash(state); + self.instance_id.hash(state); + self.metric_key.hash(state); } } #[derive(Clone, Debug)] -pub struct JsonArchiveWriter { +struct TrendRow { + window_start_unix_secs: i64, + service_name: String, + instance_id: String, + metric_key: String, + min: f64, + max: f64, + avg: f64, + count: u64, + first: f64, + last: f64, + written_at_unix_ms: i64, +} + +#[derive(Clone, Debug)] +struct ForensicRow { + observed_unix_ms: i64, + service_name: String, + instance_id: String, + record_type: String, + severity: Option, + source: Option, + watch_target: Option, + message_signature: Option, + sample_message: Option, + payload_json: String, +} + +pub struct ParquetArchiveWriter { enabled: bool, dir: String, max_file_bytes: u64, retain_files: usize, file_stem: String, + mode: ArchiveMode, + window_secs: u64, + compression: ArchiveCompression, pub total_records: u64, pub total_bytes: u64, pub healthy: bool, pub last_error: Option, + default_service_name: String, + default_instance_id: String, + trend_windows: BTreeMap>, } -impl JsonArchiveWriter { +impl ParquetArchiveWriter { pub fn from_config(config: &ArchiveStorageConfig) -> Self { Self { enabled: config.enabled, @@ -60,40 +253,466 @@ impl JsonArchiveWriter { max_file_bytes: config.max_file_bytes, retain_files: config.retain_files, file_stem: config.file_stem.clone(), + mode: config.mode.clone(), + window_secs: config.window_secs.max(1), + compression: config.compression.clone(), total_records: 0, total_bytes: 0, healthy: true, last_error: None, + default_service_name: String::new(), + default_instance_id: String::new(), + trend_windows: BTreeMap::new(), } } + pub fn set_default_identity(&mut self, service_name: &str, instance_id: &str) { + self.default_service_name = service_name.to_string(); + self.default_instance_id = instance_id.to_string(); + } + pub fn write_json_line(&mut self, value: &Value) { + self.write_snapshot(value); + } + + pub fn flush(&mut self) { if !self.enabled { return; } - if let Err(err) = self.write_json_line_impl(value) { - self.healthy = false; - self.last_error = Some(err.to_string()); - } else { - self.healthy = true; - self.last_error = None; + if let ArchiveMode::Trend = self.mode { + if let Err(err) = self.flush_all_trend_windows() { + self.healthy = false; + self.last_error = Some(err.to_string()); + } } } - fn write_json_line_impl(&mut self, value: &Value) -> anyhow::Result<()> { + fn write_snapshot_impl(&mut self, value: &Value) -> anyhow::Result<()> { fs::create_dir_all(&self.dir)?; - let path = format!("{}/{}.ndjson", self.dir, self.file_stem); - self.rotate_if_needed(&path)?; - - let mut file = OpenOptions::new().create(true).append(true).open(&path)?; - let line = value.to_string(); - file.write_all(line.as_bytes())?; - file.write_all(b"\n")?; self.total_records = self.total_records.saturating_add(1); self.total_bytes = self .total_bytes - .saturating_add((line.len() as u64).saturating_add(1)); + .saturating_add((value.to_string().len() as u64).saturating_add(1)); + + match self.mode { + ArchiveMode::Trend => self.ingest_snapshot_to_trend(value), + ArchiveMode::Forensic | ArchiveMode::Lossless => { + self.write_forensic_rows(vec![self.build_forensic_row("snapshot", value)]) + } + } + } + + fn write_log_batch_impl(&mut self, values: &[Value]) -> anyhow::Result<()> { + if values.is_empty() { + return Ok(()); + } + + fs::create_dir_all(&self.dir)?; + self.total_records = self.total_records.saturating_add(values.len() as u64); + self.total_bytes = self.total_bytes.saturating_add( + values + .iter() + .map(|v| v.to_string().len() as u64 + 1) + .sum::(), + ); + + match self.mode { + ArchiveMode::Trend => self.ingest_logs_to_trend(values), + ArchiveMode::Forensic | ArchiveMode::Lossless => { + let rows = values + .iter() + .map(|v| self.build_forensic_row("log", v)) + .collect::>(); + self.write_forensic_rows(rows) + } + } + } + + fn ingest_snapshot_to_trend(&mut self, value: &Value) -> anyhow::Result<()> { + let now_secs = now_unix_secs(); + let window_start = align_window_start(now_secs, self.window_secs); + let (service_name, instance_id) = self.resolve_identity(value); + + if let Some(obj) = value.as_object() { + for (k, v) in obj { + let mut path = vec![k.to_string()]; + self.flatten_numeric(v, &mut path, window_start, &service_name, &instance_id); + } + } + + self.flush_closed_windows(window_start) + } + + fn ingest_logs_to_trend(&mut self, values: &[Value]) -> anyhow::Result<()> { + let now_secs = now_unix_secs(); + let window_start = align_window_start(now_secs, self.window_secs); + + for value in values { + let (service_name, instance_id) = self.resolve_identity(value); + self.add_trend_point( + window_start, + TrendKey { + service_name: service_name.clone(), + instance_id: instance_id.clone(), + metric_key: "log.count".to_string(), + }, + 1.0, + ); + + if let Some(severity) = value.get("severity_text").and_then(Value::as_str) { + self.add_trend_point( + window_start, + TrendKey { + service_name: service_name.clone(), + instance_id: instance_id.clone(), + metric_key: format!("log.severity.{}.count", sanitize_segment(severity)), + }, + 1.0, + ); + } + if let Some(source) = value.get("source").and_then(Value::as_str) { + self.add_trend_point( + window_start, + TrendKey { + service_name: service_name.clone(), + instance_id: instance_id.clone(), + metric_key: format!("log.source.{}.count", sanitize_segment(source)), + }, + 1.0, + ); + } + if let Some(target) = value.get("watch_target").and_then(Value::as_str) { + self.add_trend_point( + window_start, + TrendKey { + service_name: service_name.clone(), + instance_id: instance_id.clone(), + metric_key: format!("log.watch_target.{}.count", sanitize_segment(target)), + }, + 1.0, + ); + } + } + + let mut signatures = HashMap::<(String, String, u64), (u64, String)>::new(); + for value in values { + if let Some(body) = value.get("body").and_then(Value::as_str) { + let (service_name, instance_id) = self.resolve_identity(value); + let sig = stable_signature(body); + let entry = signatures + .entry((service_name, instance_id, sig)) + .or_insert((0, body.to_string())); + entry.0 = entry.0.saturating_add(1); + } + } + + let mut ranked = signatures.into_iter().collect::>(); + ranked.sort_by(|a, b| b.1 .0.cmp(&a.1 .0)); + for ((service_name, instance_id, sig), (count, _sample)) in + ranked.into_iter().take(DEFAULT_TOP_K_LOG_SIGNATURES) + { + let metric_key = format!("log.topk.signature.{}.count", sig); + self.add_trend_point( + window_start, + TrendKey { + service_name, + instance_id, + metric_key, + }, + count as f64, + ); + } + + self.flush_closed_windows(window_start) + } + fn flatten_numeric( + &mut self, + value: &Value, + path: &mut Vec, + window_start: i64, + service_name: &str, + instance_id: &str, + ) { + match value { + Value::Number(num) => { + if let Some(v) = num.as_f64() { + if v.is_finite() { + self.add_trend_point( + window_start, + TrendKey { + service_name: service_name.to_string(), + instance_id: instance_id.to_string(), + metric_key: path.join("."), + }, + v, + ); + } + } + } + Value::Object(map) => { + for (k, v) in map { + path.push(k.to_string()); + self.flatten_numeric(v, path, window_start, service_name, instance_id); + path.pop(); + } + } + Value::Array(values) => { + for (idx, v) in values.iter().enumerate() { + path.push(idx.to_string()); + self.flatten_numeric(v, path, window_start, service_name, instance_id); + path.pop(); + } + } + _ => {} + } + } + + fn add_trend_point(&mut self, window_start: i64, key: TrendKey, value: f64) { + let map = self.trend_windows.entry(window_start).or_default(); + let entry = map.entry(key).or_insert_with(|| TrendStats::new(value)); + if entry.count > 0 { + entry.update(value); + } + } + + fn flush_closed_windows(&mut self, current_window: i64) -> anyhow::Result<()> { + let closed = self + .trend_windows + .keys() + .copied() + .filter(|w| *w < current_window) + .collect::>(); + for window in closed { + if let Some(values) = self.trend_windows.remove(&window) { + self.write_trend_window(window, values)?; + } + } + Ok(()) + } + + fn flush_all_trend_windows(&mut self) -> anyhow::Result<()> { + let windows = self.trend_windows.keys().copied().collect::>(); + for window in windows { + if let Some(values) = self.trend_windows.remove(&window) { + self.write_trend_window(window, values)?; + } + } + Ok(()) + } + + fn write_trend_window( + &mut self, + window_start: i64, + values: HashMap, + ) -> anyhow::Result<()> { + if values.is_empty() { + return Ok(()); + } + + let written_at = now_unix_millis(); + let rows = values + .into_iter() + .map(|(key, stats)| TrendRow { + window_start_unix_secs: window_start, + service_name: key.service_name, + instance_id: key.instance_id, + metric_key: key.metric_key, + min: stats.min, + max: stats.max, + avg: stats.avg(), + count: stats.count, + first: stats.first, + last: stats.last, + written_at_unix_ms: written_at, + }) + .collect::>(); + + self.write_trend_rows(rows) + } + + fn write_trend_rows(&mut self, rows: Vec) -> anyhow::Result<()> { + let schema = trend_schema(); + + let columns: Vec = vec![ + Arc::new(Int64Array::from( + rows.iter() + .map(|r| r.window_start_unix_secs) + .collect::>(), + )), + Arc::new(StringArray::from( + rows.iter() + .map(|r| r.service_name.as_str()) + .collect::>(), + )), + Arc::new(StringArray::from( + rows.iter() + .map(|r| r.instance_id.as_str()) + .collect::>(), + )), + Arc::new(StringArray::from( + rows.iter() + .map(|r| r.metric_key.as_str()) + .collect::>(), + )), + Arc::new(Float64Array::from( + rows.iter().map(|r| r.min).collect::>(), + )), + Arc::new(Float64Array::from( + rows.iter().map(|r| r.max).collect::>(), + )), + Arc::new(Float64Array::from( + rows.iter().map(|r| r.avg).collect::>(), + )), + Arc::new(UInt64Array::from( + rows.iter().map(|r| r.count).collect::>(), + )), + Arc::new(Float64Array::from( + rows.iter().map(|r| r.first).collect::>(), + )), + Arc::new(Float64Array::from( + rows.iter().map(|r| r.last).collect::>(), + )), + Arc::new(StringArray::from(vec![ARCHIVE_SCHEMA_VERSION; rows.len()])), + Arc::new(StringArray::from(vec![self.mode.as_str(); rows.len()])), + Arc::new(Int64Array::from( + rows.iter() + .map(|r| r.written_at_unix_ms) + .collect::>(), + )), + ]; + + let batch = RecordBatch::try_new(schema, columns)?; + self.write_batch(batch, "trend") + } + + fn build_forensic_row(&self, record_type: &str, value: &Value) -> ForensicRow { + let (service_name, instance_id) = self.resolve_identity(value); + let body = value + .get("body") + .and_then(Value::as_str) + .map(ToOwned::to_owned); + + ForensicRow { + observed_unix_ms: now_unix_millis(), + service_name, + instance_id, + record_type: record_type.to_string(), + severity: value + .get("severity_text") + .and_then(Value::as_str) + .map(ToOwned::to_owned), + source: value + .get("source") + .and_then(Value::as_str) + .map(ToOwned::to_owned), + watch_target: value + .get("watch_target") + .and_then(Value::as_str) + .map(ToOwned::to_owned), + message_signature: body.as_ref().map(|s| stable_signature(s)), + sample_message: body, + payload_json: value.to_string(), + } + } + + fn write_forensic_rows(&mut self, rows: Vec) -> anyhow::Result<()> { + if rows.is_empty() { + return Ok(()); + } + let schema = forensic_schema(); + + let columns: Vec = vec![ + Arc::new(Int64Array::from( + rows.iter().map(|r| r.observed_unix_ms).collect::>(), + )), + Arc::new(StringArray::from( + rows.iter() + .map(|r| r.service_name.as_str()) + .collect::>(), + )), + Arc::new(StringArray::from( + rows.iter() + .map(|r| r.instance_id.as_str()) + .collect::>(), + )), + Arc::new(StringArray::from( + rows.iter() + .map(|r| r.record_type.as_str()) + .collect::>(), + )), + Arc::new(StringArray::from( + rows.iter() + .map(|r| r.severity.as_deref()) + .collect::>>(), + )), + Arc::new(StringArray::from( + rows.iter() + .map(|r| r.source.as_deref()) + .collect::>>(), + )), + Arc::new(StringArray::from( + rows.iter() + .map(|r| r.watch_target.as_deref()) + .collect::>>(), + )), + Arc::new(UInt64Array::from( + rows.iter() + .map(|r| r.message_signature) + .collect::>>(), + )), + Arc::new(StringArray::from( + rows.iter() + .map(|r| r.sample_message.as_deref()) + .collect::>>(), + )), + Arc::new(StringArray::from( + rows.iter() + .map(|r| r.payload_json.as_str()) + .collect::>(), + )), + Arc::new(StringArray::from(vec![ARCHIVE_SCHEMA_VERSION; rows.len()])), + Arc::new(StringArray::from(vec![self.mode.as_str(); rows.len()])), + ]; + + let batch = RecordBatch::try_new(schema, columns)?; + let suffix = match self.mode { + ArchiveMode::Lossless => "lossless", + _ => "forensic", + }; + self.write_batch(batch, suffix) + } + + fn resolve_identity(&self, value: &Value) -> (String, String) { + let (service_name, instance_id) = extract_identity(value); + let resolved_service = if service_name.is_empty() { + self.default_service_name.clone() + } else { + service_name + }; + let resolved_instance = if instance_id.is_empty() { + self.default_instance_id.clone() + } else { + instance_id + }; + (resolved_service, resolved_instance) + } + + fn write_batch(&mut self, batch: RecordBatch, suffix: &str) -> anyhow::Result<()> { + let path = format!("{}/{}-{}.parquet", self.dir, self.file_stem, suffix); + self.rotate_if_needed(&path)?; + + let mut metadata_builder = WriterProperties::builder(); + metadata_builder = metadata_builder + .set_compression(self.compression.to_parquet()) + .set_created_by("ojo-archive".to_string()); + + let props = metadata_builder.build(); + let file = File::create(&path)?; + let mut writer = ArrowWriter::try_new(file, batch.schema(), Some(props))?; + writer.write(&batch)?; + writer.flush()?; + writer.finish()?; self.prune_rotated_files(&path)?; Ok(()) } @@ -106,6 +725,7 @@ impl JsonArchiveWriter { if metadata.len() < self.max_file_bytes { return Ok(()); } + for idx in (1..=self.retain_files).rev() { let src = format!("{}.{}", path, idx); let dst = format!("{}.{}", path, idx + 1); @@ -134,6 +754,164 @@ impl JsonArchiveWriter { } } +impl ArchiveWriter for ParquetArchiveWriter { + fn write_snapshot(&mut self, value: &Value) { + if !self.enabled { + return; + } + if let Err(err) = self.write_snapshot_impl(value) { + self.healthy = false; + self.last_error = Some(err.to_string()); + } else { + self.healthy = true; + self.last_error = None; + } + } + + fn write_log_batch(&mut self, values: &[Value]) { + if !self.enabled { + return; + } + if let Err(err) = self.write_log_batch_impl(values) { + self.healthy = false; + self.last_error = Some(err.to_string()); + } else { + self.healthy = true; + self.last_error = None; + } + } + + fn is_healthy(&self) -> bool { + self.healthy + } + + fn total_records(&self) -> u64 { + self.total_records + } + + fn total_bytes(&self) -> u64 { + self.total_bytes + } + + fn last_error(&self) -> Option<&str> { + self.last_error.as_deref() + } +} + +impl Drop for ParquetArchiveWriter { + fn drop(&mut self) { + self.flush(); + } +} + +pub type JsonArchiveWriter = ParquetArchiveWriter; + +fn trend_schema() -> Arc { + Arc::new(Schema::new(vec![ + Field::new("window_start_unix_secs", DataType::Int64, false), + Field::new("service_name", DataType::Utf8, false), + Field::new("instance_id", DataType::Utf8, false), + Field::new("metric_key", DataType::Utf8, false), + Field::new("min", DataType::Float64, false), + Field::new("max", DataType::Float64, false), + Field::new("avg", DataType::Float64, false), + Field::new("count", DataType::UInt64, false), + Field::new("first", DataType::Float64, false), + Field::new("last", DataType::Float64, false), + Field::new("archive_schema_version", DataType::Utf8, false), + Field::new("archive_mode", DataType::Utf8, false), + Field::new("written_at_unix_ms", DataType::Int64, false), + ])) +} + +fn forensic_schema() -> Arc { + Arc::new(Schema::new(vec![ + Field::new("observed_unix_ms", DataType::Int64, false), + Field::new("service_name", DataType::Utf8, false), + Field::new("instance_id", DataType::Utf8, false), + Field::new("record_type", DataType::Utf8, false), + Field::new("severity", DataType::Utf8, true), + Field::new("source", DataType::Utf8, true), + Field::new("watch_target", DataType::Utf8, true), + Field::new("message_signature", DataType::UInt64, true), + Field::new("sample_message", DataType::Utf8, true), + Field::new("payload_json", DataType::Utf8, false), + Field::new("archive_schema_version", DataType::Utf8, false), + Field::new("archive_mode", DataType::Utf8, false), + ])) +} + +fn sanitize_segment(raw: &str) -> String { + let lowered = raw.to_ascii_lowercase(); + let cleaned = lowered + .chars() + .map(|ch| { + if ch.is_ascii_alphanumeric() || ch == '_' || ch == '-' { + ch + } else { + '_' + } + }) + .collect::(); + + cleaned.trim_matches('_').to_string() +} + +fn stable_signature(value: &str) -> u64 { + let mut hasher = std::collections::hash_map::DefaultHasher::new(); + value.hash(&mut hasher); + hasher.finish() +} + +fn now_unix_secs() -> i64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_secs() as i64 +} + +fn now_unix_millis() -> i64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as i64 +} + +fn align_window_start(timestamp_secs: i64, window_secs: u64) -> i64 { + let ws = window_secs.max(1) as i64; + (timestamp_secs / ws) * ws +} + +fn extract_identity(value: &Value) -> (String, String) { + let service_name = value + .get("service_name") + .and_then(Value::as_str) + .or_else(|| { + value + .get("service") + .and_then(Value::as_object) + .and_then(|m| m.get("name")) + .and_then(Value::as_str) + }) + .unwrap_or_default() + .to_string(); + + let instance_id = value + .get("instance_id") + .and_then(Value::as_str) + .or_else(|| { + value + .get("service") + .and_then(Value::as_object) + .and_then(|m| m.get("instance_id")) + .and_then(Value::as_str) + }) + .unwrap_or_default() + .to_string(); + + (service_name, instance_id) +} + #[derive(Clone, Debug)] pub struct PrefixFilter { include: Vec, diff --git a/crates/host-collectors/src/tests.rs b/crates/host-collectors/src/tests.rs index 006196b..278ecb5 100644 --- a/crates/host-collectors/src/tests.rs +++ b/crates/host-collectors/src/tests.rs @@ -1,8 +1,10 @@ use super::{ build_meter_provider, build_tracer_provider, default_protocol_for_endpoint, hostname, - init_meter_provider, init_tracer_provider, ArchiveStorageConfig, JsonArchiveWriter, - OtlpSettings, PrefixFilter, StdoutSpanExporter, METRIC_PREFIX_SYSTEM, + init_meter_provider, init_tracer_provider, ArchiveCompression, ArchiveFormat, ArchiveMode, + ArchiveStorageConfig, JsonArchiveWriter, OtlpSettings, PrefixFilter, StdoutSpanExporter, + METRIC_PREFIX_SYSTEM, }; +use arrow_array::{Array, StringArray}; use opentelemetry::trace::{ SpanContext, SpanId, SpanKind, Status, TraceFlags, TraceId, TraceState, }; @@ -11,8 +13,10 @@ use opentelemetry::KeyValue; use opentelemetry_sdk::error::OTelSdkError; use opentelemetry_sdk::resource::Resource; use opentelemetry_sdk::trace::{SpanData, SpanEvents, SpanExporter, SpanLinks}; +use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder; use std::collections::BTreeMap; use std::fs; +use std::fs::File; use std::path::PathBuf; use std::time::Duration; use std::time::{SystemTime, UNIX_EPOCH}; @@ -121,6 +125,13 @@ fn default_protocol_without_scheme_uses_grpc() { ); } +#[test] +fn archive_mode_parses_lossless_and_forensic() { + assert_eq!(ArchiveMode::parse(Some("lossless")), ArchiveMode::Lossless); + assert_eq!(ArchiveMode::parse(Some("forensic")), ArchiveMode::Forensic); + assert_eq!(ArchiveMode::parse(Some("trend")), ArchiveMode::Trend); +} + #[test] fn build_meter_provider_grpc_succeeds() { let rt = tokio::runtime::Runtime::new().expect("runtime"); @@ -279,6 +290,10 @@ fn json_archive_writer_from_config_copies_fields_and_initializes_state() { max_file_bytes: 128, retain_files: 3, file_stem: "host".to_string(), + format: ArchiveFormat::Parquet, + mode: ArchiveMode::Forensic, + window_secs: 60, + compression: ArchiveCompression::Zstd, }; let writer = JsonArchiveWriter::from_config(&config); @@ -312,6 +327,10 @@ fn json_archive_writer_writes_and_tracks_bytes() { max_file_bytes: 1024, retain_files: 0, file_stem: "metrics".to_string(), + format: ArchiveFormat::Parquet, + mode: ArchiveMode::Trend, + window_secs: 60, + compression: ArchiveCompression::Zstd, }; let mut writer = JsonArchiveWriter::from_config(&config); let payload = serde_json::json!({"name": "cpu", "value": 1}); @@ -328,6 +347,56 @@ fn json_archive_writer_writes_and_tracks_bytes() { fs::remove_dir_all(&dir).expect("remove temp directory"); } +#[test] +fn json_archive_writer_uses_default_identity_when_missing_in_payload() { + let dir = unique_temp_dir("host_collectors_archive_identity_defaults"); + fs::create_dir_all(&dir).expect("create temp directory"); + + let config = ArchiveStorageConfig { + enabled: true, + archive_dir: dir.to_string_lossy().into_owned(), + max_file_bytes: 1024 * 1024, + retain_files: 0, + file_stem: "metrics".to_string(), + format: ArchiveFormat::Parquet, + mode: ArchiveMode::Trend, + window_secs: 1, + compression: ArchiveCompression::Zstd, + }; + let mut writer = JsonArchiveWriter::from_config(&config); + writer.set_default_identity("svc-default", "inst-default"); + writer.write_json_line(&serde_json::json!({"metric": {"value": 42}})); + writer.flush(); + + let path = dir.join("metrics-trend.parquet"); + let file = File::open(path).expect("open archive parquet"); + let mut reader = ParquetRecordBatchReaderBuilder::try_new(file) + .expect("build parquet reader") + .build() + .expect("create record batch reader"); + let batch = reader + .next() + .expect("at least one record batch") + .expect("read batch"); + let service_col = batch + .column_by_name("service_name") + .expect("service_name column") + .as_any() + .downcast_ref::() + .expect("service_name should be utf8"); + let instance_col = batch + .column_by_name("instance_id") + .expect("instance_id column") + .as_any() + .downcast_ref::() + .expect("instance_id should be utf8"); + + assert_eq!(service_col.value(0), "svc-default"); + assert_eq!(instance_col.value(0), "inst-default"); + + fs::remove_dir_all(&dir).expect("remove temp directory"); +} + #[test] fn json_archive_writer_sets_health_false_when_archive_dir_is_invalid() { let dir = unique_temp_dir("host_collectors_archive_err"); @@ -341,6 +410,10 @@ fn json_archive_writer_sets_health_false_when_archive_dir_is_invalid() { max_file_bytes: 1024, retain_files: 1, file_stem: "metrics".to_string(), + format: ArchiveFormat::Parquet, + mode: ArchiveMode::Trend, + window_secs: 60, + compression: ArchiveCompression::Zstd, }; let mut writer = JsonArchiveWriter::from_config(&config); writer.write_json_line(&serde_json::json!({"name": "cpu"})); @@ -357,7 +430,7 @@ fn json_archive_writer_sets_health_false_when_archive_dir_is_invalid() { fn json_archive_writer_rotates_and_prunes_files() { let dir = unique_temp_dir("host_collectors_archive_rotate"); fs::create_dir_all(&dir).expect("create temp directory"); - let base = dir.join("metrics.ndjson"); + let base = dir.join("metrics-forensic.parquet"); fs::write(&base, vec![b'x'; 20]).expect("seed oversized current file"); fs::write(format!("{}.1", base.to_string_lossy()), b"older-1").expect("seed rotated file 1"); fs::write(format!("{}.2", base.to_string_lossy()), b"older-2").expect("seed rotated file 2"); @@ -369,6 +442,10 @@ fn json_archive_writer_rotates_and_prunes_files() { max_file_bytes: 10, retain_files: 2, file_stem: "metrics".to_string(), + format: ArchiveFormat::Parquet, + mode: ArchiveMode::Trend, + window_secs: 60, + compression: ArchiveCompression::Zstd, }; let mut writer = JsonArchiveWriter::from_config(&config); writer.write_json_line(&serde_json::json!({"rotated": true})); @@ -376,8 +453,6 @@ fn json_archive_writer_rotates_and_prunes_files() { assert!(base.exists()); assert!(PathBuf::from(format!("{}.1", base.to_string_lossy())).exists()); assert!(PathBuf::from(format!("{}.2", base.to_string_lossy())).exists()); - assert!(PathBuf::from(format!("{}.3", base.to_string_lossy())).exists()); - assert!(!PathBuf::from(format!("{}.4", base.to_string_lossy())).exists()); fs::remove_dir_all(&dir).expect("remove temp directory"); } diff --git a/src/bin/archive-replay.rs b/src/bin/archive-replay.rs new file mode 100644 index 0000000..df74390 --- /dev/null +++ b/src/bin/archive-replay.rs @@ -0,0 +1,735 @@ +use anyhow::{anyhow, Context, Result}; +use arrow_array::{Array, Float64Array, Int64Array, RecordBatch, StringArray}; +use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder; +use prost::Message; +use reqwest::blocking::Client; +use serde_json::Value; +use snap::raw::Encoder; +use std::collections::HashMap; +use std::env; +use std::fs::{self, File}; +use std::path::{Path, PathBuf}; + +#[derive(Clone, Debug)] +struct ReplayPoint { + timestamp_ms: i64, + service_name: String, + instance_id: String, + metric_key: String, + aggregation: String, + value: f64, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum Protocol { + Otlp, + RemoteWrite, + ShellLogs, +} + +#[derive(Clone, Debug)] +struct Config { + archive_dir: PathBuf, + endpoint: Option, + protocol: Protocol, + batch_size: usize, + service_name: String, +} + +type SeriesKey = (String, String, String, String); +type SamplePoint = (i64, f64); + +#[derive(Clone, PartialEq, Message)] +struct WriteRequest { + #[prost(message, repeated, tag = "1")] + timeseries: Vec, +} + +#[derive(Clone, PartialEq, Message)] +struct TimeSeries { + #[prost(message, repeated, tag = "1")] + labels: Vec