fix: add first event time of queryable data

gurjotkaur20 · gurjotkaur20 · commit 150f6abc0282 · 2024-01-23T21:58:02.000+05:30
diff --git a/server/src/handlers/http/ingest.rs b/server/src/handlers/http/ingest.rs
@@ -19,6 +19,7 @@
 use actix_web::{http::header::ContentType, HttpRequest, HttpResponse};
 use arrow_schema::Field;
 use bytes::Bytes;
+use chrono::Local;
 use http::StatusCode;
 use serde_json::Value;
 use std::collections::{BTreeMap, HashMap};
@@ -31,7 +32,8 @@ use crate::handlers::{
     LOG_SOURCE_KEY, LOG_SOURCE_KINESIS, LOG_SOURCE_OTEL, PREFIX_META, PREFIX_TAGS, SEPARATOR,
     STREAM_NAME_HEADER_KEY,
 };
-use crate::metadata::STREAM_INFO;
+use crate::metadata::{self, STREAM_INFO};
+use crate::option::CONFIG;
 use crate::utils::header_parsing::{collect_labelled_headers, ParseHeaderError};
 
 use super::kinesis;
@@ -47,8 +49,37 @@ pub async fn ingest(req: HttpRequest, body: Bytes) -> Result<HttpResponse, PostE
         .find(|&(key, _)| key == STREAM_NAME_HEADER_KEY)
     {
         let stream_name = stream_name.to_str().unwrap().to_owned();
+
+        let mut is_new_stream = true;
+        if STREAM_INFO.stream_exists(&stream_name) {
+            is_new_stream = false;
+        }
+
         create_stream_if_not_exists(&stream_name).await?;
 
+        if is_new_stream {
+            let first_event_at = Local::now().to_rfc3339();
+            if let Err(err) = CONFIG
+                .storage()
+                .get_object_store()
+                .put_first_event_at(&stream_name, first_event_at.as_str())
+                .await
+            {
+                log::error!(
+                    "Failed to update first_event_at in metadata for stream {:?} {err:?}",
+                    stream_name
+                );
+            }
+
+            if let Err(err) = metadata::STREAM_INFO.set_first_event_at(&stream_name, first_event_at)
+            {
+                log::error!(
+                    "Failed to update first_event_at in streaminfo for stream {:?} {err:?}",
+                    stream_name
+                );
+            }
+        }
+
         flatten_and_push_logs(req, body, stream_name).await?;
         Ok(HttpResponse::Ok().finish())
     } else {
diff --git a/server/src/handlers/http/logstream.rs b/server/src/handlers/http/logstream.rs
@@ -269,16 +269,16 @@ pub async fn get_stats(req: HttpRequest) -> Result<impl Responder, StreamError>
         .ok_or(StreamError::StreamNotFound(stream_name.clone()))?;
 
     let hash_map = STREAM_INFO.read().unwrap();
-    let stream_creation_time = &hash_map
+    let stream_meta = &hash_map
         .get(&stream_name)
-        .ok_or(StreamError::StreamNotFound(stream_name.clone()))?
-        .created_at;
+        .ok_or(StreamError::StreamNotFound(stream_name.clone()))?;
 
     let time = Utc::now();
 
     let stats = serde_json::json!({
         "stream": stream_name,
-        "creation_time": stream_creation_time,
+        "creation_time": stream_meta.created_at,
+        "first_event_at": stream_meta.first_event_at,
         "time": time,
         "ingestion": {
             "count": stats.events,
diff --git a/server/src/metadata.rs b/server/src/metadata.rs
@@ -45,6 +45,7 @@ pub struct LogStreamMetadata {
     pub alerts: Alerts,
     pub cache_enabled: bool,
     pub created_at: String,
+    pub first_event_at: String,
 }
 
 // It is very unlikely that panic will occur when dealing with metadata.
@@ -128,6 +129,19 @@ impl StreamInfo {
             })
     }
 
+    pub fn set_first_event_at(
+        &self,
+        stream_name: &str,
+        first_event_at: String,
+    ) -> Result<(), MetadataError> {
+        let mut map = self.write().expect(LOCK_EXPECT);
+        map.get_mut(stream_name)
+            .ok_or(MetadataError::StreamMetaNotFound(stream_name.to_string()))
+            .map(|metadata| {
+                metadata.first_event_at = first_event_at;
+            })
+    }
+
     pub fn add_stream(&self, stream_name: String, created_at: String) {
         let mut map = self.write().expect(LOCK_EXPECT);
         let metadata = LogStreamMetadata {
@@ -170,6 +184,7 @@ impl StreamInfo {
                 alerts,
                 cache_enabled: meta.cache_enabled,
                 created_at: meta.created_at,
+                first_event_at: meta.first_event_at,
             };
 
             let mut map = self.write().expect(LOCK_EXPECT);
diff --git a/server/src/stats.rs b/server/src/stats.rs
@@ -16,7 +16,14 @@
  *
  */
 
+use chrono::{DateTime, TimeZone, Utc};
+use datafusion::arrow::json::writer::record_batches_to_json_rows;
+use datafusion::arrow::record_batch::RecordBatch;
+use datafusion::error::DataFusionError;
+
 use crate::metrics::{EVENTS_INGESTED, EVENTS_INGESTED_SIZE, STORAGE_SIZE};
+use crate::query;
+use crate::query::error::ExecuteError;
 
 /// Helper struct type created by copying stats values from metadata
 #[derive(Debug, Default, serde::Serialize, serde::Deserialize, Clone, Copy, PartialEq, Eq)]
@@ -71,3 +78,49 @@ fn event_labels<'a>(stream_name: &'a str, format: &'static str) -> [&'a str; 2]
 fn storage_size_labels(stream_name: &str) -> [&str; 3] {
     ["data", stream_name, "parquet"]
 }
+
+pub async fn get_first_event_at(stream_name: &str) -> Result<Option<String>, QueryError> {
+    let query_string = format!(
+        "SELECT p_timestamp FROM {} ORDER BY p_timestamp LIMIT 1",
+        stream_name
+    );
+    let first_time: DateTime<Utc> = Utc
+        .timestamp_opt(0, 0)
+        .single()
+        .expect("Failed to get the first UTC time");
+    let now_time: DateTime<Utc> = Utc::now();
+    let session_state = query::QUERY_SESSION.state();
+    let logical_plan = session_state.create_logical_plan(&query_string).await?;
+
+    let query = query::Query {
+        raw_logical_plan: logical_plan,
+        start: first_time,
+        end: now_time,
+        filter_tag: Some(Vec::new()),
+    };
+
+    let (records, _fields) = query.execute().await?;
+    let records_itr: Vec<&RecordBatch> = records.iter().collect();
+    let json_records = record_batches_to_json_rows(&records_itr).unwrap();
+
+    if let Some(single_record) = json_records.first() {
+        if let Some(p_timestamp_value) = single_record.get("p_timestamp") {
+            let p_timestamp_str = p_timestamp_value.as_str().unwrap_or_default();
+            return Ok(Some(p_timestamp_str.to_string()));
+        }
+    }
+
+    Ok(None)
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum QueryError {
+    #[error("While generating times for 'now' failed to parse duration")]
+    NotValidDuration(#[from] humantime::DurationError),
+    #[error("Parsed duration out of range")]
+    OutOfRange(#[from] chrono::OutOfRangeError),
+    #[error("Datafusion Error: {0}")]
+    Datafusion(#[from] DataFusionError),
+    #[error("Query execution failed due to {0}")]
+    Execute(#[from] ExecuteError),
+}
diff --git a/server/src/storage.rs b/server/src/storage.rs
@@ -69,6 +69,8 @@ pub struct ObjectStoreFormat {
     pub objectstore_format: String,
     #[serde(rename = "created-at")]
     pub created_at: String,
+    #[serde(rename = "first-event-at")]
+    pub first_event_at: String,
     pub owner: Owner,
     pub permissions: Vec<Permisssion>,
     pub stats: Stats,
@@ -113,6 +115,7 @@ impl Default for ObjectStoreFormat {
             version: CURRENT_SCHEMA_VERSION.to_string(),
             objectstore_format: CURRENT_OBJECT_STORE_VERSION.to_string(),
             created_at: Local::now().to_rfc3339(),
+            first_event_at: Default::default(),
             owner: Owner::new("".to_string(), "".to_string()),
             permissions: vec![Permisssion::new("parseable".to_string())],
             stats: Stats::default(),
diff --git a/server/src/storage/object_storage.rs b/server/src/storage/object_storage.rs
@@ -161,6 +161,23 @@ pub trait ObjectStorage: Sync + 'static {
         self.put_object(&path, to_bytes(&stream_metadata)).await
     }
 
+    async fn put_first_event_at(
+        &self,
+        stream_name: &str,
+        first_event_at: &str,
+    ) -> Result<(), ObjectStorageError> {
+        let path = stream_json_path(stream_name);
+        let stream_metadata = self.get_object(&path).await?;
+        let first_event_ts =
+            serde_json::to_value(first_event_at).expect("first_event_at is perfectly serializable");
+        let mut stream_metadata: serde_json::Value =
+            serde_json::from_slice(&stream_metadata).expect("parseable config is valid json");
+
+        stream_metadata["first-event-at"] = first_event_ts;
+
+        self.put_object(&path, to_bytes(&stream_metadata)).await
+    }
+
     async fn put_metadata(
         &self,
         parseable_metadata: &StorageMetadata,
diff --git a/server/src/storage/retention.rs b/server/src/storage/retention.rs
@@ -175,12 +175,12 @@ impl From<Retention> for Vec<TaskView> {
 }
 
 mod action {
-    use chrono::{Days, NaiveDate, Utc};
+    use chrono::{Days, Local, NaiveDate, NaiveDateTime, TimeZone, Utc};
     use futures::{stream::FuturesUnordered, StreamExt};
     use itertools::Itertools;
     use relative_path::RelativePathBuf;
 
-    use crate::option::CONFIG;
+    use crate::{metadata, option::CONFIG, stats};
 
     pub(super) async fn delete(stream_name: String, days: u32) {
         log::info!("running retention task - delete");
@@ -219,6 +219,46 @@ mod action {
                 log::error!("Failed to run delete task {err:?}")
             }
         }
+
+        // update first-event-at after the cleanup
+        let first_event = stats::get_first_event_at(&stream_name).await;
+        let first_event_at = match first_event {
+            Ok(Some(value)) => Some(value.clone()),
+            Ok(None) => None,
+            Err(_err) => None,
+        };
+        if let Some(first_event_str) = first_event_at {
+            let naive_datetime =
+                NaiveDateTime::parse_from_str(&first_event_str, "%Y-%m-%dT%H:%M:%S%.3f")
+                    .expect("Failed to parse timestamp first_event_at");
+
+            let parsed_timestamp = Utc
+                .from_utc_datetime(&naive_datetime)
+                .with_timezone(&Local)
+                .format("%Y-%m-%dT%H:%M:%S%.9f%:z")
+                .to_string();
+
+            if let Err(err) = CONFIG
+                .storage()
+                .get_object_store()
+                .put_first_event_at(&stream_name, parsed_timestamp.as_str())
+                .await
+            {
+                log::error!(
+                    "Failed to update first_event_at in metadata for stream {:?} {err:?}",
+                    stream_name
+                );
+            }
+
+            if let Err(err) =
+                metadata::STREAM_INFO.set_first_event_at(&stream_name, parsed_timestamp)
+            {
+                log::error!(
+                    "Failed to update first_event_at in streaminfo for stream {:?} {err:?}",
+                    stream_name
+                );
+            }
+        }
     }
 
     fn get_retain_until(current_date: NaiveDate, days: u64) -> NaiveDate {
