Move leftover data.parquet to tmp with valid prefix on server startup (#73)

When server is restarted it should try to sync data on the 
local directory that is not yet synced to s3. Otherwise the 
data will get lost when new event arrives.

Make use of last modified system time to generate a prefix 
and move data.parquet to tmp directory, It will be synced 
to s3 on next s3_sync run.

Fixes #53 

Co-authored-by: Nitish Tiwari <[email protected]>

Author: trueleo

server/Cargo.toml

@@ -28,6 +28,7 @@ object_store = { version = "0.4", features=["aws"] }
 derive_more = "0.99.17"
 env_logger = "0.9.0"
 futures = "0.3"
+filetime = "0.2.17"
 http = "0.2.4"
 lazy_static = "1.4.0"
 log = "0.4.14"

server/src/main.rs

@@ -22,13 +22,17 @@ 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 openssl::ssl::{SslAcceptor, SslFiletype, SslMethod};
 
 include!(concat!(env!("OUT_DIR"), "/generated.rs"));
 
+use std::fs;
 use std::panic::{catch_unwind, AssertUnwindSafe};
+use std::path::Path;
 use std::thread::{self, JoinHandle};
 use std::time::Duration;
 use tokio::sync::oneshot;
@@ -51,7 +55,7 @@ mod validator;
 use error::Error;
 use option::CONFIG;
 use s3::S3;
-use storage::ObjectStorage;
+use storage::{ObjectStorage, StorageDir};
 
 // Global configurations
 const MAX_EVENT_PAYLOAD_SIZE: usize = 102400;
@@ -69,6 +73,10 @@ async fn main() -> anyhow::Result<()> {
         warn!("could not populate local metadata. {:?}", e);
     }
 
+    // Move all exiting data.parquet 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();
 
@@ -98,6 +106,66 @@ async fn main() -> anyhow::Result<()> {
     }
 }
 
+fn startup_sync() {
+    if !Path::new(&CONFIG.parseable.local_disk_path).exists() {
+        return;
+    }
+
+    for stream in metadata::STREAM_INFO.list_streams() {
+        let dir = StorageDir::new(&stream);
+        // if data.parquet file is not present then skip this stream
+        if !dir.parquet_path_exists() {
+            continue;
+        }
+        if let Err(e) = dir.create_temp_dir() {
+            log::error!(
+                "Error creating tmp directory for {} due to error [{}]",
+                &stream,
+                e
+            );
+            continue;
+        }
+        // create prefix for this file from its last modified time
+        let path = dir.data_path.join("data.parquet");
+
+        // metadata.modified gives us system time
+        // This may not work on all platfomns
+        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(last_modified.unix_seconds(), 0);
+        let last_modified: DateTime<Utc> = DateTime::from_utc(last_modified, Utc);
+
+        let uri = utils::date_to_prefix(last_modified.date())
+            + &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_parquet_to_temp(parquet_file_local) {
+            log::warn!(
+                "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::<()>();

server/src/storage.rs

@@ -164,18 +164,18 @@ impl StorageDir {
         }
     }
 
-    fn create_temp_dir(&self) -> io::Result<()> {
+    pub fn create_temp_dir(&self) -> io::Result<()> {
         fs::create_dir_all(&self.temp_dir)
     }
 
-    fn move_parquet_to_temp(&self, filename: String) -> io::Result<()> {
+    pub fn move_parquet_to_temp(&self, filename: String) -> io::Result<()> {
         fs::rename(
             self.data_path.join("data.parquet"),
             self.temp_dir.join(filename),
         )
     }
 
-    fn parquet_path_exists(&self) -> bool {
+    pub fn parquet_path_exists(&self) -> bool {
         self.data_path.join("data.parquet").exists()
     }
 }