enhancement: add stream type to stream definition

1. add stream_type=UserDefined for user defined streams
2. add stream_type=Internal for pmeta
3. migrate existing streams' stream.json from v4 to v5
4. add field stream_type in migration for existing streams
5. add field stream_type in GET /logstream/{logstream}/info API

Author: nikhilsinhaparseable

server/src/event.rs

@@ -27,7 +27,7 @@ use std::sync::Arc;
 
 use self::error::EventError;
 pub use self::writer::STREAM_WRITERS;
-use crate::{handlers::http::ingest::PostError, metadata};
+use crate::{handlers::http::ingest::PostError, metadata, storage::StreamType};
 use chrono::NaiveDateTime;
 use std::collections::HashMap;
 
@@ -45,6 +45,7 @@ pub struct Event {
     pub parsed_timestamp: NaiveDateTime,
     pub time_partition: Option<String>,
     pub custom_partition_values: HashMap<String, String>,
+    pub stream_type: StreamType,
 }
 
 // Events holds the schema related to a each event for a single log stream
@@ -75,6 +76,7 @@ impl Event {
             self.rb.clone(),
             self.parsed_timestamp,
             &self.custom_partition_values,
+            &self.stream_type,
         )?;
 
         metadata::STREAM_INFO.update_stats(
@@ -106,6 +108,7 @@ impl Event {
             self.rb.clone(),
             self.parsed_timestamp,
             &self.custom_partition_values,
+            &self.stream_type,
         )
         .map_err(PostError::Event)
     }
@@ -122,13 +125,15 @@ impl Event {
         rb: RecordBatch,
         parsed_timestamp: NaiveDateTime,
         custom_partition_values: &HashMap<String, String>,
+        stream_type: &StreamType,
     ) -> Result<(), EventError> {
         STREAM_WRITERS.append_to_local(
             stream_name,
             schema_key,
             rb,
             parsed_timestamp,
             custom_partition_values.clone(),
+            stream_type,
         )?;
         Ok(())
     }

server/src/event/writer.rs

@@ -26,8 +26,8 @@ use std::{
 };
 
 use crate::{
-    handlers::http::cluster::INTERNAL_STREAM_NAME,
     option::{Mode, CONFIG},
+    storage::StreamType,
     utils,
 };
 
@@ -92,6 +92,7 @@ impl WriterTable {
         record: RecordBatch,
         parsed_timestamp: NaiveDateTime,
         custom_partition_values: HashMap<String, String>,
+        stream_type: &StreamType,
     ) -> Result<(), StreamWriterError> {
         let hashmap_guard = self.read().unwrap();
 
@@ -104,6 +105,7 @@ impl WriterTable {
                     record,
                     parsed_timestamp,
                     &custom_partition_values,
+                    stream_type,
                 )?;
             }
             None => {
@@ -118,12 +120,14 @@ impl WriterTable {
                     record,
                     parsed_timestamp,
                     &custom_partition_values,
+                    stream_type,
                 )?;
             }
         };
         Ok(())
     }
 
