fix for time partition limit

server/src/event/format/json.rs

@@ -48,7 +48,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, false)?;
+        let data = flatten_json_body(self.data, None, None, false)?;
         let stream_schema = schema;
 
         // incoming event may be a single json or a json array

server/src/handlers.rs

@@ -24,6 +24,7 @@ const PREFIX_META: &str = "x-p-meta-";
 const STREAM_NAME_HEADER_KEY: &str = "x-p-stream";
 const LOG_SOURCE_KEY: &str = "x-p-log-source";
 const TIME_PARTITION_KEY: &str = "x-p-time-partition";
+const TIME_PARTITION_LIMIT_KEY: &str = "x-p-time-partition-limit";
 const STATIC_SCHEMA_FLAG: &str = "x-p-static-schema-flag";
 const AUTHORIZATION_KEY: &str = "authorization";
 const SEPARATOR: char = '^';

server/src/handlers/http/ingest.rs

@@ -104,6 +104,7 @@ async fn push_logs(stream_name: String, req: HttpRequest, body: Bytes) -> Result
         .map_err(|_err| PostError::StreamNotFound(stream_name.clone()))?;
 
     let time_partition = object_store_format.time_partition;
+    let time_partition_limit = object_store_format.time_partition_limit;
     let static_schema_flag = object_store_format.static_schema_flag;
     let body_val: Value = serde_json::from_slice(&body)?;
     let size: usize = body.len();
@@ -129,7 +130,11 @@ async fn push_logs(stream_name: String, req: HttpRequest, body: Bytes) -> Result
         .process()
         .await?;
     } else {
-        let data = convert_array_to_object(body_val.clone(), time_partition.clone())?;
+        let data = convert_array_to_object(
+            body_val.clone(),
+            time_partition.clone(),
+            time_partition_limit,
+        )?;
         for value in data {
             let body_timestamp = value.get(&time_partition.clone().unwrap().to_string());
             parsed_timestamp = body_timestamp
@@ -210,6 +215,7 @@ pub async fn create_stream_if_not_exists(stream_name: &str) -> Result<(), PostEr
                 stream_name.to_string(),
                 "",
                 "",
+                "",
                 Arc::new(Schema::empty()),
             )
             .await?;

server/src/handlers/http/logstream.rs

@@ -21,7 +21,7 @@ use super::base_path_without_preceding_slash;
 use super::cluster::fetch_stats_from_ingestors;
 use super::cluster::utils::{merge_quried_stats, IngestionStats, QueriedStats, StorageStats};
 use crate::alerts::Alerts;
-use crate::handlers::{STATIC_SCHEMA_FLAG, TIME_PARTITION_KEY};
+use crate::handlers::{STATIC_SCHEMA_FLAG, TIME_PARTITION_KEY, TIME_PARTITION_LIMIT_KEY};
 use crate::metadata::STREAM_INFO;
 use crate::option::{Mode, CONFIG};
 use crate::static_schema::{convert_static_schema_to_arrow_schema, StaticSchema};
@@ -40,6 +40,7 @@ use itertools::Itertools;
 use serde_json::Value;
 use std::collections::HashMap;
 use std::fs;
+use std::num::NonZeroU32;
 use std::sync::Arc;
 
 pub async fn delete(req: HttpRequest) -> Result<impl Responder, StreamError> {
@@ -191,6 +192,29 @@ pub async fn put_stream(req: HttpRequest, body: Bytes) -> Result<impl Responder,
     } else {
         ""
     };
+    let mut time_partition_in_days: &str = "";
+    if let Some((_, time_partition_limit_name)) = req
+        .headers()
+        .iter()
+        .find(|&(key, _)| key == TIME_PARTITION_LIMIT_KEY)
+    {
+        let time_partition_limit = time_partition_limit_name.to_str().unwrap();
+        if !time_partition_limit.ends_with('d') {
+            return Err(StreamError::Custom {
+                msg: "missing 'd' suffix for duration value".to_string(),
+                status: StatusCode::BAD_REQUEST,
+            });
+        }
+        let days = &time_partition_limit[0..time_partition_limit.len() - 1];
+        if days.parse::<NonZeroU32>().is_err() {
+            return Err(StreamError::Custom {
+                msg: "could not convert duration to an unsigned number".to_string(),
+                status: StatusCode::BAD_REQUEST,
+            });
+        } else {
+            time_partition_in_days = days;
+        }
+    }
     let static_schema_flag = if let Some((_, static_schema_flag)) = req
         .headers()
         .iter()
@@ -235,7 +259,14 @@ pub async fn put_stream(req: HttpRequest, body: Bytes) -> Result<impl Responder,
                 });
     }
 
