tracing: subscribe to span creation events, log important ingestion and sync steps (#1039)

* refactor: `log` ~> `tracing` (#1013)

Using tracing will allow us to capture span related info in addition to the event based context that we already capture

---------

Signed-off-by: Devdutt Shenoi <[email protected]>

* note span creation and deletion

* trace: log steps of ingest and sync

---------

Signed-off-by: Devdutt Shenoi <[email protected]>

Author: Devdutt Shenoi

Cargo.lock

@@ -62,7 +62,6 @@ hostname = "0.4.0"
 http = "0.2.7"
 humantime-serde = "1.1"
 itertools = "0.13.0"
-log = "0.4"
 num_cpus = "1.15"
 once_cell = "1.17.1"
 prometheus = { version = "0.13", features = ["process"] }
@@ -105,6 +104,7 @@ path-clean = "1.0.1"
 prost = "0.13.3"
 prometheus-parse = "0.2.5"
 sha2 = "0.10.8"
+tracing = "0.1.41"
 
 [build-dependencies]
 cargo_toml = "0.20.1"

Cargo.toml

@@ -28,6 +28,7 @@ use chrono::Utc;
 use http::{header::AUTHORIZATION, HeaderMap, HeaderValue};
 use humantime_serde::re::humantime;
 use reqwest::ClientBuilder;
+use tracing::error;
 
 use crate::utils::json;
 
@@ -239,7 +240,7 @@ impl CallableTarget for SlackWebHook {
         };
 
         if let Err(e) = client.post(&self.endpoint).json(&alert).send().await {
-            log::error!("Couldn't make call to webhook, error: {}", e)
+            error!("Couldn't make call to webhook, error: {}", e)
         }
     }
 }
@@ -277,7 +278,7 @@ impl CallableTarget for OtherWebHook {
             .headers((&self.headers).try_into().expect("valid_headers"));
 
         if let Err(e) = request.body(alert).send().await {
-            log::error!("Couldn't make call to webhook, error: {}", e)
+            error!("Couldn't make call to webhook, error: {}", e)
         }
     }
 }
@@ -356,7 +357,7 @@ impl CallableTarget for AlertManager {
         };
 
         if let Err(e) = client.post(&self.endpoint).json(&alerts).send().await {
-            log::error!("Couldn't make call to alertmanager, error: {}", e)
+            error!("Couldn't make call to alertmanager, error: {}", e)
         }
     }
 }

src/alerts/target.rs

@@ -37,6 +37,7 @@ use std::collections::HashMap;
 use std::sync::Mutex;
 use std::time::Duration;
 use sysinfo::System;
+use tracing::{error, info};
 use ulid::Ulid;
 
 const ANALYTICS_SERVER_URL: &str = "https://analytics.parseable.io:80";