+    #[allow(clippy::too_many_arguments)]
     fn handle_existing_writer(
         &self,
         stream_writer: &Mutex<Writer>,
@@ -132,8 +136,9 @@ impl WriterTable {
         record: RecordBatch,
         parsed_timestamp: NaiveDateTime,
         custom_partition_values: &HashMap<String, String>,
+        stream_type: &StreamType,
     ) -> Result<(), StreamWriterError> {
-        if CONFIG.parseable.mode != Mode::Query || stream_name == INTERNAL_STREAM_NAME {
+        if CONFIG.parseable.mode != Mode::Query || *stream_type == StreamType::Internal {
             stream_writer.lock().unwrap().push(
                 stream_name,
                 schema_key,
@@ -151,6 +156,7 @@ impl WriterTable {
         Ok(())
     }
 
+    #[allow(clippy::too_many_arguments)]
     fn handle_missing_writer(
         &self,
         mut map: RwLockWriteGuard<HashMap<String, Mutex<Writer>>>,
@@ -159,10 +165,11 @@ impl WriterTable {
         record: RecordBatch,
         parsed_timestamp: NaiveDateTime,
         custom_partition_values: &HashMap<String, String>,
+        stream_type: &StreamType,
     ) -> Result<(), StreamWriterError> {
         match map.get(stream_name) {
             Some(writer) => {
-                if CONFIG.parseable.mode != Mode::Query || stream_name == INTERNAL_STREAM_NAME {
+                if CONFIG.parseable.mode != Mode::Query || *stream_type == StreamType::Internal {
                     writer.lock().unwrap().push(
                         stream_name,
                         schema_key,
@@ -175,7 +182,7 @@ impl WriterTable {
                 }
             }
             None => {
-                if CONFIG.parseable.mode != Mode::Query || stream_name == INTERNAL_STREAM_NAME {
+                if CONFIG.parseable.mode != Mode::Query || *stream_type == StreamType::Internal {
                     let mut writer = Writer::default();
                     writer.push(
                         stream_name,

server/src/handlers/http/ingest.rs

@@ -34,7 +34,7 @@ use crate::localcache::CacheError;
 use crate::metadata::error::stream_info::MetadataError;
 use crate::metadata::{self, STREAM_INFO};
 use crate::option::{Mode, CONFIG};
-use crate::storage::{LogStream, ObjectStorageError};
+use crate::storage::{LogStream, ObjectStorageError, StreamType};
 use crate::utils::header_parsing::{collect_labelled_headers, ParseHeaderError};
 use crate::utils::json::convert_array_to_object;
 use actix_web::{http::header::ContentType, HttpRequest, HttpResponse};
@@ -57,13 +57,15 @@ 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();
-        if stream_name.eq(INTERNAL_STREAM_NAME) {
+        let internal_stream_names = STREAM_INFO.list_internal_streams();
+
+        if internal_stream_names.contains(&stream_name) || stream_name == INTERNAL_STREAM_NAME {
             return Err(PostError::Invalid(anyhow::anyhow!(
-                "Stream {} is an internal stream and cannot be ingested into",
+                "The stream {} is reserved for internal use and cannot be ingested into",
                 stream_name
             )));
         }
-        create_stream_if_not_exists(&stream_name, false).await?;
+        create_stream_if_not_exists(&stream_name, StreamType::UserDefined).await?;
 
         flatten_and_push_logs(req, body, stream_name).await?;
         Ok(HttpResponse::Ok().finish())
@@ -73,7 +75,7 @@ pub async fn ingest(req: HttpRequest, body: Bytes) -> Result<HttpResponse, PostE
 }
 
 pub async fn ingest_internal_stream(stream_name: String, body: Bytes) -> Result<(), PostError> {
-    create_stream_if_not_exists(&stream_name, true).await?;
+    create_stream_if_not_exists(&stream_name, StreamType::Internal).await?;
     let size: usize = body.len();
     let parsed_timestamp = Utc::now().naive_utc();
     let (rb, is_first) = {
@@ -100,6 +102,7 @@ pub async fn ingest_internal_stream(stream_name: String, body: Bytes) -> Result<
         parsed_timestamp,
         time_partition: None,
         custom_partition_values: HashMap::new(),
+        stream_type: StreamType::Internal,
     }
     .process()
     .await?;
@@ -116,7 +119,7 @@ pub async fn ingest_otel_logs(req: HttpRequest, body: Bytes) -> Result<HttpRespo
         .find(|&(key, _)| key == STREAM_NAME_HEADER_KEY)
     {
         let stream_name = stream_name.to_str().unwrap().to_owned();
-        create_stream_if_not_exists(&stream_name, false).await?;
+        create_stream_if_not_exists(&stream_name, StreamType::UserDefined).await?;
 
         //flatten logs
         if let Some((_, log_source)) = req.headers().iter().find(|&(key, _)| key == LOG_SOURCE_KEY)
@@ -176,7 +179,8 @@ async fn flatten_and_push_logs(
 // fails if the logstream does not exist
 pub async fn post_event(req: HttpRequest, body: Bytes) -> Result<HttpResponse, PostError> {
     let stream_name: String = req.match_info().get("logstream").unwrap().parse().unwrap();
-    if stream_name.eq(INTERNAL_STREAM_NAME) {
+    let internal_stream_names = STREAM_INFO.list_internal_streams();
+    if internal_stream_names.contains(&stream_name) || stream_name == INTERNAL_STREAM_NAME {
         return Err(PostError::Invalid(anyhow::anyhow!(
             "Stream {} is an internal stream and cannot be ingested into",
             stream_name
@@ -202,6 +206,7 @@ pub async fn push_logs_unchecked(
         time_partition: None,
         is_first_event: true,                    // NOTE: Maybe should be false
         custom_partition_values: HashMap::new(), // should be an empty map for unchecked push
+        stream_type: StreamType::UserDefined,
     };
     unchecked_event.process_unchecked()?;
 
@@ -369,6 +374,7 @@ async fn create_process_record_batch(
         parsed_timestamp,
         time_partition: time_partition.clone(),
         custom_partition_values: custom_partition_values.clone(),
+        stream_type: StreamType::UserDefined,
     }
     .process()
     .await?;
@@ -413,7 +419,7 @@ fn into_event_batch(
 // Check if the stream exists and create a new stream if doesn't exist
 pub async fn create_stream_if_not_exists(
     stream_name: &str,
-    internal_stream: bool,
+    stream_type: StreamType,
 ) -> Result<(), PostError> {
     if STREAM_INFO.stream_exists(stream_name) {
         return Ok(());
@@ -427,7 +433,7 @@ pub async fn create_stream_if_not_exists(
                 "",
                 "",
                 Arc::new(Schema::empty()),
-                internal_stream,
+                stream_type,
             )
             .await?;
         }

server/src/handlers/http/logstream.rs

@@ -34,6 +34,7 @@ use crate::option::validation::bytes_to_human_size;
 use crate::option::{Mode, CONFIG};
 use crate::static_schema::{convert_static_schema_to_arrow_schema, StaticSchema};
 use crate::stats::{event_labels_date, storage_size_labels_date, Stats};
+use crate::storage::StreamType;
 use crate::storage::{retention::Retention, LogStream, StorageDir, StreamInfo};
 use crate::{
     catalog::{self, remove_manifest_from_snapshot},
@@ -377,7 +378,7 @@ async fn create_update_stream(
         &custom_partition,
         &static_schema_flag,
         schema,
-        false,
+        StreamType::UserDefined,
     )
     .await?;
 
@@ -835,11 +836,11 @@ pub async fn create_stream(
     custom_partition: &str,
     static_schema_flag: &str,
     schema: Arc<Schema>,
-    internal_stream: bool,
+    stream_type: StreamType,
 ) -> Result<(), CreateStreamError> {
     // fail to proceed if invalid stream name
-    if !internal_stream {
-        validator::stream_name(&stream_name)?;
+    if stream_type != StreamType::Internal {
+        validator::stream_name(&stream_name, &stream_type)?;
     }
     // Proceed to create log stream if it doesn't exist
     let storage = CONFIG.storage().get_object_store();
@@ -852,6 +853,7 @@ pub async fn create_stream(
             custom_partition,
             static_schema_flag,
             schema.clone(),
+            stream_type,
         )
         .await
     {
@@ -908,6 +910,7 @@ pub async fn get_stream_info(req: HttpRequest) -> Result<impl Responder, StreamE
         .ok_or(StreamError::StreamNotFound(stream_name.clone()))?;
 
     let stream_info: StreamInfo = StreamInfo {
+        stream_type: stream_meta.stream_type.clone(),
         created_at: stream_meta.created_at.clone(),
         first_event_at: stream_meta.first_event_at.clone(),
         time_partition: stream_meta.time_partition.clone(),

server/src/metadata.rs

@@ -36,7 +36,7 @@ use crate::metrics::{
 use crate::option::{Mode, CONFIG};
 use crate::storage::retention::Retention;
 use crate::storage::{
-    LogStream, ObjectStorage, ObjectStoreFormat, StorageDir, STREAM_METADATA_FILE_NAME,
+    LogStream, ObjectStorage, ObjectStoreFormat, StorageDir, StreamType, STREAM_METADATA_FILE_NAME,
     STREAM_ROOT_DIRECTORY,
 };
 use crate::utils::arrow::MergedRecordReader;
@@ -62,6 +62,7 @@ pub struct LogStreamMetadata {
     pub custom_partition: Option<String>,
     pub static_schema_flag: Option<String>,
     pub hot_tier_enabled: Option<bool>,
+    pub stream_type: StreamType,
 }
 
 // It is very unlikely that panic will occur when dealing with metadata.
@@ -341,6 +342,7 @@ impl StreamInfo {
             custom_partition: meta.custom_partition,
             static_schema_flag: meta.static_schema_flag,
             hot_tier_enabled: meta.hot_tier_enabled,
+            stream_type: meta.stream_type,
         };
 
         let mut map = self.write().expect(LOCK_EXPECT);
@@ -357,6 +359,15 @@ impl StreamInfo {
             .collect()
     }
 
+    pub fn list_internal_streams(&self) -> Vec<String> {
+        self.read()
+            .expect(LOCK_EXPECT)
+            .iter()
+            .filter(|(_, v)| v.stream_type != StreamType::UserDefined)
+            .map(|(k, _)| k.clone())
+            .collect()
+    }
+
     pub fn update_stats(
         &self,
         stream_name: &str,
@@ -471,6 +482,7 @@ pub async fn load_stream_metadata_on_server_start(
         custom_partition,
         static_schema_flag: meta.static_schema_flag.clone(),
         hot_tier_enabled: meta.hot_tier_enabled,
+        stream_type: meta.stream_type,
     };
 
     let mut map = STREAM_INFO.write().expect(LOCK_EXPECT);

server/src/migration.rs

@@ -144,6 +144,8 @@ async fn migration_stream(stream: &str, storage: &dyn ObjectStorage) -> anyhow::
         match version {
             Some("v1") => {
                 stream_metadata_value = stream_metadata_migration::v1_v4(stream_metadata_value);
+                stream_metadata_value =
+                    stream_metadata_migration::v4_v5(stream_metadata_value, stream);
                 storage
                     .put_object(&path, to_bytes(&stream_metadata_value))
                     .await?;
@@ -155,6 +157,8 @@ async fn migration_stream(stream: &str, storage: &dyn ObjectStorage) -> anyhow::
             }
             Some("v2") => {
                 stream_metadata_value = stream_metadata_migration::v2_v4(stream_metadata_value);
+                stream_metadata_value =
+                    stream_metadata_migration::v4_v5(stream_metadata_value, stream);
                 storage
                     .put_object(&path, to_bytes(&stream_metadata_value))
                     .await?;
@@ -167,6 +171,15 @@ async fn migration_stream(stream: &str, storage: &dyn ObjectStorage) -> anyhow::
             }
             Some("v3") => {
                 stream_metadata_value = stream_metadata_migration::v3_v4(stream_metadata_value);
+                stream_metadata_value =
+                    stream_metadata_migration::v4_v5(stream_metadata_value, stream);
+                storage
+                    .put_object(&path, to_bytes(&stream_metadata_value))
+                    .await?;
+            }
+            Some("v4") => {
+                stream_metadata_value =
+                    stream_metadata_migration::v4_v5(stream_metadata_value, stream);
                 storage
                     .put_object(&path, to_bytes(&stream_metadata_value))
                     .await?;