add system defined tags and free form labels to datasets

PUT /api/v1/logstream/{name} accepts X-P-Dataset-Tags and
X-P-Dataset-Labels headers (comma-separated) on stream creation
PUT /api/prism/v1/datasets/{name} - update tags and labels
GET /api/prism/v1/datasets/{name}/correlated - find datasets sharing tags or labels
GET /api/prism/v1/datasets/tags/{tag} - find all datasets with a specific tag
include tags and labels in home api response

Author: nikhilsinhaparseable

src/connectors/kafka/processor.rs

@@ -64,7 +64,8 @@ impl ParseableSinkProcessor {
                 vec![log_source_entry],
                 TelemetryType::default(),
                 tenant_id,
-                None,
+                vec![],
+                vec![],
             )
             .await?;
 

src/handlers/http/ingest.rs

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

src/handlers/http/mod.rs

@@ -32,6 +32,7 @@ pub mod about;
 pub mod alerts;
 pub mod cluster;
 pub mod correlation;
+pub mod datasets;
 pub mod demo_data;
 pub mod health_check;
 pub mod ingest;

src/handlers/http/modal/server.rs

@@ -200,14 +200,39 @@ impl Server {
     }
 
     pub fn get_prism_datasets() -> Scope {
-        web::scope("/datasets").route(
-            "",
-            web::post()
-                .to(http::prism_logstream::post_datasets)
-                .authorize_for_resource(Action::GetStreamInfo)
-                .authorize_for_resource(Action::GetStats)
-                .authorize_for_resource(Action::GetRetention),
-        )
+        web::scope("/datasets")
+            .route(
+                "",
+                web::post()
+                    .to(http::prism_logstream::post_datasets)
+                    .authorize_for_resource(Action::GetStreamInfo)
+                    .authorize_for_resource(Action::GetStats)
+                    .authorize_for_resource(Action::GetRetention),
+            )
+            .route(
+                "/tags/{tag}",
+                web::get()
+                    .to(http::datasets::get_datasets_by_tag)
+                    .authorize_for_resource(Action::GetStreamInfo),
+            )
+            .route(
+                "/{name}/correlated",
+                web::get()
+                    .to(http::datasets::get_correlated_datasets)
+                    .authorize_for_resource(Action::GetStreamInfo),
+            )
+            .route(
+                "/{name}/tags",
+                web::put()
+                    .to(http::datasets::put_dataset_tags)
+                    .authorize_for_resource(Action::CreateStream),
+            )
+            .route(
+                "/{name}/labels",
+                web::put()
+                    .to(http::datasets::put_dataset_labels)
+                    .authorize_for_resource(Action::CreateStream),
+            )
     }
 
     pub fn get_demo_data_webscope() -> Scope {

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

@@ -16,17 +16,18 @@
  *
  */
 
+use actix_web::http::header::HeaderMap;
+
 use crate::{
     event::format::LogSource,
     handlers::{
-        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,
+        CUSTOM_PARTITION_KEY, DATASET_LABELS_KEY, DATASET_TAG_KEY, DATASET_TAGS_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,
+        parse_dataset_labels, parse_dataset_tags,
     },
     storage::StreamType,
 };
-use actix_web::http::header::HeaderMap;
-use tracing::warn;
 
 #[derive(Debug, Default)]
 pub struct PutStreamHeaders {
@@ -38,7 +39,8 @@ pub struct PutStreamHeaders {
     pub stream_type: StreamType,
     pub log_source: LogSource,
     pub telemetry_type: TelemetryType,
-    pub dataset_tag: Option<DatasetTag>,
+    pub dataset_tags: Vec<DatasetTag>,
+    pub dataset_labels: Vec<String>,
 }
 
 impl From<&HeaderMap> for PutStreamHeaders {
@@ -72,16 +74,17 @@ 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)
+            dataset_tags: headers
+                .get(DATASET_TAGS_KEY)
+                .or_else(|| 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
-                    }
-                }),
+                .map(parse_dataset_tags)
+                .unwrap_or_default(),
+            dataset_labels: headers
+                .get(DATASET_LABELS_KEY)
+                .and_then(|v| v.to_str().ok())
+                .map(parse_dataset_labels)
+                .unwrap_or_default(),
         }
     }
 }

