Change local cache directory structure (#202)

This PR introduces changes to local directory and removes tmp directory
All conversion is still carried out in s3 sync cycle. StorageDir is used
as primary class for determining what is the data path for a log stream.
RecordBatch file is signified by extension .arrows.

Also remove startup_sync step as it is not required.

Author: trueleo

server/Cargo.toml

@@ -25,7 +25,6 @@ object_store = { version = "0.5.1", features = ["aws"] }
 derive_more = "0.99.17"
 env_logger = "0.9.0"
 futures = "0.3"
-filetime = "0.2.17"
 http = "0.2.4"
 humantime-serde = "1.1.1"
 lazy_static = "1.4.0"
@@ -52,7 +51,6 @@ tokio = { version = "1.13.1", default-features = false, features = [
 clokwerk = "0.4.0-rc1"
 actix-web-static-files = "4.0"
 static-files = "0.2.1"
-walkdir = "2"
 ureq = { version = "2.5.0", features = ["json"] }
 uuid = { version = "1.2.1", features = ["v4", "fast-rng", "serde"] }
 

server/src/event.rs

@@ -17,7 +17,6 @@
  *
  */
 use actix_web::rt::spawn;
-use datafusion::arrow;
 use datafusion::arrow::datatypes::Schema;
 use datafusion::arrow::error::ArrowError;
 use datafusion::arrow::ipc::writer::StreamWriter;
@@ -36,11 +35,10 @@ use std::sync::RwLock;
 
 use crate::metadata;
 use crate::metadata::LOCK_EXPECT;
-use crate::option::CONFIG;
 use crate::s3;
-use crate::storage::ObjectStorage;
+use crate::storage::{ObjectStorage, StorageDir};
 
-use self::error::EventError;
+use self::error::{EventError, StreamWriterError};
 
 type LocalWriter = Mutex<Option<StreamWriter<std::fs::File>>>;
 type LocalWriterGuard<'a> = MutexGuard<'a, Option<StreamWriter<std::fs::File>>>;
@@ -91,18 +89,7 @@ impl STREAM_WRITERS {
             .write()
             .map_err(|_| StreamWriterError::RwPoisoned)?;
 
-        let file = OpenOptions::new()
-            .append(true)
-            .create_new(true)
-            .open(data_file_path(&stream))
-            .map_err(StreamWriterError::Io)?;
-
-        let mut stream_writer = StreamWriter::try_new(file, &record.schema())
-            .expect("File and RecordBatch both are checked");
-
-        stream_writer
-            .write(record)
-            .map_err(StreamWriterError::Writer)?;
+        let stream_writer = init_new_stream_writer_file(&stream, record)?;
 
         hashmap_guard.insert(stream, Mutex::new(Some(stream_writer)));
 
@@ -125,63 +112,51 @@ impl STREAM_WRITERS {
         stream: &str,
         record: &RecordBatch,
     ) -> Result<(), StreamWriterError> {
-        let file = OpenOptions::new()
-            .append(true)
-            .create_new(true)
-            .open(data_file_path(stream))
-            .map_err(StreamWriterError::Io)?;
-
-        let mut stream_writer = StreamWriter::try_new(file, &record.schema())
-            .expect("File and RecordBatch both are checked");
-
-        stream_writer
-            .write(record)
-            .map_err(StreamWriterError::Writer)?;
+        let stream_writer = init_new_stream_writer_file(stream, record)?;
 
         writer_guard.replace(stream_writer); // replace the stream writer behind this mutex
 
         Ok(())
     }
 
-    // Unset the entry so that
-    pub fn unset_entry(stream: &str) -> Result<(), StreamWriterError> {
-        let guard = STREAM_WRITERS
+    pub fn unset_all() -> Result<(), StreamWriterError> {
+        let map = STREAM_WRITERS
             .read()
             .map_err(|_| StreamWriterError::RwPoisoned)?;
-        let stream_writer = match guard.get(stream) {
-            Some(writer) => writer,
-            None => return Ok(()),
-        };
-        stream_writer
-            .lock()
-            .map_err(|_| StreamWriterError::MutexPoisoned)?
-            .take();
+
+        for writer in map.values() {
+            if let Some(mut streamwriter) = writer
+                .lock()
+                .map_err(|_| StreamWriterError::MutexPoisoned)?
+                .take()
+            {
+                let _ = streamwriter.finish();
+            }
+        }
 
         Ok(())
     }
 }
 
-#[derive(Debug, thiserror::Error)]
-pub enum StreamWriterError {
-    #[error("Arrow writer failed: {0}")]
-    Writer(arrow::error::ArrowError),
-    #[error("Io Error when creating new file: {0}")]
-    Io(std::io::Error),
-    #[error("RwLock was poisoned")]
-    RwPoisoned,
-    #[error("Mutex was poisoned")]
-    MutexPoisoned,
-}
+fn init_new_stream_writer_file(
+    stream_name: &str,
+    record: &RecordBatch,
+) -> Result<StreamWriter<std::fs::File>, StreamWriterError> {
+    let dir = StorageDir::new(stream_name);
+    let path = dir.path_by_current_time();
+
+    std::fs::create_dir_all(dir.data_path)?;
+
+    let file = OpenOptions::new().create(true).append(true).open(path)?;
+
+    let mut stream_writer = StreamWriter::try_new(file, &record.schema())
+        .expect("File and RecordBatch both are checked");
+
+    stream_writer
+        .write(record)
+        .map_err(StreamWriterError::Writer)?;
 
-fn data_file_path(stream_name: &str) -> String {
-    format!(
-        "{}/{}",
-        CONFIG
-            .parseable
-            .local_stream_data_path(stream_name)
-            .to_string_lossy(),
-        "data.records"
-    )
+    Ok(stream_writer)
 }
 
 #[derive(Clone)]
@@ -309,7 +284,7 @@ impl Event {
         infer_json_schema(&mut buf_reader, None)
     }
 
-    fn get_reader(&self, arrow_schema: arrow::datatypes::Schema) -> json::Reader<&[u8]> {
+    fn get_reader(&self, arrow_schema: Schema) -> json::Reader<&[u8]> {
         json::Reader::new(
             self.body.as_bytes(),
             Arc::new(arrow_schema),
@@ -348,8 +323,6 @@ pub mod error {
     use crate::storage::ObjectStorageError;
     use datafusion::arrow::error::ArrowError;
 
-    use super::StreamWriterError;
-
     #[derive(Debug, thiserror::Error)]
     pub enum EventError {
         #[error("Missing Record from event body")]
@@ -365,4 +338,16 @@ pub mod error {
         #[error("Schema Mismatch: {0}")]
         ObjectStorage(#[from] ObjectStorageError),
     }
+
+    #[derive(Debug, thiserror::Error)]
+    pub enum StreamWriterError {
+        #[error("Arrow writer failed: {0}")]
+        Writer(#[from] ArrowError),
+        #[error("Io Error when creating new file: {0}")]
+        Io(#[from] std::io::Error),
+        #[error("RwLock was poisoned")]
+        RwPoisoned,
+        #[error("Mutex was poisoned")]
+        MutexPoisoned,
+    }
 }

server/src/handlers/logstream.rs

@@ -70,7 +70,7 @@ pub async fn delete(req: HttpRequest) -> HttpResponse {
         )
     }
 
-    let stream_dir = StorageDir::new(stream_name.clone());
+    let stream_dir = StorageDir::new(&stream_name);
     if fs::remove_dir_all(&stream_dir.data_path).is_err() {
         log::warn!(
             "failed to delete local data for stream {}. Clean {} manually",

server/src/main.rs

@@ -22,9 +22,7 @@ use actix_web::{middleware, web, App, HttpServer};
 use actix_web_httpauth::extractors::basic::BasicAuth;
 use actix_web_httpauth::middleware::HttpAuthentication;
 use actix_web_static_files::ResourceFiles;
-use chrono::{DateTime, NaiveDateTime, Timelike, Utc};
 use clokwerk::{AsyncScheduler, Scheduler, TimeUnits};
-use filetime::FileTime;
 use log::warn;
 use rustls::{Certificate, PrivateKey, ServerConfig};
 use rustls_pemfile::{certs, pkcs8_private_keys};
@@ -33,7 +31,7 @@ use thread_priority::{ThreadBuilder, ThreadPriority};
 
 include!(concat!(env!("OUT_DIR"), "/generated.rs"));
 
-use std::fs::{self, File};
+use std::fs::File;
 use std::io::BufReader;
 use std::panic::{catch_unwind, AssertUnwindSafe};
 use std::thread::{self, JoinHandle};
@@ -57,7 +55,7 @@ mod validator;
 
 use option::CONFIG;
 use s3::S3;
-use storage::{ObjectStorage, StorageDir};
+use storage::ObjectStorage;
 
 // Global configurations
 const MAX_EVENT_PAYLOAD_SIZE: usize = 1024000;
@@ -75,10 +73,6 @@ async fn main() -> anyhow::Result<()> {
         warn!("could not populate local metadata. {:?}", e);
     }
 
-    // Move all exiting data.records file to their respective tmp directory
-    // they will be synced to object store on next s3 sync cycle
-    startup_sync();
-
     let (localsync_handler, mut localsync_outbox, localsync_inbox) = run_local_sync();
     let (mut s3sync_handler, mut s3sync_outbox, mut s3sync_inbox) = s3_sync();
 
@@ -108,59 +102,6 @@ async fn main() -> anyhow::Result<()> {
     }
 }
 
-fn startup_sync() {
-    for stream in metadata::STREAM_INFO.list_streams() {
-        let dir = StorageDir::new(stream.clone());
-
-        dir.create_temp_dir()
-            .expect("Could not create temporary directory. Please check if Parseable is running with correct permissions.");
-
-        // if data.records file is not present then skip this stream
-        if !dir.local_data_exists() {
-            continue;
-        }
-
-        // create prefix for this file from its last modified time
-        let path = dir.data_path.join("data.records");
-
-        // metadata.modified gives us system time
-        // This may not work on all platforms
-        let metadata = match fs::metadata(&path) {
-            Ok(meta) => meta,
-            Err(err) => {
-                log::warn!(
-                    "Failed to get file metadata for {} due to {:?}. Skipping!",
-                    path.display(),
-                    err
-                );
-                continue;
-            }
-        };
-
-        let last_modified = FileTime::from_last_modification_time(&metadata);
-        let last_modified = NaiveDateTime::from_timestamp_opt(last_modified.unix_seconds(), 0);
-        let last_modified: DateTime<Utc> = DateTime::from_utc(last_modified.unwrap(), Utc);
-
-        let uri = utils::date_to_prefix(last_modified.date_naive())
-            + &utils::hour_to_prefix(last_modified.hour())
-            + &utils::minute_to_prefix(
-                last_modified.minute(),
-                storage::OBJECT_STORE_DATA_GRANULARITY,
-            )
-            .unwrap();
-        let local_uri = str::replace(&uri, "/", ".");
-        let hostname = utils::hostname_unchecked();
-        let parquet_file_local = format!("{}{}.data.parquet", local_uri, hostname);
-        if let Err(err) = dir.move_local_to_temp(parquet_file_local) {
-            panic!(
-                "Failed to move parquet file at {} to tmp directory due to error {}",
-                path.display(),
-                err
-            )
-        }
-    }
-}
-
 fn s3_sync() -> (JoinHandle<()>, oneshot::Receiver<()>, oneshot::Sender<()>) {
     let (outbox_tx, outbox_rx) = oneshot::channel::<()>();
     let (inbox_tx, inbox_rx) = oneshot::channel::<()>();
@@ -218,7 +159,7 @@ fn run_local_sync() -> (JoinHandle<()>, oneshot::Receiver<()>, oneshot::Sender<(
                 scheduler
                     .every((storage::LOCAL_SYNC_INTERVAL as u32).seconds())
                     .run(move || {
-                        if let Err(e) = S3::new().local_sync() {
+                        if let Err(e) = crate::event::STREAM_WRITERS::unset_all() {
                             warn!("failed to sync local data. {:?}", e);
                         }
                     });