Merge branch 'main' into multi-tenancy

Author: parmesant

src/handlers/http/ingest.rs

@@ -120,6 +120,7 @@ pub async fn ingest(
             vec![log_source_entry.clone()],
             telemetry_type,
             &tenant_id,
+            None
         )
         .await
         .map_err(|e| {
@@ -238,6 +239,7 @@ pub async fn setup_otel_stream(
             vec![log_source_entry.clone()],
             telemetry_type,
             &tenant_id,
+            None
         )
         .await?;
     let mut time_partition = None;

src/handlers/http/modal/utils/logstream_utils.rs

@@ -19,13 +19,14 @@
 use crate::{
     event::format::LogSource,
     handlers::{
-        CUSTOM_PARTITION_KEY, LOG_SOURCE_KEY, STATIC_SCHEMA_FLAG, STREAM_TYPE_KEY,
-        TELEMETRY_TYPE_KEY, TIME_PARTITION_KEY, TIME_PARTITION_LIMIT_KEY, TelemetryType,
-        UPDATE_STREAM_KEY,
+        CUSTOM_PARTITION_KEY, DATASET_TAG_KEY, DatasetTag, LOG_SOURCE_KEY, STATIC_SCHEMA_FLAG,
+        STREAM_TYPE_KEY, TELEMETRY_TYPE_KEY, TIME_PARTITION_KEY, TIME_PARTITION_LIMIT_KEY,
+        TelemetryType, UPDATE_STREAM_KEY,
     },
     storage::StreamType,
 };
 use actix_web::http::header::HeaderMap;
+use tracing::warn;
 
 #[derive(Debug, Default)]
 pub struct PutStreamHeaders {
@@ -37,6 +38,7 @@ pub struct PutStreamHeaders {
     pub stream_type: StreamType,
     pub log_source: LogSource,
     pub telemetry_type: TelemetryType,
+    pub dataset_tag: Option<DatasetTag>,
 }
 
 impl From<&HeaderMap> for PutStreamHeaders {
@@ -70,6 +72,16 @@ impl From<&HeaderMap> for PutStreamHeaders {
                 .get(TELEMETRY_TYPE_KEY)
                 .and_then(|v| v.to_str().ok())
                 .map_or(TelemetryType::Logs, TelemetryType::from),
+            dataset_tag: headers
+                .get(DATASET_TAG_KEY)
+                .and_then(|v| v.to_str().ok())
+                .and_then(|v| match DatasetTag::try_from(v) {
+                    Ok(tag) => Some(tag),
+                    Err(err) => {
+                        warn!("Invalid dataset tag '{v}': {err}");
+                        None
+                    }
+                }),
         }
     }
 }

src/handlers/mod.rs

@@ -35,6 +35,7 @@ pub const AUTHORIZATION_KEY: &str = "authorization";
 pub const UPDATE_STREAM_KEY: &str = "x-p-update-stream";
 pub const STREAM_TYPE_KEY: &str = "x-p-stream-type";
 pub const TELEMETRY_TYPE_KEY: &str = "x-p-telemetry-type";
+pub const DATASET_TAG_KEY: &str = "x-p-dataset-tag";
 const COOKIE_AGE_DAYS: usize = 7;
 const SESSION_COOKIE_NAME: &str = "session";
 const USER_COOKIE_NAME: &str = "username";
@@ -81,3 +82,37 @@ impl Display for TelemetryType {
         })
     }
 }
+
+/// Tag for categorizing datasets/streams by observability domain
+#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "kebab-case")]
+pub enum DatasetTag {
+    AgentObservability,
+    K8sObservability,
+    DatabaseObservability,
+}
+
+impl TryFrom<&str> for DatasetTag {
+    type Error = &'static str;
+
+    fn try_from(s: &str) -> Result<Self, Self::Error> {
+        match s.to_lowercase().as_str() {
+            "agent-observability" => Ok(DatasetTag::AgentObservability),
+            "k8s-observability" => Ok(DatasetTag::K8sObservability),
+            "database-observability" => Ok(DatasetTag::DatabaseObservability),
+            _ => Err(
+                "Invalid dataset tag. Supported values: agent-observability, k8s-observability, database-observability",
+            ),
+        }
+    }
+}
+
+impl Display for DatasetTag {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str(match self {
+            DatasetTag::AgentObservability => "agent-observability",
+            DatasetTag::K8sObservability => "k8s-observability",
+            DatasetTag::DatabaseObservability => "database-observability",
+        })
+    }
+}

src/metadata.rs

@@ -25,7 +25,7 @@ use std::sync::Arc;
 
 use crate::catalog::snapshot::ManifestItem;
 use crate::event::format::LogSourceEntry;
