-
Notifications
You must be signed in to change notification settings - Fork 124
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: Implement ExecutionLog on
fs 🚀 (#1356)
* 🚀 Interface placeholders * feat: introduce log sink * api implementation (#3) * feat: implement log reader (#5) * feat: implement log reader * feat: implement write schemas (#6) --------- Co-authored-by: VG <[email protected]> Co-authored-by: VG <[email protected]> * fix compilation * feat: write schemas properly * temp: schema and proto bug fix * path bug fix * file not created * chore: fix log reader * chore: allow cache_dir config * chore: include args in example * chore: fix cache_dir * feat: mofiy schema * chore: fix clippy * chore: fix fmt * chore: fix admin test * chore: ignore conflict resolution tests * chore: add buffered capacity --------- Co-authored-by: VG <[email protected]> Co-authored-by: VG <[email protected]>
- Loading branch information
3 people
authored
Apr 3, 2023
1 parent
ea8217c
commit a4f8746
Showing
44 changed files
with
1,067 additions
and
1,017 deletions.
There are no files selected for viewing
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,26 @@ | ||
| use std::{env, path::Path}; | ||
|
|
||
| use dozer_api::LogReader; | ||
| use futures_util::StreamExt; | ||
|
|
||
| #[tokio::main] | ||
| async fn main() { | ||
| let args: Vec<String> = env::args().collect(); | ||
|
|
||
| let mut path = ".dozer/pipeline/logs/trips"; | ||
| if args.len() == 2 { | ||
| path = &args[1]; | ||
| }; | ||
| let log_reader = LogReader::new(Path::new(path), "logs", 0, None).unwrap(); | ||
|
|
||
| tokio::pin!(log_reader); | ||
|
|
||
| let mut counter = 0; | ||
| while let Some(_op) = log_reader.next().await { | ||
| counter += 1; | ||
|
|
||
| if counter > 100000 { | ||
| break; | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,135 @@ | ||
| use std::{ | ||
| fs::{File, OpenOptions}, | ||
| io::{BufReader, Read}, | ||
| path::Path, | ||
| thread::sleep, | ||
| time::Duration, | ||
| }; | ||
|
|
||
| use dozer_cache::errors::{CacheError, LogError}; | ||
| use dozer_core::executor::ExecutorOperation; | ||
| use dozer_types::{ | ||
| bincode, | ||
| indicatif::{MultiProgress, ProgressBar, ProgressStyle}, | ||
| log::trace, | ||
| }; | ||
| use futures_util::{FutureExt, Stream}; | ||
|
|
||
| pub struct LogReader { | ||
| reader: BufReader<File>, | ||
| name: String, | ||
| pos: u64, | ||
| pb: ProgressBar, | ||
| count: u64, | ||
| } | ||
| const SLEEP_TIME_MS: u16 = 300; | ||
| impl LogReader { | ||
| pub fn new( | ||
| path: &Path, | ||
| name: &str, | ||
| pos: u64, | ||
| multi_pb: Option<MultiProgress>, | ||
| ) -> Result<Self, CacheError> { | ||
| let file = OpenOptions::new() | ||
| .read(true) | ||
| .open(path) | ||
| .map_err(|_| CacheError::LogFileNotFound(path.to_path_buf()))?; | ||
|
|
||
| let mut reader = BufReader::new(file); | ||
|
|
||
| reader | ||
| .seek_relative(pos as i64) | ||
| .map_err(|e| CacheError::LogError(LogError::SeekError(name.to_string(), pos, e)))?; | ||
|
|
||
| let pb = attach_progress(multi_pb); | ||
| pb.set_message(format!("reader: {}", name)); | ||
| Ok(Self { | ||
| reader, | ||
| name: name.to_string(), | ||
| pos, | ||
| pb, | ||
| count: 0, | ||
| }) | ||
| } | ||
| async fn next_op(&mut self) -> ExecutorOperation { | ||
| loop { | ||
| let msg = read_msg(&mut self.reader); | ||
| match msg { | ||
| Ok((msg, len)) => { | ||
| self.pos += len; | ||
| self.count += 1; | ||
| self.pb.set_position(self.count); | ||
| return msg; | ||
| } | ||
| Err(e) => { | ||
| trace!( | ||
| "Error reading log : {}, Going to sleep : {} ms, Error : {:?}", | ||
| self.name, | ||
| SLEEP_TIME_MS, | ||
| e | ||
| ); | ||
|
|
||
| // go to sleep for a bit | ||
| sleep(Duration::from_millis(SLEEP_TIME_MS.into())); | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| impl Stream for LogReader { | ||
| type Item = ExecutorOperation; | ||
|
|
||
| fn poll_next( | ||
| self: std::pin::Pin<&mut Self>, | ||
| cx: &mut std::task::Context<'_>, | ||
| ) -> std::task::Poll<Option<Self::Item>> { | ||
| let this = self.get_mut(); | ||
| match Box::pin(this.next_op()).poll_unpin(cx) { | ||
| std::task::Poll::Ready(msg) => std::task::Poll::Ready(Some(msg)), | ||
| std::task::Poll::Pending => std::task::Poll::Pending, | ||
| } | ||
| } | ||
| } | ||
|
|
||
| fn read_msg(reader: &mut BufReader<File>) -> Result<(ExecutorOperation, u64), LogError> { | ||
| let mut buf = [0; 8]; | ||
| reader.read_exact(&mut buf).map_err(LogError::ReadError)?; | ||
| let len = u64::from_le_bytes(buf); | ||
|
|
||
| let buf = read_n(reader, len); | ||
| let msg = bincode::deserialize(&buf).map_err(LogError::DeserializationError)?; | ||
| Ok((msg, len + 8)) | ||
| } | ||
| fn read_n<R>(reader: R, bytes_to_read: u64) -> Vec<u8> | ||
| where | ||
| R: Read, | ||
| { | ||
| let mut buf = vec![]; | ||
| let mut chunk = reader.take(bytes_to_read); | ||
|
|
||
| let n = chunk.read_to_end(&mut buf).expect("Didn't read enough"); | ||
| assert_eq!(bytes_to_read as usize, n); | ||
| buf | ||
| } | ||
|
|
||
| fn attach_progress(multi_pb: Option<MultiProgress>) -> ProgressBar { | ||
| let pb = ProgressBar::new_spinner(); | ||
| multi_pb.as_ref().map(|m| m.add(pb.clone())); | ||
| pb.set_style( | ||
| ProgressStyle::with_template("{spinner:.blue} {msg}: {pos}: {per_sec}") | ||
| .unwrap() | ||
| // For more spinners check out the cli-spinners project: | ||
| // https://github.com/sindresorhus/cli-spinners/blob/master/spinners.json | ||
| .tick_strings(&[ | ||
| "▹▹▹▹▹", | ||
| "▸▹▹▹▹", | ||
| "▹▸▹▹▹", | ||
| "▹▹▸▹▹", | ||
| "▹▹▹▸▹", | ||
| "▹▹▹▹▸", | ||
| "▪▪▪▪▪", | ||
| ]), | ||
| ); | ||
| pb | ||
| } |
Oops, something went wrong.