update for distributed

Author: nikhilsinhaparseable

src/handlers/http/logstream.rs

@@ -309,6 +309,14 @@ pub async fn get_stream_info(stream_name: Path<String>) -> Result<impl Responder
             None
         };
 
+    let stream_log_source = storage
+        .get_log_source_from_storage(&stream_name)
+        .await
+        .unwrap_or_default();
+    PARSEABLE
+        .update_log_source(&stream_name, stream_log_source)
+        .await?;
+
     let hash_map = PARSEABLE.streams.read().unwrap();
     let stream_meta = hash_map
         .get(&stream_name)

src/parseable/mod.rs

@@ -822,6 +822,25 @@ impl Parseable {
 
         Some(first_event_at.to_string())
     }
+
+    pub async fn update_log_source(
+        &self,
+        stream_name: &str,
+        log_source: Vec<LogSourceEntry>,
+    ) -> Result<(), StreamError> {
+        let storage = self.storage.get_object_store();
+        if let Err(err) = storage
+            .update_log_source_in_stream(stream_name, &log_source)
+            .await
+        {
+            return Err(StreamError::Storage(err));
+        }
+
+        let stream = self.get_stream(stream_name).expect(STREAM_EXISTS);
+        stream.set_log_source(log_source);
+
+        Ok(())
+    }
 }
 
 pub fn validate_static_schema(

src/storage/object_storage.rs

@@ -687,6 +687,41 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static {
         Ok(stream_metas)
     }
 
+    async fn get_log_source_from_storage(
+        &self,
+        stream_name: &str,
+    ) -> Result<Vec<LogSourceEntry>, ObjectStorageError> {
+        let mut all_log_sources: Vec<LogSourceEntry> = Vec::new();
+        let stream_metas = self.get_stream_meta_from_storage(stream_name).await;
+        if let Ok(stream_metas) = stream_metas {
+            for stream_meta in stream_metas.iter() {
+                // fetch unique log sources and their fields
+                all_log_sources.extend(stream_meta.log_source.clone());
+            }
+        }
+
+        //merge fields of same log source
+        let mut merged_log_sources: Vec<LogSourceEntry> = Vec::new();
+        let mut log_source_map: HashMap<LogSource, HashSet<String>> = HashMap::new();
+        for log_source_entry in all_log_sources {
+            let log_source_format = log_source_entry.log_source_format;
+            let fields = log_source_entry.fields;
+
+            log_source_map
+                .entry(log_source_format)
+                .or_default()
+                .extend(fields);
+        }
+
+        for (log_source_format, fields) in log_source_map {
+            merged_log_sources.push(LogSourceEntry {
+                log_source_format,
+                fields: fields.into_iter().collect(),
+            });
+        }
+        Ok(merged_log_sources)
+    }
+
     /// Retrieves the earliest first-event-at from the storage for the specified stream.
     ///
     /// This function fetches the object-store format from all the stream.json files for the given stream from the storage,