-use crate::handlers::TelemetryType;
+use crate::handlers::{DatasetTag, TelemetryType};
 use crate::hottier::StreamHotTier;
 use crate::metrics::{
     EVENTS_INGESTED, EVENTS_INGESTED_DATE, EVENTS_INGESTED_SIZE, EVENTS_INGESTED_SIZE_DATE,
@@ -93,6 +93,7 @@ pub struct LogStreamMetadata {
     pub stream_type: StreamType,
     pub log_source: Vec<LogSourceEntry>,
     pub telemetry_type: TelemetryType,
+    pub dataset_tag: Option<DatasetTag>,
 }
 
 impl LogStreamMetadata {
@@ -108,6 +109,7 @@ impl LogStreamMetadata {
         schema_version: SchemaVersion,
         log_source: Vec<LogSourceEntry>,
         telemetry_type: TelemetryType,
+        dataset_tag: Option<DatasetTag>,
     ) -> Self {
         LogStreamMetadata {
             created_at: if created_at.is_empty() {
@@ -132,6 +134,7 @@ impl LogStreamMetadata {
             schema_version,
             log_source,
             telemetry_type,
+            dataset_tag,
             ..Default::default()
         }
     }

src/migration/mod.rs

@@ -418,6 +418,7 @@ async fn setup_logstream_metadata(
         stream_type,
         log_source,
         telemetry_type,
+        dataset_tag,
         ..
     } = serde_json::from_value(stream_metadata_value).unwrap_or_default();
 
@@ -463,6 +464,7 @@ async fn setup_logstream_metadata(
         stream_type,
         log_source,
         telemetry_type,
+        dataset_tag,
     };
 
     Ok(metadata)

src/parseable/mod.rs

@@ -52,7 +52,7 @@ use crate::{
         format::{LogSource, LogSourceEntry},
     },
     handlers::{
-        STREAM_TYPE_KEY, TelemetryType,
+        DatasetTag, STREAM_TYPE_KEY, TelemetryType,
         http::{
             cluster::{PMETA_STREAM_NAME, sync_streams_with_ingestors},
             ingest::PostError,
@@ -415,6 +415,7 @@ impl Parseable {
         let schema_version = stream_metadata.schema_version;
         let log_source = stream_metadata.log_source;
         let telemetry_type = stream_metadata.telemetry_type;
+        let dataset_tag = stream_metadata.dataset_tag;
         let mut metadata = LogStreamMetadata::new(
             created_at,
             time_partition,
@@ -426,6 +427,7 @@ impl Parseable {
             schema_version,
             log_source,
             telemetry_type,
+            dataset_tag,
         );
 
         // Set hot tier fields from the stored metadata
@@ -471,6 +473,7 @@ impl Parseable {
                     vec![log_source_entry.clone()],
                     TelemetryType::Logs,
                     &tenant_id,
+                    None
                 )
                 .await;
 
@@ -517,6 +520,7 @@ impl Parseable {
         log_source: Vec<LogSourceEntry>,
         telemetry_type: TelemetryType,
         tenant_id: &Option<String>,
+        dataset_tag: Option<DatasetTag>,
     ) -> Result<bool, PostError> {
         if self.streams.contains(stream_name, tenant_id) {
             return Ok(true);
@@ -549,6 +553,7 @@ impl Parseable {
             log_source,
             telemetry_type,
             tenant_id,
+            dataset_tag,
         )
         .await?;
 
@@ -625,6 +630,7 @@ impl Parseable {
             stream_type,
             log_source,
             telemetry_type,
+            dataset_tag,
         } = headers.into();
 
         let stream_in_memory_dont_update =
@@ -698,6 +704,7 @@ impl Parseable {
             vec![log_source_entry],
             telemetry_type,
             tenant_id,
+            dataset_tag,
         )
         .await?;
 
@@ -759,6 +766,7 @@ impl Parseable {
         log_source: Vec<LogSourceEntry>,
         telemetry_type: TelemetryType,
         tenant_id: &Option<String>,
+        dataset_tag: Option<DatasetTag>,
     ) -> Result<(), CreateStreamError> {
         // fail to proceed if invalid stream name
         if stream_type != StreamType::Internal {
@@ -783,6 +791,7 @@ impl Parseable {
             },
             log_source: log_source.clone(),
             telemetry_type,
+            dataset_tag,
             ..Default::default()
         };
 
@@ -812,6 +821,7 @@ impl Parseable {
                     SchemaVersion::V1, // New stream
                     log_source,
                     telemetry_type,
+                    dataset_tag,
                 );
                 let ingestor_id = INGESTOR_META
                     .get()