Impl migration of server metadata files on Object Store

Author: Eshanatnight

server/src/handlers/http/cluster/mod.rs

@@ -23,6 +23,7 @@ use crate::handlers::http::logstream::error::StreamError;
 use crate::option::CONFIG;
 
 use crate::metrics::prom_utils::Metrics;
+use crate::storage::PARSEABLE_ROOT_DIRECTORY;
 use actix_web::http::header;
 use actix_web::Responder;
 use http::StatusCode;
@@ -334,7 +335,7 @@ pub async fn get_cluster_metrics() -> Result<impl Responder, PostError> {
 pub async fn get_ingester_info() -> anyhow::Result<IngesterMetadataArr> {
     let store = CONFIG.storage().get_object_store();
 
-    let root_path = RelativePathBuf::from("");
+    let root_path = RelativePathBuf::from(PARSEABLE_ROOT_DIRECTORY);
     let arr = store
         .get_objects(Some(&root_path))
         .await?

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

@@ -102,6 +102,17 @@ impl ParseableServer for IngestServer {
     /// implement the init method will just invoke the initialize method
     async fn init(&self) -> anyhow::Result<()> {
         self.validate()?;
+        // check for querier state. Is it there, or was it there in the past
+        self.check_querier_state().await?;
+        // to get the .parseable.json file in staging
+        self.validate_credentials().await?;
+
+        let metadata = storage::resolve_parseable_metadata().await?;
+        banner::print(&CONFIG, &metadata).await;
+        rbac::map::init(&metadata);
+        // set the info in the global metadata
+        metadata.set_global();
+
         self.initialize().await
     }
 
@@ -267,19 +278,6 @@ impl IngestServer {
     }
 
     async fn initialize(&self) -> anyhow::Result<()> {
-        // check for querier state. Is it there, or was it there in the past
-        self.check_querier_state().await?;
-        // to get the .parseable.json file in staging
-        self.validate_credentials().await?;
-
-        let metadata = storage::resolve_parseable_metadata().await?;
-        banner::print(&CONFIG, &metadata).await;
-
-        rbac::map::init(&metadata);
-
-        // set the info in the global metadata
-        metadata.set_global();
-
         if let Some(cache_manager) = LocalCacheManager::global() {
             cache_manager
                 .validate(CONFIG.parseable.local_cache_size)

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

@@ -98,6 +98,14 @@ impl ParseableServer for QueryServer {
     /// implementation of init should just invoke a call to initialize
     async fn init(&self) -> anyhow::Result<()> {
         self.validate()?;
+        migration::meta_file_migration(&CONFIG).await?;
+        migration::run_metadata_migration(&CONFIG).await?;
+        let metadata = storage::resolve_parseable_metadata().await?;
+        banner::print(&CONFIG, &metadata).await;
+        // initialize the rbac map
+        rbac::map::init(&metadata);
+        // keep metadata info in mem
+        metadata.set_global();
         self.initialize().await
     }
 
@@ -153,18 +161,6 @@ impl QueryServer {
 
     /// initialize the server, run migrations as needed and start the server
     async fn initialize(&self) -> anyhow::Result<()> {
-        migration::run_metadata_migration(&CONFIG).await?;
-
-        let metadata = storage::resolve_parseable_metadata().await?;
-
-        banner::print(&CONFIG, &metadata).await;
-
-        // initialize the rbac map
-        rbac::map::init(&metadata);
-
-        // keep metadata info in mem
-        metadata.set_global();
-
         let prometheus = metrics::build_metrics_handler();
         CONFIG.storage().register_store_metrics(&prometheus);
 
@@ -177,7 +173,6 @@ impl QueryServer {
 
         // track all parquet files already in the data directory
         storage::retention::load_retention_from_global();
-
         // load data from stats back to prometheus metrics
         metrics::fetch_stats_from_storage().await;
 

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

@@ -405,6 +405,7 @@ impl Server {
     }
 
     async fn initialize(&self) -> anyhow::Result<()> {
+        migration::meta_file_migration(&CONFIG).await?;
         migration::run_metadata_migration(&CONFIG).await?;
         let metadata = storage::resolve_parseable_metadata().await?;
         banner::print(&CONFIG, &metadata).await;

server/src/migration.rs

@@ -30,7 +30,7 @@ use serde::Serialize;
 use crate::{
     option::Config,
     storage::{
-        object_storage::{parseable_json_path, stream_json_path}, ObjectStorage, ObjectStorageError,SCHEMA_FILE_NAME,
+        object_storage::{parseable_json_path, stream_json_path}, ObjectStorage, ObjectStorageError, PARSEABLE_METADATA_FILE_NAME, PARSEABLE_ROOT_DIRECTORY, SCHEMA_FILE_NAME
     },
 };
 
@@ -195,7 +195,6 @@ pub async fn put_remote_metadata(
 
 pub fn put_staging_metadata(config: &Config, metadata: &serde_json::Value) -> anyhow::Result<()> {
     let path = parseable_json_path().to_path(config.staging_dir());
-    //config.staging_dir().join(PARSEABLE_METADATA_FILE_NAME);
     let mut file = OpenOptions::new()
         .create(true)
         .truncate(true)
@@ -204,3 +203,43 @@ pub fn put_staging_metadata(config: &Config, metadata: &serde_json::Value) -> an
     serde_json::to_writer(&mut file, metadata)?;
     Ok(())
 }
+
+pub async fn meta_file_migration(config: &Config) -> anyhow::Result<()> {
+    let object_store = config.storage().get_object_store();
+
+    let old_meta_file_path = RelativePathBuf::from(PARSEABLE_METADATA_FILE_NAME);
+
+    // if this errors that means migrations is already done
+    if let Err(err) = object_store.get_object(&old_meta_file_path).await {
+        if matches!(err, ObjectStorageError::NoSuchKey(_)) {
+            return Ok(());
+        }
+        return Err(err.into());
+    }
+
+    log::info!("Migrating metadata files to new location");
+
+    // get the list of all meta files
+    let mut meta_files = object_store.get_ingester_meta_file_paths().await?;
+    meta_files.push(old_meta_file_path);
+
+    for file in meta_files {
+        match object_store.get_object(&file).await {
+            Ok(bytes) => {
+                // we can unwrap here because we know the file exists
+                let new_path = RelativePathBuf::from_iter([PARSEABLE_ROOT_DIRECTORY, file.file_name().unwrap()]);
+                object_store.put_object(&new_path, bytes).await?;
+                object_store.delete_object(&file).await?;
+            },
+            Err(err) => {
+                // if error is not a no such key error, something weird happened
+                // so return the error
+                if !matches!(err, ObjectStorageError::NoSuchKey(_)) {
+                    return Err(err.into());
+                }
+            }
+        }
+    }
+
+    Ok(())
+}

server/src/storage/localfs.rs

@@ -27,7 +27,7 @@ use bytes::Bytes;
 use datafusion::{datasource::listing::ListingTableUrl, execution::runtime_env::RuntimeConfig};
 use fs_extra::file::CopyOptions;
 use futures::{stream::FuturesUnordered, TryStreamExt};
-use relative_path::RelativePath;
+use relative_path::{RelativePath, RelativePathBuf};
 use tokio::fs::{self, DirEntry};
 use tokio_stream::wrappers::ReadDirStream;
 
@@ -114,6 +114,39 @@ impl ObjectStorage for LocalFS {
         res
     }
 
+    async fn get_ingester_meta_file_paths(
+        &self,
+    ) -> Result<Vec<RelativePathBuf>, ObjectStorageError> {
+        let time = Instant::now();
+
+        let mut path_arr = vec![];
+        let mut entries = fs::read_dir(&self.root).await?;
+
+        while let Some(entry) = entries.next_entry().await? {
+            let flag = entry
+                .path()
+                .file_name()
+                .unwrap_or_default()
+                .to_str()
+                .unwrap_or_default()
+                .contains("ingester");
+
+            if flag {
+                path_arr.push(
+                    RelativePathBuf::from_path(entry.path().file_name().unwrap())
+                        .map_err(|err| ObjectStorageError::Custom(err.to_string()))?,
+                );
+            }
+        }
+
+        let time = time.elapsed().as_secs_f64();
+        REQUEST_RESPONSE_TIME
+            .with_label_values(&["GET", "200"]) // this might not be the right status code
+            .observe(time);
+
+        Ok(path_arr)
+    }
+
     async fn get_objects(
         &self,
         base_path: Option<&RelativePath>,
@@ -183,6 +216,12 @@ impl ObjectStorage for LocalFS {
         Ok(())
     }
 
+    async fn delete_object(&self, path: &RelativePath) -> Result<(), ObjectStorageError> {
+        let path = self.path_in_root(path);
+        tokio::fs::remove_file(path).await?;
+        Ok(())
+    }
+
     async fn check(&self) -> Result<(), ObjectStorageError> {
         fs::create_dir_all(&self.root)
             .await

server/src/storage/object_storage.rs

@@ -82,7 +82,8 @@ pub trait ObjectStorage: Sync + 'static {
     async fn list_dirs(&self) -> Result<Vec<String>, ObjectStorageError>;
     async fn list_dates(&self, stream_name: &str) -> Result<Vec<String>, ObjectStorageError>;
     async fn upload_file(&self, key: &str, path: &Path) -> Result<(), ObjectStorageError>;
-
+    async fn delete_object(&self, path: &RelativePath) -> Result<(), ObjectStorageError>;
+    async fn get_ingester_meta_file_paths(&self) -> Result<Vec<RelativePathBuf>, ObjectStorageError>;
     /// Returns the amount of time taken by the `ObjectStore` to perform a get
     /// call.
     async fn get_latency(&self) -> Duration {

server/src/storage/s3.rs

@@ -29,7 +29,7 @@ use object_store::aws::{AmazonS3, AmazonS3Builder, AmazonS3ConfigKey, Checksum};
 use object_store::limit::LimitStore;
 use object_store::path::Path as StorePath;
 use object_store::{ClientOptions, ObjectStore};
-use relative_path::RelativePath;
+use relative_path::{RelativePath, RelativePathBuf};
 use tokio::fs::OpenOptions;
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 
@@ -452,6 +452,29 @@ impl ObjectStorage for S3 {
         Ok(res)
     }
 
+    async fn get_ingester_meta_file_paths(
+        &self,
+    ) -> Result<Vec<RelativePathBuf>, ObjectStorageError> {
+        let time = Instant::now();
+        let mut path_arr = vec![];
+        let mut object_stream = self.client.list(Some(&self.root)).await?;
+
+        while let Some(meta) = object_stream.next().await.transpose()? {
+            let flag = meta.location.filename().unwrap().starts_with("ingester");
+
+            if flag {
+                path_arr.push(RelativePathBuf::from(meta.location.as_ref()));
+            }
+        }
+
+        let time = time.elapsed().as_secs_f64();
+        REQUEST_RESPONSE_TIME
+            .with_label_values(&["GET", "200"])
+            .observe(time);
+
+        Ok(path_arr)
+    }
+
     async fn put_object(
         &self,
         path: &RelativePath,
@@ -470,6 +493,10 @@ impl ObjectStorage for S3 {
         Ok(())
     }
 
+    async fn delete_object(&self, path: &RelativePath) -> Result<(), ObjectStorageError> {
+        Ok(self.client.delete(&to_object_store_path(path)).await?)
+    }
+
     async fn check(&self) -> Result<(), ObjectStorageError> {
         Ok(self
             .client

server/src/storage/store_metadata.rs

@@ -126,33 +126,39 @@ pub async fn resolve_parseable_metadata() -> Result<StorageMetadata, ObjectStora
             Err("Could not start the server because staging directory indicates stale data from previous deployment, please choose an empty staging directory and restart the server")
         }
         EnvChange::NewStaging(mut metadata) => {
-            create_dir_all(CONFIG.staging_dir())?;
-            metadata.staging = CONFIG.staging_dir().canonicalize()?;
-            // this flag is set to true so that metadata is copied to staging
-            overwrite_staging = true;
-            // overwrite remote in all and query mode
-            // because staging dir has changed.
-            match CONFIG.parseable.mode {
-                Mode::All => {
-                    standalone_when_distributed(Mode::from_string(&metadata.server_mode).expect("mode should be valid at here"))
-                        .map_err(|err| {
-                            ObjectStorageError::Custom(err.to_string())
-                        })?;
+            // if server is started in ingest mode,we need to make sure that query mode has been started
+            // i.e the metadata is updated to reflect the server mode = Query
+            if Mode::from_string(&metadata.server_mode).unwrap() == Mode::All || CONFIG.parseable.mode == Mode::Ingest {
+                Err("Starting Ingest Mode is not allowed, Since Query Server has not been started yet")
+            } else {
+                create_dir_all(CONFIG.staging_dir())?;
+                metadata.staging = CONFIG.staging_dir().canonicalize()?;
+                // this flag is set to true so that metadata is copied to staging
+                overwrite_staging = true;
+                // overwrite remote in all and query mode
+                // because staging dir has changed.
+                match CONFIG.parseable.mode {
+                    Mode::All => {
+                        standalone_when_distributed(Mode::from_string(&metadata.server_mode).expect("mode should be valid at here"))
+                            .map_err(|err| {
+                                ObjectStorageError::Custom(err.to_string())
+                            })?;
+                            overwrite_remote = true;
+                    },
+                    Mode::Query => {
                         overwrite_remote = true;
-                },
-                Mode::Query => {
-                    overwrite_remote = true;
-                    metadata.server_mode = CONFIG.parseable.mode.to_string();
-                    metadata.staging = CONFIG.staging_dir().to_path_buf();
-                },
-                Mode::Ingest => {
-                    // if ingest server is started fetch the metadata from remote
-                    // update the server mode for local metadata
-                    metadata.server_mode = CONFIG.parseable.mode.to_string();
-                    metadata.staging = CONFIG.staging_dir().to_path_buf();
-                },
+                        metadata.server_mode = CONFIG.parseable.mode.to_string();
+                        metadata.staging = CONFIG.staging_dir().to_path_buf();
+                    },
+                    Mode::Ingest => {
+                        // if ingest server is started fetch the metadata from remote
+                        // update the server mode for local metadata
+                        metadata.server_mode = CONFIG.parseable.mode.to_string();
+                        metadata.staging = CONFIG.staging_dir().to_path_buf();
+                    },
+                }
+                Ok(metadata)
             }
-            Ok(metadata)
         }
         EnvChange::CreateBoth => {
             create_dir_all(CONFIG.staging_dir())?;