src/handlers/mod.rs

@@ -16,9 +16,11 @@
  *
  */
 
+use std::collections::HashSet;
 use std::fmt::Display;
 
 use serde::{Deserialize, Serialize};
+use tracing::warn;
 
 pub mod airplane;
 pub mod http;
@@ -36,6 +38,8 @@ 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";
+pub const DATASET_TAGS_KEY: &str = "x-p-dataset-tags";
+pub const DATASET_LABELS_KEY: &str = "x-p-dataset-labels";
 pub const TENANT_ID: &str = "x-p-tenant";
 const COOKIE_AGE_DAYS: usize = 7;
 const SESSION_COOKIE_NAME: &str = "session";
@@ -85,24 +89,28 @@ impl Display for TelemetryType {
 }
 
 /// Tag for categorizing datasets/streams by observability domain
-#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
+#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Hash)]
 #[serde(rename_all = "kebab-case")]
 pub enum DatasetTag {
-    AgentObservability,
-    K8sObservability,
+    AgentMonitoring,
+    K8sMonitoring,
     DatabaseObservability,
+    ApplicationMonitoring,
+    ServiceMap,
 }
 
 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),
+            "agent-monitoring" => Ok(DatasetTag::AgentMonitoring),
+            "k8s-monitoring" => Ok(DatasetTag::K8sMonitoring),
             "database-observability" => Ok(DatasetTag::DatabaseObservability),
+            "application-monitoring" => Ok(DatasetTag::ApplicationMonitoring),
+            "service-map" => Ok(DatasetTag::ServiceMap),
             _ => Err(
-                "Invalid dataset tag. Supported values: agent-observability, k8s-observability, database-observability",
+                "Invalid dataset tag. Supported values: agent-monitoring, k8s-monitoring, database-observability, application-monitoring, service-map",
             ),
         }
     }
@@ -111,9 +119,43 @@ impl TryFrom<&str> for DatasetTag {
 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::AgentMonitoring => "agent-monitoring",
+            DatasetTag::K8sMonitoring => "k8s-monitoring",
             DatasetTag::DatabaseObservability => "database-observability",
+            DatasetTag::ApplicationMonitoring => "application-monitoring",
+            DatasetTag::ServiceMap => "service-map",
         })
     }
 }
+
+pub fn parse_dataset_tags(header_value: &str) -> Vec<DatasetTag> {
+    header_value
+        .split(',')
+        .filter_map(|s| {
+            let trimmed = s.trim();
+            if trimmed.is_empty() {
+                None
+            } else {
+                match DatasetTag::try_from(trimmed) {
+                    Ok(tag) => Some(tag),
+                    Err(err) => {
+                        warn!("Invalid dataset tag '{trimmed}': {err}");
+                        None
+                    }
+                }
+            }
+        })
+        .collect::<HashSet<_>>()
+        .into_iter()
+        .collect()
+}
+
+pub fn parse_dataset_labels(header_value: &str) -> Vec<String> {
+    header_value
+        .split(',')
+        .map(|s| s.trim().to_string())
+        .filter(|s| !s.is_empty())
+        .collect::<HashSet<_>>()
+        .into_iter()
+        .collect()
+}

src/metadata.rs