-    create_stream(stream_name, time_partition, static_schema_flag, schema).await?;
+    create_stream(
+        stream_name,
+        time_partition,
+        time_partition_in_days,
+        static_schema_flag,
+        schema,
+    )
+    .await?;
 
     Ok(("log stream created", StatusCode::OK))
 }
@@ -516,6 +547,7 @@ fn remove_id_from_alerts(value: &mut Value) {
 pub async fn create_stream(
     stream_name: String,
     time_partition: &str,
+    time_partition_limit: &str,
     static_schema_flag: &str,
     schema: Arc<Schema>,
 ) -> Result<(), CreateStreamError> {
@@ -528,6 +560,7 @@ pub async fn create_stream(
         .create_stream(
             &stream_name,
             time_partition,
+            time_partition_limit,
             static_schema_flag,
             schema.clone(),
         )
@@ -557,6 +590,7 @@ pub async fn create_stream(
         stream_name.to_string(),
         created_at,
         time_partition.to_string(),
+        time_partition_limit.to_string(),
         static_schema_flag.to_string(),
         static_schema,
     );
@@ -595,6 +629,7 @@ pub async fn get_stream_info(req: HttpRequest) -> Result<impl Responder, StreamE
         created_at: stream_meta.created_at.clone(),
         first_event_at: stream_meta.first_event_at.clone(),
         time_partition: stream_meta.time_partition.clone(),
+        time_partition_limit: stream_meta.time_partition_limit.clone(),
         cache_enabled: stream_meta.cache_enabled,
         static_schema_flag: stream_meta.static_schema_flag.clone(),
     };

server/src/metadata.rs

@@ -24,12 +24,11 @@ use once_cell::sync::Lazy;
 use std::collections::HashMap;
 use std::sync::{Arc, RwLock};
 
+use self::error::stream_info::{CheckAlertError, LoadError, MetadataError};
 use crate::alerts::Alerts;
 use crate::metrics::{EVENTS_INGESTED, EVENTS_INGESTED_SIZE};
 use crate::storage::{LogStream, ObjectStorage, StorageDir};
 use crate::utils::arrow::MergedRecordReader;
-
-use self::error::stream_info::{CheckAlertError, LoadError, MetadataError};
 use derive_more::{Deref, DerefMut};
 
 // TODO: make return type be of 'static lifetime instead of cloning
@@ -47,6 +46,7 @@ pub struct LogStreamMetadata {
     pub created_at: String,
     pub first_event_at: Option<String>,
     pub time_partition: Option<String>,
+    pub time_partition_limit: Option<String>,
     pub static_schema_flag: Option<String>,
 }
 
@@ -166,6 +166,7 @@ impl StreamInfo {
         stream_name: String,
         created_at: String,
         time_partition: String,
+        time_partition_limit: String,
         static_schema_flag: String,
         static_schema: HashMap<String, Arc<Field>>,
     ) {
@@ -181,6 +182,11 @@ impl StreamInfo {
             } else {
                 Some(time_partition)
             },
+            time_partition_limit: if time_partition_limit.is_empty() {
+                None
+            } else {
+                Some(time_partition_limit)
+            },
             static_schema_flag: if static_schema_flag != "true" {
                 None
             } else {
@@ -237,6 +243,7 @@ impl StreamInfo {
             created_at: meta.created_at,
             first_event_at: meta.first_event_at,
             time_partition: meta.time_partition,
+            time_partition_limit: meta.time_partition_limit,
             static_schema_flag: meta.static_schema_flag,
         };
 

server/src/storage.rs

@@ -30,16 +30,15 @@ mod s3;
 pub mod staging;
 mod store_metadata;
 
+use self::retention::Retention;
+pub use self::staging::StorageDir;
 pub use localfs::FSConfig;
 pub use object_storage::{ObjectStorage, ObjectStorageProvider};
 pub use s3::S3Config;
 pub use store_metadata::{
     put_remote_metadata, put_staging_metadata, resolve_parseable_metadata, StorageMetadata,
 };
 
-use self::retention::Retention;
-pub use self::staging::StorageDir;
-
 // metadata file names in a Stream prefix
 pub const STREAM_METADATA_FILE_NAME: &str = ".stream.json";
 pub const PARSEABLE_METADATA_FILE_NAME: &str = ".parseable.json";
@@ -94,6 +93,8 @@ pub struct ObjectStoreFormat {
     #[serde(skip_serializing_if = "Option::is_none")]
     pub time_partition: Option<String>,
     #[serde(skip_serializing_if = "Option::is_none")]
+    pub time_partition_limit: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
     pub static_schema_flag: Option<String>,
 }
 
@@ -109,6 +110,8 @@ pub struct StreamInfo {
     #[serde(skip_serializing_if = "Option::is_none")]
     pub time_partition: Option<String>,
     #[serde(skip_serializing_if = "Option::is_none")]
+    pub time_partition_limit: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
     pub static_schema_flag: Option<String>,
 }
 
@@ -155,6 +158,7 @@ impl Default for ObjectStoreFormat {
             cache_enabled: false,
             retention: None,
             time_partition: None,
+            time_partition_limit: None,
             static_schema_flag: None,
         }
     }

server/src/storage/object_storage.rs

@@ -127,6 +127,7 @@ pub trait ObjectStorage: Sync + 'static {
         &self,
         stream_name: &str,
         time_partition: &str,
+        time_partition_limit: &str,
         static_schema_flag: &str,
         schema: Arc<Schema>,
     ) -> Result<(), ObjectStorageError> {
@@ -139,6 +140,11 @@ pub trait ObjectStorage: Sync + 'static {
         } else {
             format.time_partition = Some(time_partition.to_string());
         }
+        if time_partition_limit.is_empty() {
+            format.time_partition_limit = None;
+        } else {
+            format.time_partition_limit = Some(time_partition_limit.to_string());
+        }
         if static_schema_flag != "true" {
             format.static_schema_flag = None;
         } else {

server/src/utils/json.rs

@@ -24,16 +24,24 @@ pub mod flatten;
 pub fn flatten_json_body(
     body: serde_json::Value,
     time_partition: Option<String>,
+    time_partition_limit: Option<String>,
     validation_required: bool,
 ) -> Result<Value, anyhow::Error> {
-    flatten::flatten(body, "_", time_partition, validation_required)
+    flatten::flatten(
+        body,
+        "_",
+        time_partition,
+        time_partition_limit,
+        validation_required,
+    )
 }
 
 pub fn convert_array_to_object(
     body: Value,
     time_partition: Option<String>,
+    time_partition_limit: Option<String>,
 ) -> Result<Vec<Value>, anyhow::Error> {
-    let data = flatten_json_body(body, time_partition, true)?;
+    let data = flatten_json_body(body, time_partition, time_partition_limit, true)?;
     let value_arr = match data {
         Value::Array(arr) => arr,
         value @ Value::Object(_) => vec![value],