remove node metadata from parseable struct, add as OnceCell

Author: nikhilsinhaparseable

src/handlers/http/cluster/mod.rs

@@ -581,25 +581,16 @@ pub async fn get_cluster_info() -> Result<impl Responder, StreamError> {
         })
         .map_err(|err| StreamError::Anyhow(err.into()))?;
 
-    // Get self metadata
-    let self_metadata = if let Some(metadata) = PARSEABLE.get_metadata() {
-        vec![metadata]
-    } else {
-        vec![]
-    };
-
     // Fetch info for all nodes concurrently
-    let (querier_infos, ingestor_infos, indexer_infos, self_info) = future::join4(
+    let (querier_infos, ingestor_infos, indexer_infos) = future::join3(
         fetch_nodes_info(querier_metadata),
         fetch_nodes_info(ingestor_metadata),
         fetch_nodes_info(indexer_metadata),
-        fetch_nodes_info(self_metadata),
     )
     .await;
 
     // Combine results from all node types
     let mut infos = Vec::new();
-    infos.extend(self_info?);
     infos.extend(querier_infos?);
     infos.extend(ingestor_infos?);
     infos.extend(indexer_infos?);
@@ -903,12 +894,6 @@ async fn fetch_cluster_metrics() -> Result<Vec<Metrics>, PostError> {
     )
     .await;
 
-    let self_metadata = if let Some(metadata) = PARSEABLE.get_metadata() {
-        vec![metadata]
-    } else {
-        vec![]
-    };
-
     // Handle querier metadata result
     let querier_metadata: Vec<NodeMetadata> = querier_result.map_err(|err| {
         error!("Fatal: failed to get querier info: {:?}", err);
@@ -925,8 +910,7 @@ async fn fetch_cluster_metrics() -> Result<Vec<Metrics>, PostError> {
         PostError::Invalid(err)
     })?;
     // Fetch metrics from ingestors and indexers concurrently
-    let (self_metrics, querier_metrics, ingestor_metrics, indexer_metrics) = future::join4(
-        fetch_nodes_metrics(self_metadata),
+    let (querier_metrics, ingestor_metrics, indexer_metrics) = future::join3(
         fetch_nodes_metrics(querier_metadata),
         fetch_nodes_metrics(ingestor_metadata),
         fetch_nodes_metrics(indexer_metadata),
@@ -936,12 +920,6 @@ async fn fetch_cluster_metrics() -> Result<Vec<Metrics>, PostError> {
     // Combine all metrics
     let mut all_metrics = Vec::new();
 
-    // Add self metrics
-    match self_metrics {
-        Ok(metrics) => all_metrics.extend(metrics),
-        Err(err) => return Err(err),
-    }
-
     // Add querier metrics
     match querier_metrics {
         Ok(metrics) => all_metrics.extend(metrics),

src/handlers/http/modal/ingest_server.rs

@@ -27,8 +27,9 @@ use bytes::Bytes;
 use relative_path::RelativePathBuf;
 use serde_json::Value;
 use tokio::sync::oneshot;
+use tokio::sync::OnceCell;
 
-use crate::option::Mode;
+use crate::handlers::http::modal::NodeType;
 use crate::{
     analytics,
     handlers::{
@@ -46,13 +47,14 @@ use crate::{
     sync, Server,
 };
 
+use super::IngestorMetadata;
 use super::{
     ingest::{ingestor_logstream, ingestor_rbac, ingestor_role},
     OpenIdClient, ParseableServer,
 };
 
 pub const INGESTOR_EXPECT: &str = "Ingestor Metadata should be set in ingestor mode";
-
+pub static INGESTOR_META: OnceCell<IngestorMetadata> = OnceCell::const_new();
 pub struct IngestServer;
 
 #[async_trait]
@@ -98,6 +100,15 @@ impl ParseableServer for IngestServer {
         prometheus: &PrometheusMetrics,
         shutdown_rx: oneshot::Receiver<()>,
     ) -> anyhow::Result<()> {
+        // write the ingestor metadata to storage
+        INGESTOR_META
+            .get_or_init(|| async {
+                IngestorMetadata::load_node_metadata(NodeType::Ingestor)
+                    .await
+                    .expect("Ingestor Metadata should be set in ingestor mode")
+            })
+            .await;
+
         PARSEABLE.storage.register_store_metrics(prometheus);
 
         migration::run_migration(&PARSEABLE).await?;
@@ -108,9 +119,6 @@ impl ParseableServer for IngestServer {
 
         tokio::spawn(airplane::server());
 
-        // write the ingestor metadata to storage
-        PARSEABLE.store_metadata(Mode::Ingest).await?;
-
         // Ingestors shouldn't have to deal with OpenId auth flow
         let result = self.start(shutdown_rx, prometheus.clone(), None).await;
         // Cancel sync jobs

src/handlers/http/modal/mod.rs

@@ -277,28 +277,92 @@ impl NodeMetadata {
         }
     }
 
-    /// Capture metadata information by either loading it from staging or starting fresh
-    pub fn load(
-        options: &Options,
-        storage: &dyn ObjectStorageProvider,
-        node_type: NodeType,
-    ) -> Arc<Self> {
-        let staging_path = options.staging_dir();
+    pub async fn load_node_metadata(node_type: NodeType) -> anyhow::Result<Self> {
+        let staging_path = PARSEABLE.options.staging_dir();
         let node_type_str = node_type.as_str();
 
         // Attempt to load metadata from staging
-        if let Some(mut meta) = Self::load_from_staging(staging_path, node_type_str, options) {
-            Self::update_metadata(&mut meta, options, node_type);
-            meta.put_on_disk(staging_path)
-                .expect("Couldn't write updated metadata to disk");
-            return Arc::new(meta);
+        if let Some(meta) = Self::load_from_staging(staging_path, node_type_str, &PARSEABLE.options)
+        {
+            return Self::process_and_store_metadata(meta, staging_path, node_type).await;
+        }
+
+        // Attempt to load metadata from storage
+        let storage_metas = Self::load_from_storage(node_type_str.to_string()).await;
+        let url = PARSEABLE.options.get_url(node_type.to_mode());
+        let port = url.port().unwrap_or(80).to_string();
+        let url = url.to_string();
+
+        for storage_meta in storage_metas {
+            if storage_meta.domain_name == url && storage_meta.port == port {
+                return Self::process_and_store_metadata(storage_meta, staging_path, node_type)
+                    .await;
+            }
         }
 
-        // If no metadata is found in staging, create a new one
-        let meta = Self::create_new_metadata(options, storage, node_type);
+        // If no metadata is found, create a new one
+        let meta = Self::create_new_metadata(&PARSEABLE.options, &*PARSEABLE.storage, node_type);
+        Self::store_new_metadata(meta, staging_path).await
+    }
+
+    /// Process and store metadata
+    async fn process_and_store_metadata(
+        mut meta: Self,
+        staging_path: &Path,
+        node_type: NodeType,
+    ) -> anyhow::Result<Self> {
+        Self::update_metadata(&mut meta, &PARSEABLE.options, node_type);
+        meta.put_on_disk(staging_path)
+            .expect("Couldn't write updated metadata to disk");
+
+        let path = meta.file_path();
+        let resource = serde_json::to_vec(&meta)?.into();
+        let store = PARSEABLE.storage.get_object_store();
+        store.put_object(&path, resource).await?;
+
+        Ok(meta)
+    }
+
+    /// Store new metadata
+    async fn store_new_metadata(meta: Self, staging_path: &Path) -> anyhow::Result<Self> {
         meta.put_on_disk(staging_path)
             .expect("Couldn't write new metadata to disk");
-        Arc::new(meta)
+
+        let path = meta.file_path();
+        let resource = serde_json::to_vec(&meta)?.into();
+        let store = PARSEABLE.storage.get_object_store();
+        store.put_object(&path, resource).await?;
+
+        Ok(meta)
+    }
+
+    async fn load_from_storage(node_type: String) -> Vec<NodeMetadata> {
+        let path = RelativePathBuf::from(PARSEABLE_ROOT_DIRECTORY);
+        let glob_storage = PARSEABLE.storage.get_object_store();
+        let obs = glob_storage
+            .get_objects(
+                Some(&path),
+                Box::new({
+                    let node_type = node_type.clone();
+                    move |file_name| file_name.contains(&node_type)
+                }),
+            )
+            .await;
+
+        let mut metadata = vec![];
+        if let Ok(obs) = obs {
+            for object in obs {
+                //convert to NodeMetadata
+                match serde_json::from_slice::<NodeMetadata>(&object) {
+                    Ok(node_metadata) => metadata.push(node_metadata),
+                    Err(e) => error!("Failed to deserialize NodeMetadata: {:?}", e),
+                }
+            }
+        } else {
+            error!("Couldn't read from storage");
+        }
+        // Return the metadata
+        metadata
     }
 
     /// Load metadata from the staging directory
@@ -467,30 +531,6 @@ impl NodeMetadata {
         Ok(metadata)
     }
 
-    pub async fn migrate(&self) -> anyhow::Result<Option<Self>> {
-        let imp = self.file_path();
-        let bytes = match PARSEABLE.storage.get_object_store().get_object(&imp).await {
-            Ok(bytes) => bytes,
-            Err(_) => {
-                return Ok(None);
-            }
-        };
-
-        let mut resource = Self::from_bytes(&bytes, PARSEABLE.options.flight_port)?;
-        resource.node_type.clone_from(&self.node_type);
-        let bytes = Bytes::from(serde_json::to_vec(&resource)?);
-
-        resource.put_on_disk(PARSEABLE.options.staging_dir())?;
-
-        PARSEABLE
-            .storage
-            .get_object_store()
-            .put_object(&imp, bytes)
-            .await?;
-
-        Ok(Some(resource))
-    }
-
     /// Puts the node info into the staging.
     ///
     /// This function takes the node info as a parameter and stores it in staging.

src/handlers/http/modal/query_server.rs

@@ -32,17 +32,17 @@ use actix_web::{web, Scope};
 use actix_web_prometheus::PrometheusMetrics;
 use async_trait::async_trait;
 use bytes::Bytes;
-use tokio::sync::oneshot;
+use tokio::sync::{oneshot, OnceCell};
 use tracing::info;
 
 use crate::parseable::PARSEABLE;
 use crate::Server;
 
 use super::query::{querier_ingest, querier_logstream, querier_rbac, querier_role};
-use super::{load_on_init, OpenIdClient, ParseableServer};
+use super::{load_on_init, NodeType, OpenIdClient, ParseableServer, QuerierMetadata};
 
 pub struct QueryServer;
-
+pub static QUERIER_META: OnceCell<QuerierMetadata> = OnceCell::const_new();
 #[async_trait]
 impl ParseableServer for QueryServer {
     // configure the api routes
@@ -99,7 +99,14 @@ impl ParseableServer for QueryServer {
         shutdown_rx: oneshot::Receiver<()>,
     ) -> anyhow::Result<()> {
         PARSEABLE.storage.register_store_metrics(prometheus);
-
+        // write the ingestor metadata to storage
+        QUERIER_META
+            .get_or_init(|| async {
+                QuerierMetadata::load_node_metadata(NodeType::Querier)
+                    .await
+                    .expect("Querier Metadata should be set in ingestor mode")
+            })
+            .await;
         migration::run_migration(&PARSEABLE).await?;
 
         //create internal stream at server start