include null, add query param in prism home api

nikhilsinhaparseable · nikhilsinhaparseable · commit 5ed9fd30f7e9 · 2025-06-16T02:48:28.000-04:00
diff --git a/src/handlers/http/prism_home.rs b/src/handlers/http/prism_home.rs
@@ -26,7 +26,7 @@ use crate::{
 };
 
 const HOME_SEARCH_QUERY_PARAM: &str = "key";
-
+const HOME_QUERY_PARAM: &str = "includeInternal";
 /// Fetches the data to populate Prism's home
 ///
 ///
@@ -36,8 +36,14 @@ const HOME_SEARCH_QUERY_PARAM: &str = "key";
 pub async fn home_api(req: HttpRequest) -> Result<impl Responder, PrismHomeError> {
     let key = extract_session_key_from_req(&req)
         .map_err(|err| PrismHomeError::Anyhow(anyhow::Error::msg(err.to_string())))?;
+    let query_map = web::Query::<HashMap<String, String>>::from_query(req.query_string())
+        .map_err(|_| PrismHomeError::InvalidQueryParameter(HOME_QUERY_PARAM.to_string()))?;
 
-    let res = generate_home_response(&key).await?;
+    let include_internal = query_map
+        .get(HOME_QUERY_PARAM)
+        .map_or(false, |v| v == "true");
+
+    let res = generate_home_response(&key, include_internal).await?;
 
     Ok(web::Json(res))
 }
@@ -52,11 +58,12 @@ pub async fn home_search(req: HttpRequest) -> Result<impl Responder, PrismHomeEr
         return Ok(web::Json(serde_json::json!({})));
     }
 
-    let query_value = query_map
+    let query_key = query_map
         .get(HOME_SEARCH_QUERY_PARAM)
         .ok_or_else(|| PrismHomeError::InvalidQueryParameter(HOME_SEARCH_QUERY_PARAM.to_string()))?
         .to_lowercase();
-    let res = generate_home_search_response(&key, &query_value).await?;
+
+    let res = generate_home_search_response(&key, &query_key).await?;
     let json_res = serde_json::to_value(res)
         .map_err(|err| PrismHomeError::Anyhow(anyhow::Error::msg(err.to_string())))?;
 
diff --git a/src/otel/traces.rs b/src/otel/traces.rs
@@ -336,7 +336,6 @@ fn flatten_span_record(span_record: &Span) -> Vec<Map<String, Value>> {
     span_records_json
 }
 
-
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -360,13 +359,21 @@ mod tests {
             KeyValue {
                 key: "service.name".to_string(),
                 value: Some(AnyValue {
-                    value: Some(opentelemetry_proto::tonic::common::v1::any_value::Value::StringValue("test-service".to_string())),
+                    value: Some(
+                        opentelemetry_proto::tonic::common::v1::any_value::Value::StringValue(
+                            "test-service".to_string(),
+                        ),
+                    ),
                 }),
             },
             KeyValue {
                 key: "http.method".to_string(),
                 value: Some(AnyValue {
-                    value: Some(opentelemetry_proto::tonic::common::v1::any_value::Value::StringValue("GET".to_string())),
+                    value: Some(
+                        opentelemetry_proto::tonic::common::v1::any_value::Value::StringValue(
+                            "GET".to_string(),
+                        ),
+                    ),
                 }),
             },
         ]
@@ -398,7 +405,8 @@ mod tests {
             assert_eq!(
                 result.get("span_status_description").unwrap(),
                 &Value::String(expected_description.to_string()),
-                "Status description should match expected value for code {}", code
+                "Status description should match expected value for code {}",
+                code
             );
             assert_eq!(
                 result.get("span_status_message").unwrap(),
@@ -432,7 +440,8 @@ mod tests {
             assert_eq!(
                 result.get("span_kind_description").unwrap(),
                 &Value::String(expected_description.to_string()),
-                "Span kind description should match expected value for kind {}", kind
+                "Span kind description should match expected value for kind {}",
+                kind
             );
         }
     }
@@ -459,7 +468,8 @@ mod tests {
             assert_eq!(
                 result.get("span_flags_description").unwrap(),
                 &Value::String(expected_description.to_string()),
-                "Span flags description should match expected value for flags {}", flags
+                "Span flags description should match expected value for flags {}",
+                flags
             );
         }
     }
