trace: log steps of ingest and sync

Author: Devdutt Shenoi

src/event/mod.rs

@@ -24,7 +24,7 @@ use arrow_array::RecordBatch;
 use arrow_schema::{Field, Fields, Schema};
 use itertools::Itertools;
 use std::sync::Arc;
-use tracing::error;
+use tracing::{error, instrument};
 
 use self::error::EventError;
 pub use self::writer::STREAM_WRITERS;
@@ -36,7 +36,7 @@ pub const DEFAULT_TIMESTAMP_KEY: &str = "p_timestamp";
 pub const DEFAULT_TAGS_KEY: &str = "p_tags";
 pub const DEFAULT_METADATA_KEY: &str = "p_metadata";
 
-#[derive(Clone)]
+#[derive(Clone, Debug)]
 pub struct Event {
     pub stream_name: String,
     pub rb: RecordBatch,
@@ -51,6 +51,7 @@ pub struct Event {
 
 // Events holds the schema related to a each event for a single log stream
 impl Event {
+    #[instrument(level = "trace")]
     pub async fn process(&self) -> Result<(), EventError> {
         let mut key = get_schema_key(&self.rb.schema().fields);
         if self.time_partition.is_some() {
@@ -120,6 +121,7 @@ impl Event {
 
     // event process all events after the 1st event. Concatenates record batches
     // and puts them in memory store for each event.
+    #[instrument(level = "trace")]
     fn process_event(
         stream_name: &str,
         schema_key: &str,

src/event/writer/mod.rs

@@ -38,6 +38,7 @@ use chrono::NaiveDateTime;
 use chrono::Utc;
 use derive_more::{Deref, DerefMut};
 use once_cell::sync::Lazy;
+use tracing::debug;
 
 pub static STREAM_WRITERS: Lazy<WriterTable> = Lazy::new(WriterTable::default);
 
@@ -124,6 +125,9 @@ impl WriterTable {
                 )?;
             }
         };
+
+        debug!("Successful append to local");
+
         Ok(())
     }
 

src/handlers/http/ingest.rs

@@ -43,10 +43,12 @@ use http::StatusCode;
 use serde_json::Value;
 use std::collections::HashMap;
 use std::sync::Arc;
+use tracing::{debug, error, instrument, trace};
 
 // Handler for POST /api/v1/ingest
 // ingests events by extracting stream name from header
 // creates if stream does not exist
+#[instrument(level = "trace")]
 pub async fn ingest(req: HttpRequest, body: Bytes) -> Result<HttpResponse, PostError> {
     if let Some((_, stream_name)) = req
         .headers()
@@ -66,6 +68,7 @@ pub async fn ingest(req: HttpRequest, body: Bytes) -> Result<HttpResponse, PostE
         flatten_and_push_logs(req, body, stream_name).await?;
         Ok(HttpResponse::Ok().finish())
     } else {
+        error!("Ingestion request doesn't specify stream name");
         Err(PostError::Header(ParseHeaderError::MissingStreamName))
     }
 }
@@ -143,6 +146,7 @@ pub async fn ingest_otel_logs(req: HttpRequest, body: Bytes) -> Result<HttpRespo
 // Handler for POST /api/v1/logstream/{logstream}
 // only ingests events into the specified logstream
 // fails if the logstream does not exist
+#[instrument(level = "trace")]
 pub async fn post_event(req: HttpRequest, body: Bytes) -> Result<HttpResponse, PostError> {
     let stream_name: String = req.match_info().get("logstream").unwrap().parse().unwrap();
     let internal_stream_names = STREAM_INFO.list_internal_streams();
@@ -198,6 +202,7 @@ pub async fn create_stream_if_not_exists(
     let mut stream_exists = false;
     if STREAM_INFO.stream_exists(stream_name) {
         stream_exists = true;
+        trace!("Stream: {stream_name} already exists");
         return Ok(stream_exists);
     }
 
@@ -221,6 +226,8 @@ pub async fn create_stream_if_not_exists(
     )
     .await?;
 
+    debug!("Stream: {stream_name} created");
+
     Ok(stream_exists)
 }
 

src/handlers/http/modal/utils/ingest_utils.rs

@@ -26,6 +26,7 @@ use arrow_schema::Field;
 use bytes::Bytes;
 use chrono::{DateTime, NaiveDateTime, Utc};
 use serde_json::Value;
+use tracing::{debug, instrument, warn};
 
 use crate::{
     event::{
@@ -56,7 +57,7 @@ pub async fn flatten_and_push_logs(
                 json = otel::flatten_otel_logs(&body);
             }
             _ => {
-                log::warn!("Unknown log source: {}", log_source);
+                warn!("Unknown log source: {}", log_source);
                 push_logs(stream_name.to_string(), req.clone(), body).await?;
             }
         }
@@ -70,6 +71,7 @@ pub async fn flatten_and_push_logs(
     Ok(())
 }
 
+#[instrument(level = "trace")]
 pub async fn push_logs(
     stream_name: String,
     req: HttpRequest,
@@ -172,10 +174,13 @@ pub async fn push_logs(
         }
     }
 
+    debug!("Ingestion successful on stream: {stream_name}");
+
     Ok(())
 }
 
 #[allow(clippy::too_many_arguments)]
+#[instrument(level = "trace")]
 pub async fn create_process_record_batch(
     stream_name: String,
     req: HttpRequest,

src/storage/azure_blob.rs

@@ -191,6 +191,7 @@ pub fn to_object_store_path(path: &RelativePath) -> StorePath {
 
 // ObjStoreClient is generic client to enable interactions with different cloudprovider's
 // object store such as S3 and Azure Blob
+#[derive(Debug)]
 pub struct BlobStore {
     client: LimitStore<MicrosoftAzure>,
     account: String,

src/storage/localfs.rs

@@ -80,6 +80,7 @@ impl ObjectStorageProvider for FSConfig {
     }
 }
 
+#[derive(Debug)]
 pub struct LocalFS {
     // absolute path of the data directory
     root: PathBuf,

src/storage/object_storage.rs

@@ -48,7 +48,7 @@ use datafusion::{datasource::listing::ListingTableUrl, execution::runtime_env::R
 use itertools::Itertools;
 use relative_path::RelativePath;
 use relative_path::RelativePathBuf;
-use tracing::error;
+use tracing::{debug, error, instrument, trace};
 
 use std::collections::BTreeMap;
 use std::{
@@ -67,7 +67,7 @@ pub trait ObjectStorageProvider: StorageMetrics + std::fmt::Debug + Send + Sync
 }
 
 #[async_trait]
-pub trait ObjectStorage: Send + Sync + 'static {
+pub trait ObjectStorage: std::fmt::Debug + Send + Sync + 'static {
     async fn get_object(&self, path: &RelativePath) -> Result<Bytes, ObjectStorageError>;
     // TODO: make the filter function optional as we may want to get all objects
     async fn get_objects(
@@ -538,8 +538,10 @@ pub trait ObjectStorage: Send + Sync + 'static {
         Ok(Bytes::new())
     }
 
+    #[instrument(level = "debug")]
     async fn sync(&self, shutdown_signal: bool) -> Result<(), ObjectStorageError> {
         if !Path::new(&CONFIG.staging_dir()).exists() {
+            trace!("Nothing to sync");
             return Ok(());
         }
 
@@ -552,6 +554,7 @@ pub trait ObjectStorage: Send + Sync + 'static {
             let cache_enabled = STREAM_INFO
                 .get_cache_enabled(stream)
                 .map_err(|err| ObjectStorageError::UnhandledError(Box::new(err)))?;
+
             let time_partition = STREAM_INFO
                 .get_time_partition(stream)
                 .map_err(|err| ObjectStorageError::UnhandledError(Box::new(err)))?;
@@ -568,6 +571,8 @@ pub trait ObjectStorage: Send + Sync + 'static {
             )
             .map_err(|err| ObjectStorageError::UnhandledError(Box::new(err)))?;
 
+            debug!("Arrow files compressed into parquet for stream: {stream}");
+
             if let Some(schema) = schema {
                 let static_schema_flag = STREAM_INFO
                     .get_static_schema_flag(stream)
@@ -614,6 +619,8 @@ pub trait ObjectStorage: Send + Sync + 'static {
                 if let Err(e) = self.upload_file(&stream_relative_path, &file).await {
                     error!("Failed to upload file {}: {:?}", filename, e);
                     continue; // Skip to the next file
+                } else {
+                    debug!("Parquet file uploaded to s3 for stream: {stream}");
                 }
 
                 let absolute_path = self
@@ -622,6 +629,7 @@ pub trait ObjectStorage: Send + Sync + 'static {
                 let store = CONFIG.storage().get_object_store();
                 let manifest =
                     catalog::create_from_parquet_file(absolute_path.clone(), &file).unwrap();
+
                 catalog::update_snapshot(store, stream, manifest).await?;
                 if cache_enabled && cache_manager.is_some() {
                     cache_updates

src/storage/s3.rs

@@ -316,6 +316,7 @@ fn to_object_store_path(path: &RelativePath) -> StorePath {
     StorePath::from(path.as_str())
 }
 
+#[derive(Debug)]
 pub struct S3 {
     client: LimitStore<AmazonS3>,
     bucket: String,

src/storage/staging.rs

@@ -50,7 +50,7 @@ use std::{
     process,
     sync::Arc,
 };
-use tracing::{error, info};
+use tracing::{error, info, instrument};
 
 const ARROW_FILE_EXTENSION: &str = "data.arrows";
 // const PARQUET_FILE_EXTENSION: &str = "data.parquet";
@@ -223,6 +223,7 @@ impl StorageDir {
 //     data_path.join(dir)
 // }
 
+#[instrument(level = "debug")]
 pub fn convert_disk_files_to_parquet(
     stream: &str,
     dir: &StorageDir,