@@ -93,7 +93,8 @@ pub struct LogStreamMetadata {
     pub stream_type: StreamType,
     pub log_source: Vec<LogSourceEntry>,
     pub telemetry_type: TelemetryType,
-    pub dataset_tag: Option<DatasetTag>,
+    pub dataset_tags: Vec<DatasetTag>,
+    pub dataset_labels: Vec<String>,
 }
 
 impl LogStreamMetadata {
@@ -109,7 +110,8 @@ impl LogStreamMetadata {
         schema_version: SchemaVersion,
         log_source: Vec<LogSourceEntry>,
         telemetry_type: TelemetryType,
-        dataset_tag: Option<DatasetTag>,
+        dataset_tags: Vec<DatasetTag>,
+        dataset_labels: Vec<String>,
     ) -> Self {
         LogStreamMetadata {
             created_at: if created_at.is_empty() {
@@ -134,7 +136,8 @@ impl LogStreamMetadata {
             schema_version,
             log_source,
             telemetry_type,
-            dataset_tag,
+            dataset_tags,
+            dataset_labels,
             ..Default::default()
         }
     }

src/migration/mod.rs

@@ -454,7 +454,8 @@ async fn setup_logstream_metadata(
         stream_type,
         log_source,
         telemetry_type,
-        dataset_tag,
+        dataset_tags,
+        dataset_labels,
         ..
     } = serde_json::from_value(stream_metadata_value).unwrap_or_default();
 
@@ -500,7 +501,8 @@ async fn setup_logstream_metadata(
         stream_type,
         log_source,
         telemetry_type,
-        dataset_tag,
+        dataset_tags,
+        dataset_labels,
     };
 
     Ok(metadata)

src/parseable/mod.rs

@@ -416,7 +416,8 @@ 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 dataset_tags = stream_metadata.dataset_tags;
+        let dataset_labels = stream_metadata.dataset_labels;
         let mut metadata = LogStreamMetadata::new(
             created_at,
             time_partition,
@@ -428,7 +429,8 @@ impl Parseable {
             schema_version,
             log_source,
             telemetry_type,
-            dataset_tag,
+            dataset_tags,
+            dataset_labels,
         );
 
         // Set hot tier fields from the stored metadata
@@ -533,7 +535,8 @@ impl Parseable {
         log_source: Vec<LogSourceEntry>,
         telemetry_type: TelemetryType,
         tenant_id: &Option<String>,
-        dataset_tag: Option<DatasetTag>,
+        dataset_tags: Vec<DatasetTag>,
+        dataset_labels: Vec<String>,
     ) -> Result<bool, PostError> {
         if self.streams.contains(stream_name, tenant_id) {
             return Ok(true);
@@ -566,7 +569,8 @@ impl Parseable {
             log_source,
             telemetry_type,
             tenant_id,
-            dataset_tag,
+            dataset_tags,
+            dataset_labels,
         )
         .await?;
 
@@ -643,7 +647,8 @@ impl Parseable {
             stream_type,
             log_source,
             telemetry_type,
-            dataset_tag,
+            dataset_tags,
+            dataset_labels,
         } = headers.into();
 
         let stream_in_memory_dont_update =
@@ -717,7 +722,8 @@ impl Parseable {
             vec![log_source_entry],
             telemetry_type,
             tenant_id,
-            dataset_tag,
+            dataset_tags,
+            dataset_labels,
         )
         .await?;
 
@@ -779,7 +785,8 @@ impl Parseable {
         log_source: Vec<LogSourceEntry>,
         telemetry_type: TelemetryType,
         tenant_id: &Option<String>,
-        dataset_tag: Option<DatasetTag>,
+        dataset_tags: Vec<DatasetTag>,
+        dataset_labels: Vec<String>,
     ) -> Result<(), CreateStreamError> {
         // fail to proceed if invalid stream name
         if stream_type != StreamType::Internal {
@@ -804,7 +811,8 @@ impl Parseable {
             },
             log_source: log_source.clone(),
             telemetry_type,
-            dataset_tag,
+            dataset_tags: dataset_tags.clone(),
+            dataset_labels: dataset_labels.clone(),
             ..Default::default()
         };
 
@@ -834,7 +842,8 @@ impl Parseable {
                     SchemaVersion::V1, // New stream
                     log_source,
                     telemetry_type,
-                    dataset_tag,
+                    dataset_tags,
+                    dataset_labels,
                 );
                 let ingestor_id = INGESTOR_META
                     .get()