impl ingester metadata migration

Author: Eshanatnight

server/src/handlers/http/modal/ingest_server.rs

@@ -24,6 +24,7 @@ use crate::handlers::http::middleware::RouteExt;
 use crate::localcache::LocalCacheManager;
 use crate::metadata;
 use crate::metrics;
+use crate::migration::metadata_migration::migrate_ingester_metadata;
 use crate::rbac;
 use crate::rbac::role::Action;
 use crate::storage;
@@ -39,6 +40,10 @@ use super::IngestorMetadata;
 use super::OpenIdClient;
 use super::ParseableServer;
 
+use crate::{
+    handlers::http::{base_path, cross_origin_config},
+    option::CONFIG,
+};
 use actix_web::body::MessageBody;
 use actix_web::Scope;
 use actix_web::{web, App, HttpServer};
@@ -50,14 +55,9 @@ use itertools::Itertools;
 use once_cell::sync::Lazy;
 use relative_path::RelativePathBuf;
 
-use crate::{
-    handlers::http::{base_path, cross_origin_config},
-    option::CONFIG,
-};
-
 /// ! have to use a guard before using it
 pub static INGESTOR_META: Lazy<IngestorMetadata> =
-    Lazy::new(|| staging::get_ingestor_info().expect("dir is readable and writeable"));
+    Lazy::new(|| staging::get_ingestor_info().expect("Should Be valid Json"));
 
 #[derive(Default)]
 pub struct IngestServer;
@@ -106,6 +106,8 @@ impl ParseableServer for IngestServer {
 
     /// implement the init method will just invoke the initialize method
     async fn init(&self) -> anyhow::Result<()> {
+        migrate_ingester_metadata().await?;
+
         self.validate()?;
 
         // check for querier state. Is it there, or was it there in the past

server/src/migration.rs

@@ -17,7 +17,7 @@
 *
 */
 
-mod metadata_migration;
+pub mod metadata_migration;
 mod schema_migration;
 mod stream_metadata_migration;
 

server/src/migration/metadata_migration.rs

@@ -19,7 +19,12 @@
 use rand::distributions::DistString;
 use serde_json::{Map, Value as JsonValue};
 
-use crate::option::CONFIG;
+use crate::{
+    handlers::http::modal::IngestorMetadata,
+    option::CONFIG,
+    storage::{object_storage::ingestor_metadata_path, staging},
+};
+use actix_web::body::MessageBody;
 
 /*
 v1
@@ -118,3 +123,34 @@ pub fn update_v3(mut storage_metadata: JsonValue) -> JsonValue {
 
     storage_metadata
 }
+
+pub async fn migrate_ingester_metadata() -> anyhow::Result<()> {
+    let imp = ingestor_metadata_path(None);
+    let bytes = CONFIG.storage().get_object_store().get_object(&imp).await?;
+    let mut json = serde_json::from_slice::<JsonValue>(&bytes)?;
+    let meta = json
+        .as_object_mut()
+        .ok_or_else(|| anyhow::anyhow!("Unable to parse Ingester Metadata"))?;
+    let fp = meta.get("flight_port");
+
+    if fp.is_none() {
+        meta.insert(
+            "flight_port".to_owned(),
+            JsonValue::String(CONFIG.parseable.flight_port.to_string()),
+        );
+    }
+    let bytes = serde_json::to_string(&json)?
+        .try_into_bytes()
+        .map_err(|err| anyhow::anyhow!(err))?;
+
+    let resource: IngestorMetadata = serde_json::from_value(json)?;
+    staging::put_ingestor_info(resource.clone())?;
+
+    CONFIG
+        .storage()
+        .get_object_store()
+        .put_object(&imp, bytes)
+        .await?;
+
+    Ok(())
+}

server/src/storage/object_storage.rs

@@ -530,7 +530,7 @@ pub async fn commit_schema_to_storage(
 }
 
 #[inline(always)]
-fn to_bytes(any: &(impl ?Sized + serde::Serialize)) -> Bytes {
+pub fn to_bytes(any: &(impl ?Sized + serde::Serialize)) -> Bytes {
     serde_json::to_vec(any)
         .map(|any| any.into())
         .expect("serialize cannot fail")

server/src/storage/staging.rs

@@ -17,14 +17,6 @@
  *
  */
 
-use std::{
-    collections::HashMap,
-    fs,
-    path::{Path, PathBuf},
-    process,
-    sync::Arc,
-};
-
 use crate::{
     event::DEFAULT_TIMESTAMP_KEY,
     handlers::http::modal::{ingest_server::INGESTOR_META, IngestorMetadata, DEFAULT_VERSION},
@@ -36,6 +28,7 @@ use crate::{
         hostname_unchecked,
     },
 };
+use anyhow::anyhow;
 use arrow_schema::{ArrowError, Schema};
 use base64::Engine;
 use chrono::{NaiveDateTime, Timelike, Utc};
@@ -47,6 +40,14 @@ use parquet::{
     format::SortingColumn,
     schema::types::ColumnPath,
 };
+use serde_json::Value as JsonValue;
+use std::{
+    collections::HashMap,
+    fs,
+    path::{Path, PathBuf},
+    process,
+    sync::Arc,
+};
 
 const ARROW_FILE_EXTENSION: &str = "data.arrows";
 const PARQUET_FILE_EXTENSION: &str = "data.parquet";
@@ -318,7 +319,21 @@ pub fn get_ingestor_info() -> anyhow::Result<IngestorMetadata> {
 
         if flag {
             // get the ingestor metadata from staging
-            let mut meta: IngestorMetadata = serde_json::from_slice(&std::fs::read(path)?)?;
+            let mut meta: JsonValue = serde_json::from_slice(&std::fs::read(path)?)?;
+
+            // migrate the staging meta
+            let obj = meta
+                .as_object_mut()
+                .ok_or_else(|| anyhow!("Could Not parse Ingestor Metadata Json"))?;
+
+            if obj.get("flight_port").is_none() {
+                obj.insert(
+                    "flight_port".to_owned(),
+                    JsonValue::String(CONFIG.parseable.flight_port.to_string()),
+                );
+            }
+
+            let mut meta: IngestorMetadata = serde_json::from_value(meta)?;
 
             // compare url endpoint and port
             if meta.domain_name != url {
@@ -379,7 +394,7 @@ pub fn get_ingestor_info() -> anyhow::Result<IngestorMetadata> {
 /// # Parameters
 ///
 /// * `ingestor_info`: The ingestor info to be stored.
-fn put_ingestor_info(info: IngestorMetadata) -> anyhow::Result<()> {
+pub fn put_ingestor_info(info: IngestorMetadata) -> anyhow::Result<()> {
     let path = PathBuf::from(&CONFIG.parseable.local_staging_path);
     let file_name = format!("ingestor.{}.json", info.ingestor_id);
     let file_path = path.join(file_name);