Create snapshots path of collection upon trigger of snapshot api (#5574)

* Create snapshots path of collection upon trigger of snapshot api

Sync changes within toc

* [Test fix] Create snapshot on all peer nodes

---------

Co-authored-by: Gulshan Kumar <kumargu@amazon.com>

lib/storage/src/content_manager/toc/create_collection.rs

@@ -70,7 +70,9 @@ impl TableOfContent {
         }
 
         let collection_path = self.create_collection_path(collection_name).await?;
-        let snapshots_path = self.create_snapshots_path(collection_name).await?;
+        // derive the snapshots path for the collection to be used across collection operation, the directories for the snapshot
+        // is created only when a create snapshot api is invoked.
+        let snapshots_path = self.snapshots_path_for_collection(collection_name);
 
         let collection_defaults_config = self.storage_config.collection.as_ref();
 

lib/storage/src/content_manager/toc/mod.rs

@@ -97,8 +97,6 @@ impl TableOfContent {
         this_peer_id: PeerId,
         consensus_proposal_sender: Option<OperationSender>,
     ) -> Self {
-        let snapshots_path = Path::new(&storage_config.snapshots_path.clone()).to_owned();
-        create_dir_all(&snapshots_path).expect("Can't create Snapshots directory");
         let collections_path = Path::new(&storage_config.storage_path).join(COLLECTIONS_DIR);
         create_dir_all(&collections_path).expect("Can't create Collections directory");
         if let Some(path) = storage_config.temp_path.as_deref() {
@@ -128,11 +126,11 @@ impl TableOfContent {
                 .to_str()
                 .expect("A filename of one of the collection files is not a valid UTF-8")
                 .to_string();
+
+            let snapshots_path = Path::new(&storage_config.snapshots_path.clone()).to_owned();
             let collection_snapshots_path =
                 Self::collection_snapshots_path(&snapshots_path, &collection_name);
-            create_dir_all(&collection_snapshots_path).unwrap_or_else(|e| {
-                panic!("Can't create a directory for snapshot of {collection_name}: {e}")
-            });
+
             log::info!("Loading collection: {collection_name}");
             let collection = general_runtime.block_on(Collection::load(
                 collection_name.clone(),

lib/storage/src/content_manager/toc/snapshots.rs

@@ -54,9 +54,13 @@ impl TableOfContent {
 
     pub async fn create_snapshot(
         &self,
-        collection: &CollectionPass<'_>,
+        collection_pass: &CollectionPass<'_>,
     ) -> Result<SnapshotDescription, StorageError> {
-        let collection = self.get_collection(collection).await?;
+        // create all the directories of the derived collection snapshot path of
+        // the collection.
+        self.create_snapshots_path(collection_pass.name()).await?;
+
+        let collection = self.get_collection(collection_pass).await?;
         // We want to use temp dir inside the temp_path (storage if not specified), because it is possible, that
         // snapshot directory is mounted as network share and multiple writes to it could be slow
         let temp_dir = self.optional_temp_or_storage_temp_path()?;

tests/consensus_tests/test_snapshot_recovery.py

@@ -7,7 +7,6 @@ N_PEERS = 3
 N_SHARDS = 3
 COLLECTION_NAME = "test_collection"
 
-
 def create_snapshot(peer_api_uri):
     r = requests.post(f"{peer_api_uri}/collections/{COLLECTION_NAME}/snapshots")
     assert_http_ok(r)
@@ -87,9 +86,14 @@ def recover_from_snapshot(tmp_path: pathlib.Path, n_replicas):
     snapshot_name = create_snapshot(peer_api_uris[-1])
     assert snapshot_name is not None
 
-    # move file
+    # move snapshots across peers. 
+    #
     snapshot_path = Path(peer_dirs[-1]) / "snapshots" / COLLECTION_NAME / snapshot_name
     assert snapshot_path.exists()
+    # we setup the snapshots base paths on the peer where snapshot from an another peer will 
+    # be moved.
+    create_snapshot(peer_api_uris[0]) 
+    # move
     snapshot_path.rename(Path(peer_dirs[0]) / "snapshots" / COLLECTION_NAME / snapshot_name)
 
     process_peer_id = get_peer_id(peer_api_uris[-1])