diff --git a/crates/demo/src/main.rs b/crates/demo/src/main.rs index 6a313e6..ee74dce 100644 --- a/crates/demo/src/main.rs +++ b/crates/demo/src/main.rs @@ -1,4 +1,4 @@ -use bogos_binted::{ServiceExt as _, TraceDestination, in_memory_database}; +use bogos_binted::{LogDestination, LogRetriever, ServiceExt as _, in_memory_database}; use std::{pin::Pin, sync::Arc, task::Poll, time::Duration}; use tower::{Service, ServiceExt as _}; use tracing::{info, info_span}; @@ -40,19 +40,29 @@ async fn main() { info!("initialising"); let db = Arc::new(in_memory_database()); - let mut dest = TraceDestination::new(db.clone()); - let mut svc = UppercaseService.route_logs_to("uppercase_service", &mut dest); + let process_task = { + let mut dest = LogDestination::new(db.clone()); + let mut svc = UppercaseService.route_logs_to("uppercase_service", &mut dest); - tokio::spawn(dest.process()); + let process_task = tokio::spawn(dest.process()); - dbg!( - svc.ready() - .await - .unwrap() - .call("hello, world!".to_string()) - .await - .unwrap() - ); + dbg!( + svc.ready() + .await + .unwrap() + .call("hello, world!".to_string()) + .await + .unwrap() + ); - // TODO: consume the produced logs through a nice api (yet to be created) + process_task + }; + + process_task.await.unwrap(); + + let retrieve = LogRetriever::new(db); + dbg!(retrieve.available_sinks()); + for x in retrieve.get_all("uppercase_service") { + dbg!(x.unwrap()); + } } diff --git a/crates/lib/src/destination.rs b/crates/lib/src/destination.rs index 6276ab6..f87549e 100644 --- a/crates/lib/src/destination.rs +++ b/crates/lib/src/destination.rs @@ -20,7 +20,7 @@ use crate::{ const CHANNEL_SIZE: usize = 8; /// Receives finished logs and writes them out. -pub struct TraceDestination { +pub struct LogDestination { receivers: Vec, db: D, } @@ -37,7 +37,7 @@ struct LogSource { recv: Receiver<(Ulid, AggregatorEntry)>, } -impl> TraceDestination { +impl> LogDestination { /// Write logs to the given [`Database`]. This may be a shared reference, so you can query the database in other places. pub fn new(db: D) -> Self { Self { @@ -69,6 +69,7 @@ impl> TraceDestination { let mut selector = futures::prelude::stream::select_all(streams); while let Some((table, payload)) = selector.next().await { + debug!(message = "received logs", ulid = ?payload.0); if let Err(e) = Self::process_one(&*self.db, table, payload) { warn!("error when writing log to database: {e}"); // TODO: dump the whole log so it doesn't get lost entirely? @@ -122,7 +123,7 @@ enum ProcessPayloadError { LogSerialisingError(rmp::decode::ValueReadError), } -impl std::fmt::Debug for TraceDestination { +impl std::fmt::Debug for LogDestination { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("TraceDestination") .field("receivers", &self.receivers) diff --git a/crates/lib/src/lib.rs b/crates/lib/src/lib.rs index e7694ce..7a336a5 100644 --- a/crates/lib/src/lib.rs +++ b/crates/lib/src/lib.rs @@ -6,6 +6,10 @@ mod destination; #[doc(inline)] pub use destination::*; +mod retrieval; +#[doc(inline)] +pub use retrieval::*; + pub mod logs; mod aggregator; diff --git a/crates/lib/src/retrieval.rs b/crates/lib/src/retrieval.rs new file mode 100644 index 0000000..f459156 --- /dev/null +++ b/crates/lib/src/retrieval.rs @@ -0,0 +1,131 @@ +use std::{ + ops::Deref, + time::{Duration, SystemTime}, +}; +use thiserror::Error; + +use redb::{Database, Range, ReadableDatabase, StorageError, TableDefinition, TableHandle}; +use ulid::Ulid; + +use crate::logs::ServiceCallLogs; + +pub struct LogRetriever { + db: D, + table_names: Vec, +} + +impl> LogRetriever { + pub fn new(db: D) -> Self { + let table_names = db + .begin_read() + .unwrap() + .list_tables() + .unwrap() + .map(|x| x.name().to_string()) + .collect::>(); + + Self { db, table_names } + } + + pub fn available_sinks(&self) -> &[String] { + &self.table_names + } + + pub fn get(&self, name: &'static str, id: Id) -> Option { + let definition: TableDefinition<'_, u128, Vec> = TableDefinition::new(name); + + let bytes = self + .db + .begin_read() + .unwrap() + .open_table(definition) + .unwrap() + .get(id.0) + .unwrap()?; + + let logs = rmp_serde::from_slice::(bytes.value().as_slice()).unwrap(); + + Some(logs) + } + + pub fn get_all(&self, name: &'static str) -> EntryIterator<'_> { + let definition = TableDefinition::new(name); + EntryIterator( + self.db + .begin_read() + .unwrap() + .open_table(definition) + .unwrap() + .range(0..u128::MAX) + .unwrap(), + ) + } + + pub fn get_in_time_range( + &self, + name: &'static str, + start: SystemTime, + end: SystemTime, + ) -> EntryIterator<'_> { + let definition = TableDefinition::new(name); + + let start_ulid = { + let timestamp = start + .duration_since(SystemTime::UNIX_EPOCH) + .unwrap_or(Duration::ZERO) + .as_millis(); + let timebits = (timestamp & 0b11111111_11111111_11111111_11111111) as u64; + + Ulid::from_parts(timebits, 0) + }; + + let end_ulid = { + let timestamp = end + .duration_since(SystemTime::UNIX_EPOCH) + .unwrap_or(Duration::ZERO) + .as_millis(); + let timebits = (timestamp & 0b11111111_11111111_11111111_11111111) as u64; + + Ulid::from_parts(timebits, u128::MAX) + }; + + EntryIterator( + self.db + .begin_read() + .unwrap() + .open_table(definition) + .unwrap() + .range(start_ulid.0..=end_ulid.0) + .unwrap(), + ) + } +} + +pub type Id = Ulid; + +pub struct EntryIterator<'a>(Range<'a, u128, Vec>); +impl Iterator for EntryIterator<'_> { + type Item = Result<(Id, ServiceCallLogs), LoadEntryErr>; + + fn next(&mut self) -> Option { + let (key, bytes) = match self.0.next()? { + Ok(x) => x, + Err(e) => return Some(Err(e.into())), + }; + let logs = match rmp_serde::from_slice::(bytes.value().as_slice()) { + Ok(x) => x, + Err(e) => return Some(Err(e.into())), + }; + + Some(Ok((key.value().into(), logs))) + } +} + +#[derive(Debug, Error)] +pub enum LoadEntryErr { + #[error("error accessing database: {0}")] + StorageError(#[from] StorageError), + + #[error("couldn't deserialise database: {0}")] + DeserialiseError(#[from] rmp_serde::decode::Error), +} diff --git a/crates/lib/src/router.rs b/crates/lib/src/router.rs index 9fecf54..91ff506 100644 --- a/crates/lib/src/router.rs +++ b/crates/lib/src/router.rs @@ -4,7 +4,7 @@ use std::{fmt::Debug, marker::PhantomData, ops::Deref, pin::Pin, sync::Arc}; use tower::Service; use tracing::instrument::WithSubscriber; -use crate::{aggregator::Aggregator, destination::TraceDestination}; +use crate::{aggregator::Aggregator, destination::LogDestination}; /// Instruments each call to a service with a subscriber that writes to an aggregator pub struct LogsRouted> { @@ -18,7 +18,7 @@ impl> LogsRouted { pub fn new>( inner: S, slug: &'static str, - dest: &mut TraceDestination, + dest: &mut LogDestination, ) -> Self { Self { inner, @@ -64,7 +64,7 @@ pub trait ServiceExt: Service { fn route_logs_to>( self, slug: &'static str, - dest: &mut TraceDestination, + dest: &mut LogDestination, ) -> LogsRouted where Self: Sized; @@ -78,7 +78,7 @@ where fn route_logs_to>( self, slug: &'static str, - dest: &mut TraceDestination, + dest: &mut LogDestination, ) -> LogsRouted { LogsRouted::new(self, slug, dest) }