update prism home api

Author: nikhilsinhaparseable

src/connectors/kafka/processor.rs

@@ -16,14 +16,6 @@
  *
  */
 
-use async_trait::async_trait;
-use futures_util::StreamExt;
-use rdkafka::consumer::{CommitMode, Consumer};
-use serde_json::Value;
-use std::collections::HashMap;
-use std::sync::Arc;
-use tokio_stream::wrappers::ReceiverStream;
-use tracing::{debug, error};
 use crate::{
     connectors::common::processor::Processor,
     event::{
@@ -34,6 +26,14 @@ use crate::{
     parseable::PARSEABLE,
     storage::StreamType,
 };
+use async_trait::async_trait;
+use futures_util::StreamExt;
+use rdkafka::consumer::{CommitMode, Consumer};
+use serde_json::Value;
+use std::collections::HashMap;
+use std::sync::Arc;
+use tokio_stream::wrappers::ReceiverStream;
+use tracing::{debug, error};
 
 use super::{ConsumerRecord, StreamConsumer, TopicPartition, config::BufferConfig};
 

src/prism/home/mod.rs

@@ -29,16 +29,19 @@ use tracing::error;
 use crate::{
     alerts::{ALERTS, AlertError, AlertsSummary, get_alerts_summary},
     correlation::{CORRELATIONS, CorrelationError},
-    event::format::LogSource,
-    handlers::http::{cluster::fetch_daily_stats, logstream::error::StreamError},
+    handlers::{
+        TelemetryType,
+        http::{cluster::fetch_daily_stats, logstream::error::StreamError},
+    },
     parseable::PARSEABLE,
     rbac::{Users, map::SessionKey, role::Action},
     stats::Stats,
     storage::{ObjectStorageError, ObjectStoreFormat, STREAM_ROOT_DIRECTORY, StreamType},
     users::{dashboards::DASHBOARDS, filters::FILTERS},
 };
 
-type StreamMetadataResponse = Result<(String, Vec<ObjectStoreFormat>, DataSetType), PrismHomeError>;
+type StreamMetadataResponse =
+    Result<(String, Vec<ObjectStoreFormat>, TelemetryType), PrismHomeError>;
 
 #[derive(Debug, Serialize, Default)]
 pub struct DatedStats {
@@ -48,17 +51,10 @@ pub struct DatedStats {
     storage: u64,
 }
 
-#[derive(Debug, Serialize)]
-enum DataSetType {
-    Logs,
-    Metrics,
-    Traces,
-}
-
 #[derive(Debug, Serialize)]
 pub struct DataSet {
     title: String,
-    dataset_type: DataSetType,
+    dataset_type: TelemetryType,
 }
 
 #[derive(Debug, Serialize)]
@@ -208,7 +204,7 @@ fn get_top_5_streams_by_ingestion(
 
 async fn get_stream_metadata(
     stream: String,
-) -> Result<(String, Vec<ObjectStoreFormat>, DataSetType), PrismHomeError> {
+) -> Result<(String, Vec<ObjectStoreFormat>, TelemetryType), PrismHomeError> {
     let path = RelativePathBuf::from_iter([&stream, STREAM_ROOT_DIRECTORY]);
     let obs = PARSEABLE
         .storage
@@ -237,18 +233,7 @@ async fn get_stream_metadata(
         )));
     }
 
-    // let log_source = &stream_jsons[0].clone().log_source;
-    let log_source_format = stream_jsons
-        .iter()
-        .find(|sj| !sj.log_source.is_empty())
-        .map(|sj| sj.log_source[0].log_source_format.clone())
-        .unwrap_or_default();
-
-    let dataset_type = match log_source_format {
-        LogSource::OtelMetrics => DataSetType::Metrics,
-        LogSource::OtelTraces => DataSetType::Traces,
-        _ => DataSetType::Logs,
-    };
+    let dataset_type = stream_jsons[0].telemetry_type;
 
     Ok((stream, stream_jsons, dataset_type))
 }