trace: log steps of ingest and sync

Author: Devdutt Shenoi

src/event/format/json.rs

@@ -30,7 +30,7 @@ use std::{collections::HashMap, sync::Arc};
 use tracing::error;
 
 use super::{EventFormat, Metadata, Tags};
-use crate::utils::{arrow::get_field, json::flatten_json_body};
+use crate::utils::{arrow::get_field, json};
 
 pub struct Event {
     pub data: Value,
@@ -49,7 +49,7 @@ impl EventFormat for Event {
         static_schema_flag: Option<String>,
         time_partition: Option<String>,
     ) -> Result<(Self::Data, Vec<Arc<Field>>, bool, Tags, Metadata), anyhow::Error> {
-        let data = flatten_json_body(self.data, None, None, None, false)?;
+        let data = json::flatten::flatten(self.data, "_", None, None, None, false)?;
         let stream_schema = schema;
 
         // incoming event may be a single json or a json array

src/event/format/mod.rs

@@ -24,6 +24,7 @@ use arrow_array::{RecordBatch, StringArray};
 use arrow_schema::{DataType, Field, Schema, TimeUnit};
 use chrono::DateTime;
 use serde_json::Value;
+use tracing::{debug, error};
 
 use crate::utils::{self, arrow::get_field};
 
@@ -59,23 +60,17 @@ pub trait EventFormat: Sized {
             time_partition.clone(),
         )?;
 
-        if get_field(&schema, DEFAULT_TAGS_KEY).is_some() {
-            return Err(anyhow!("field {} is a reserved field", DEFAULT_TAGS_KEY));
-        };
-
-        if get_field(&schema, DEFAULT_METADATA_KEY).is_some() {
-            return Err(anyhow!(
-                "field {} is a reserved field",
-                DEFAULT_METADATA_KEY
-            ));
-        };
-
-        if get_field(&schema, DEFAULT_TIMESTAMP_KEY).is_some() {
-            return Err(anyhow!(
-                "field {} is a reserved field",
-                DEFAULT_TIMESTAMP_KEY
-            ));
-        };
+        for reserved_field in [
+            DEFAULT_TAGS_KEY,
+            DEFAULT_METADATA_KEY,
+            DEFAULT_TIMESTAMP_KEY,
+        ] {
+            if get_field(&schema, DEFAULT_TAGS_KEY).is_some() {
+                let msg = format!("{} is a reserved field", reserved_field);
+                error!("{msg}");
+                return Err(anyhow!(msg));
+            }
+        }
 
         // add the p_timestamp field to the event schema to the 0th index
         schema.insert(
@@ -100,7 +95,9 @@ pub trait EventFormat: Sized {
         // prepare the record batch and new fields to be added
         let mut new_schema = Arc::new(Schema::new(schema));
         if !Self::is_schema_matching(new_schema.clone(), storage_schema, static_schema_flag) {
-            return Err(anyhow!("Schema mismatch"));
+            let msg = "Schema mismatch";
+            error!("{msg}");
+            return Err(anyhow!(msg));
         }
         new_schema = update_field_type_in_schema(new_schema, None, time_partition, None);
         let rb = Self::decode(data, new_schema.clone())?;
@@ -269,6 +266,10 @@ pub fn update_data_type_to_datetime(
                 if let Value::Object(map) = &value {
                     if let Some(Value::String(s)) = map.get(field.name()) {
                         if DateTime::parse_from_rfc3339(s).is_ok() {
+                            debug!(
+                                "Field type updated to timestamp from string: {}",
+                                field.name()
+                            );
                             // Update the field's data type to Timestamp
                             return Field::new(
                                 field.name().clone(),

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,8 @@ 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);
     }
 
@@ -220,6 +225,7 @@ pub async fn create_stream_if_not_exists(
         stream_type,
     )
     .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(
@@ -389,8 +389,10 @@ pub trait ObjectStorage: Send + Sync + 'static {
             )),
             Err(err) => {
                 if matches!(err, ObjectStorageError::NoSuchKey(_)) {
+                    debug!("Couldn't find manifest file: {path}");
                     Ok(None)
                 } else {
+                    error!("Couldn't get the manifest file: {err}");
                     Err(err)
                 }
             }
@@ -538,8 +540,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 +556,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,12 +573,17 @@ 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)
                     .map_err(|err| ObjectStorageError::UnhandledError(Box::new(err)))?;
                 if static_schema_flag.is_none() {
-                    commit_schema_to_storage(stream, schema).await?;
+                    if let Err(err) = commit_schema_to_storage(stream, schema).await {
+                        error!("Error while committing schema to storage: {err}");
+                        return Err(err);
+                    }
                 }
             }
 
@@ -614,6 +624,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,7 +634,10 @@ 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 let Err(err) = catalog::update_snapshot(store, stream, manifest).await {
+                    error!("Error while updating snapshot: {err}");
+                }
                 if cache_enabled && cache_manager.is_some() {
                     cache_updates
                         .entry(stream)

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,