@@ -488,7 +498,10 @@ mod tests {
 
         // Check first event
         let first_event = &result[0];
-        assert!(first_event.contains_key("event_time_unix_nano"), "Should contain timestamp");
+        assert!(
+            first_event.contains_key("event_time_unix_nano"),
+            "Should contain timestamp"
+        );
         assert_eq!(
             first_event.get("event_name").unwrap(),
             &Value::String("request.start".to_string()),
@@ -499,7 +512,10 @@ mod tests {
             &Value::Number(2.into()),
             "Dropped attributes count should be preserved"
         );
-        assert!(first_event.contains_key("service.name"), "Should contain flattened attributes");
+        assert!(
+            first_event.contains_key("service.name"),
+            "Should contain flattened attributes"
+        );
 
         // Check second event
         let second_event = &result[1];
@@ -518,16 +534,14 @@ mod tests {
     #[test]
     fn test_flatten_links_structure() {
         // Test that links are properly flattened with all expected fields
-        let links = vec![
-            Link {
-                trace_id: sample_trace_id(),
-                span_id: sample_span_id(),
-                trace_state: "state1".to_string(),
-                attributes: sample_attributes(),
-                dropped_attributes_count: 1,
-                flags: 0,
-            },
-        ];
+        let links = vec![Link {
+            trace_id: sample_trace_id(),
+            span_id: sample_span_id(),
+            trace_state: "state1".to_string(),
+            attributes: sample_attributes(),
+            dropped_attributes_count: 1,
+            flags: 0,
+        }];
 
         let result = flatten_links(&links);
 
@@ -549,7 +563,10 @@ mod tests {
             &Value::Number(1.into()),
             "Dropped attributes count should be preserved"
         );
-        assert!(link.contains_key("service.name"), "Should contain flattened attributes");
+        assert!(
+            link.contains_key("service.name"),
+            "Should contain flattened attributes"
+        );
     }
 
     #[test]
@@ -611,12 +628,30 @@ mod tests {
                 &Value::String("SPAN_KIND_SERVER".to_string()),
                 "All records should contain span kind description"
             );
-            assert!(record.contains_key("span_trace_id"), "Should contain trace ID");
-            assert!(record.contains_key("span_span_id"), "Should contain span ID");
-            assert!(record.contains_key("span_start_time_unix_nano"), "Should contain start time");
-            assert!(record.contains_key("span_end_time_unix_nano"), "Should contain end time");
-            assert!(record.contains_key("service.name"), "Should contain span attributes");
-            assert!(record.contains_key("span_status_code"), "Should contain status");
+            assert!(
+                record.contains_key("span_trace_id"),
+                "Should contain trace ID"
+            );
+            assert!(
+                record.contains_key("span_span_id"),
+                "Should contain span ID"
+            );
+            assert!(
+                record.contains_key("span_start_time_unix_nano"),
+                "Should contain start time"
+            );
+            assert!(
+                record.contains_key("span_end_time_unix_nano"),
+                "Should contain end time"
+            );
+            assert!(
+                record.contains_key("service.name"),
+                "Should contain span attributes"
+            );
+            assert!(
+                record.contains_key("span_status_code"),
+                "Should contain status"
+            );
         }
 
         // One record should be an event, one should be a link
@@ -650,17 +685,30 @@ mod tests {
 
         let result = flatten_span_record(&span);
 
-        assert_eq!(result.len(), 1, "Should have exactly one record for span without events/links");
+        assert_eq!(
+            result.len(),
+            1,
+            "Should have exactly one record for span without events/links"
+        );
 
         let record = &result[0];
         assert_eq!(
             record.get("span_name").unwrap(),
             &Value::String("simple-span".to_string()),
             "Should contain span name"
         );
-        assert!(!record.contains_key("event_name"), "Should not contain event fields");
-        assert!(!record.contains_key("link_trace_id"), "Should not contain link fields");
-        assert!(!record.contains_key("span_status_code"), "Should not contain status when none provided");
+        assert!(
+            !record.contains_key("event_name"),
+            "Should not contain event fields"
+        );
+        assert!(
+            !record.contains_key("link_trace_id"),
+            "Should not contain link fields"
+        );
+        assert!(
+            !record.contains_key("span_status_code"),
+            "Should not contain status when none provided"
+        );
     }
 
     #[test]
@@ -705,10 +753,16 @@ mod tests {
                 assert_eq!(hex_span_id, "12345678", "Span ID should be lowercase hex");
             }
             if let Some(Value::String(hex_parent_span_id)) = record.get("span_parent_span_id") {
-                assert_eq!(hex_parent_span_id, "87654321", "Parent span ID should be lowercase hex");
+                assert_eq!(
+                    hex_parent_span_id, "87654321",
+                    "Parent span ID should be lowercase hex"
+                );
             }
             if let Some(Value::String(link_trace_id)) = record.get("link_trace_id") {
-                assert_eq!(link_trace_id, "ffabcdef", "Link trace ID should be lowercase hex");
+                assert_eq!(
+                    link_trace_id, "ffabcdef",
+                    "Link trace ID should be lowercase hex"
+                );
             }
         }
     }