@@ -291,7 +292,7 @@ async fn build_metrics() -> HashMap<String, Value> {
 }
 
 pub fn init_analytics_scheduler() -> anyhow::Result<()> {
-    log::info!("Setting up schedular for anonymous user analytics");
+    info!("Setting up schedular for anonymous user analytics");
 
     let mut scheduler = AsyncScheduler::new();
     scheduler
@@ -302,7 +303,7 @@ pub fn init_analytics_scheduler() -> anyhow::Result<()> {
                 .unwrap_or_else(|err| {
                     // panicing because seperate thread
                     // TODO: a better way to handle this
-                    log::error!("Error while sending analytics: {}", err.to_string());
+                    error!("Error while sending analytics: {}", err.to_string());
                     panic!("{}", err.to_string());
                 })
                 .send()

src/analytics.rs

@@ -37,6 +37,7 @@ use bytes::Bytes;
 use chrono::{DateTime, Local, NaiveTime, Utc};
 use relative_path::RelativePathBuf;
 use std::io::Error as IOError;
+use tracing::{error, info};
 pub mod column;
 pub mod manifest;
 pub mod snapshot;
@@ -280,7 +281,7 @@ async fn create_manifest(
             };
             first_event_at = Some(lower_bound.with_timezone(&Local).to_rfc3339());
             if let Err(err) = STREAM_INFO.set_first_event_at(stream_name, first_event_at.clone()) {
-                log::error!(
+                error!(
                     "Failed to update first_event_at in streaminfo for stream {:?} {err:?}",
                     stream_name
                 );
@@ -360,7 +361,7 @@ pub async fn get_first_event(
                 let manifests = meta_clone.snapshot.manifest_list;
                 let time_partition = meta_clone.time_partition;
                 if manifests.is_empty() {
-                    log::info!("No manifest found for stream {stream_name}");
+                    info!("No manifest found for stream {stream_name}");
                     return Err(ObjectStorageError::Custom("No manifest found".to_string()));
                 }
                 let manifest = &manifests[0];
@@ -400,7 +401,7 @@ pub async fn get_first_event(
                 handlers::http::cluster::get_ingestor_info()
                     .await
                     .map_err(|err| {
-                        log::error!("Fatal: failed to get ingestor info: {:?}", err);
+                        error!("Fatal: failed to get ingestor info: {:?}", err);
                         ObjectStorageError::from(err)
                     })?;
             let mut ingestors_first_event_at: Vec<String> = Vec::new();

src/catalog/mod.rs

@@ -27,9 +27,10 @@ use datafusion::arrow::util::bit_util::round_upto_multiple_of_64;
 use itertools::Itertools;
 use serde_json::Value;
 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,
@@ -48,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
@@ -224,7 +225,7 @@ fn valid_type(data_type: &DataType, value: &Value) -> bool {
         }
         DataType::Timestamp(_, _) => value.is_string() || value.is_number(),
         _ => {
-            log::error!("Unsupported datatype {:?}, value {:?}", data_type, value);
+            error!("Unsupported datatype {:?}, value {:?}", data_type, value);
             unreachable!()
         }
     }

src/event/format/json.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/format/mod.rs

@@ -24,6 +24,7 @@ use arrow_array::RecordBatch;
 use arrow_schema::{Field, Fields, Schema};
 use itertools::Itertools;
 use std::sync::Arc;
+use tracing::{error, instrument};
 
 use self::error::EventError;
 pub use self::writer::STREAM_WRITERS;
@@ -35,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,
@@ -50,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() {
@@ -93,7 +95,7 @@ impl Event {
             .check_alerts(&self.stream_name, &self.rb)
             .await
         {
-            log::error!("Error checking for alerts. {:?}", e);
+            error!("Error checking for alerts. {:?}", e);
         }
 
         Ok(())
@@ -119,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/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/event/writer/mod.rs

@@ -26,6 +26,7 @@ use serde_json::json;
 use std::net::SocketAddr;
 use std::time::Instant;
 use tonic::codec::CompressionEncoding;
+use tracing::{error, info};
 
 use futures_util::{Future, TryFutureExt};
 
@@ -135,7 +136,7 @@ impl FlightService for AirServiceImpl {
 
         let ticket = get_query_from_ticket(&req)?;
 
-        log::info!("query requested to airplane: {:?}", ticket);
+        info!("query requested to airplane: {:?}", ticket);
 
         // get the query session_state
         let session_state = QUERY_SESSION.state();
@@ -145,7 +146,7 @@ impl FlightService for AirServiceImpl {
             .create_logical_plan(&ticket.query)
             .await
             .map_err(|err| {
-                log::error!("Datafusion Error: Failed to create logical plan: {}", err);
+                error!("Datafusion Error: Failed to create logical plan: {}", err);
                 Status::internal("Failed to create logical plan")
             })?;
 
@@ -269,7 +270,7 @@ impl FlightService for AirServiceImpl {
         )
         .await
         {
-            log::error!("{}", err);
+            error!("{}", err);
         };
 
         /*