@@ -823,15 +877,36 @@ mod tests {
     fn test_known_field_list_completeness() {
         // Test that the OTEL_TRACES_KNOWN_FIELD_LIST contains all expected fields
         let expected_fields = [
-            "scope_name", "scope_version", "scope_schema_url", "scope_dropped_attributes_count",
-            "resource_schema_url", "resource_dropped_attributes_count",
-            "span_trace_id", "span_span_id", "span_name", "span_parent_span_id", "name",
-            "span_kind", "span_kind_description", "span_start_time_unix_nano", "span_end_time_unix_nano",
-            "event_name", "event_time_unix_nano", "event_dropped_attributes_count",
-            "link_span_id", "link_trace_id", "link_dropped_attributes_count",
-            "span_dropped_events_count", "span_dropped_links_count", "span_dropped_attributes_count",
-            "span_trace_state", "span_flags", "span_flags_description",
-            "span_status_code", "span_status_description", "span_status_message",
+            "scope_name",
+            "scope_version",
+            "scope_schema_url",
+            "scope_dropped_attributes_count",
+            "resource_schema_url",
+            "resource_dropped_attributes_count",
+            "span_trace_id",
+            "span_span_id",
+            "span_name",
+            "span_parent_span_id",
+            "name",
+            "span_kind",
+            "span_kind_description",
+            "span_start_time_unix_nano",
+            "span_end_time_unix_nano",
+            "event_name",
+            "event_time_unix_nano",
+            "event_dropped_attributes_count",
+            "link_span_id",
+            "link_trace_id",
+            "link_dropped_attributes_count",
+            "span_dropped_events_count",
+            "span_dropped_links_count",
+            "span_dropped_attributes_count",
+            "span_trace_state",
+            "span_flags",
+            "span_flags_description",
+            "span_status_code",
+            "span_status_description",
+            "span_status_message",
         ];
 
         assert_eq!(
diff --git a/src/parseable/streams.rs b/src/parseable/streams.rs
@@ -52,10 +52,7 @@ use crate::{
         format::{LogSource, LogSourceEntry},
         DEFAULT_TIMESTAMP_KEY,
     },
-    handlers::http::{
-        cluster::INTERNAL_STREAM_NAME, ingest::PostError,
-        modal::utils::ingest_utils::flatten_and_push_logs,
-    },
+    handlers::http::{ingest::PostError, modal::utils::ingest_utils::flatten_and_push_logs},
     metadata::{LogStreamMetadata, SchemaVersion},
     metrics,
     option::Mode,
@@ -73,7 +70,7 @@ use super::{
     },
     LogStream, ARROW_FILE_EXTENSION,
 };
-
+const DATASET_STATS_STREAM_NAME: &str = "pstats";
 const MAX_CONCURRENT_FIELD_STATS: usize = 10;
 
 #[derive(Serialize, Debug)]
@@ -790,11 +787,10 @@ impl Stream {
         record_batches: Vec<RecordBatch>,
         schema: Arc<Schema>,
     ) -> Result<(), PostError> {
-        let stats_dataset_name = format!("dataset_{INTERNAL_STREAM_NAME}");
         let log_source_entry = LogSourceEntry::new(LogSource::Json, HashSet::new());
         PARSEABLE
             .create_stream_if_not_exists(
-                &stats_dataset_name,
+                DATASET_STATS_STREAM_NAME,
                 StreamType::Internal,
                 vec![log_source_entry],
             )
@@ -818,7 +814,7 @@ impl Stream {
 
         flatten_and_push_logs(
             stats_value,
-            &stats_dataset_name,
+            DATASET_STATS_STREAM_NAME,
             &LogSource::Json,
             &HashMap::new(),
         )
@@ -946,7 +942,7 @@ impl Stream {
         field_name: &str,
     ) -> Vec<DistinctStat> {
         let sql = format!(
-        "select count(*) as distinct_count, \"{field_name}\" from \"{stream_name}\" where \"{field_name}\" is not null group by \"{field_name}\" order by distinct_count desc limit {}",
+        "select count(*) as distinct_count, \"{field_name}\" from \"{stream_name}\" group by \"{field_name}\" order by distinct_count desc limit {}",
         PARSEABLE.options.max_field_statistics
     );
         let mut distinct_stats = Vec::new();
diff --git a/src/prism/home/mod.rs b/src/prism/home/mod.rs
@@ -33,7 +33,7 @@ use crate::{
     handlers::http::{cluster::fetch_daily_stats, logstream::error::StreamError},
     parseable::PARSEABLE,
     rbac::{map::SessionKey, role::Action, Users},
-    storage::{ObjectStorageError, ObjectStoreFormat, STREAM_ROOT_DIRECTORY},
+    storage::{ObjectStorageError, ObjectStoreFormat, StreamType, STREAM_ROOT_DIRECTORY},
     users::{dashboards::DASHBOARDS, filters::FILTERS},
 };
 
@@ -88,7 +88,10 @@ pub struct HomeSearchResponse {
     resources: Vec<Resource>,
 }
 
-pub async fn generate_home_response(key: &SessionKey) -> Result<HomeResponse, PrismHomeError> {
+pub async fn generate_home_response(
+    key: &SessionKey,
+    include_internal: bool,
+) -> Result<HomeResponse, PrismHomeError> {
     // Execute these operations concurrently
     let (stream_titles_result, alerts_info_result) =
         tokio::join!(get_stream_titles(key), get_alerts_info());
@@ -120,6 +123,14 @@ pub async fn generate_home_response(key: &SessionKey) -> Result<HomeResponse, Pr
     for result in stream_metadata_results {
         match result {
             Ok((stream, metadata, dataset_type)) => {
+                // Skip internal streams if the flag is false
+                if !include_internal
+                    && metadata
+                        .iter()
+                        .any(|m| m.stream_type == StreamType::Internal)
+                {
+                    continue;
+                }
                 stream_wise_stream_json.insert(stream.clone(), metadata);
                 datasets.push(DataSet {
                     title: stream,
