test: s3 manifest (#1755)

* feat: change default manifest options

* test: s3 manifest

* feat: revert checkpoint_margin to 10

* Update src/object-store/src/test_util.rs

Co-authored-by: Lei, HUANG <6406592+v0y4g3r@users.noreply.github.com>

---------

Co-authored-by: Lei, HUANG <6406592+v0y4g3r@users.noreply.github.com>

src/storage/src/manifest/impl_.rs

@@ -36,7 +36,7 @@ use crate::manifest::checkpoint::Checkpointer;
 use crate::manifest::storage::{ManifestObjectStore, ObjectStoreLogIterator};
 
 const CHECKPOINT_ACTIONS_MARGIN: u16 = 10;
-const GC_DURATION_SECS: u64 = 30;
+const GC_DURATION_SECS: u64 = 600;
 
 #[derive(Clone, Debug)]
 pub struct ManifestImpl<S: Checkpoint<Error = Error>, M: MetaAction<Error = Error>> {

src/storage/src/manifest/region.rs

@@ -183,7 +183,8 @@ mod tests {
     use std::sync::Arc;
 
     use common_test_util::temp_dir::create_temp_dir;
-    use object_store::services::Fs;
+    use object_store::services::{Fs, S3};
+    use object_store::test_util::{s3_test_config, TempFolder};
     use object_store::ObjectStore;
     use store_api::manifest::action::ProtocolAction;
     use store_api::manifest::{Manifest, MetaActionIterator, MAX_VERSION};
@@ -195,17 +196,36 @@ mod tests {
     use crate::sst::FileId;
 
     #[tokio::test]
-    async fn test_region_manifest_compress() {
-        test_region_manifest(true).await
+    async fn test_fs_region_manifest_compress() {
+        let manifest = new_fs_manifest(true, None).await;
+        test_region_manifest(&manifest).await
     }
 
     #[tokio::test]
-    async fn test_region_manifest_uncompress() {
-        test_region_manifest(false).await
+    async fn test_fs_region_manifest_uncompress() {
+        let manifest = new_fs_manifest(false, None).await;
+        test_region_manifest(&manifest).await
     }
 
-    async fn test_region_manifest(compress: bool) {
-        common_telemetry::init_default_ut_logging();
+    #[tokio::test]
+    async fn test_s3_region_manifest_compress() {
+        if s3_test_config().is_some() {
+            let (manifest, temp_dir) = new_s3_manifest(true, None).await;
+            test_region_manifest(&manifest).await;
+            temp_dir.remove_all().await.unwrap();
+        }
+    }
+
+    #[tokio::test]
+    async fn test_s3_region_manifest_uncompress() {
+        if s3_test_config().is_some() {
+            let (manifest, temp_dir) = new_s3_manifest(false, None).await;
+            test_region_manifest(&manifest).await;
+            temp_dir.remove_all().await.unwrap();
+        }
+    }
+
+    async fn new_fs_manifest(compress: bool, gc_duration: Option<Duration>) -> RegionManifest {
         let tmp_dir = create_temp_dir("test_region_manifest");
         let mut builder = Fs::default();
         builder.root(&tmp_dir.path().to_string_lossy());
@@ -216,9 +236,43 @@ mod tests {
             object_store,
             manifest_compress_type(compress),
             None,
-            None,
+            gc_duration,
         );
         manifest.start().await.unwrap();
+        manifest
+    }
+
+    async fn new_s3_manifest(
+        compress: bool,
+        gc_duration: Option<Duration>,
+    ) -> (RegionManifest, TempFolder) {
+        let s3_config = s3_test_config().unwrap();
+        let mut builder = S3::default();
+        builder
+            .root(&s3_config.root)
+            .access_key_id(&s3_config.access_key_id)
+            .secret_access_key(&s3_config.secret_access_key)
+            .bucket(&s3_config.bucket);
+
+        if s3_config.region.is_some() {
+            builder.region(s3_config.region.as_ref().unwrap());
+        }
+        let store = ObjectStore::new(builder).unwrap().finish();
+        let temp_folder = TempFolder::new(&store, "/");
+        let manifest = RegionManifest::with_checkpointer(
+            "/manifest/",
+            store,
+            manifest_compress_type(compress),
+            None,
+            gc_duration,
+        );
+        manifest.start().await.unwrap();
+
+        (manifest, temp_folder)
+    }
+
+    async fn test_region_manifest(manifest: &RegionManifest) {
+        common_telemetry::init_default_ut_logging();
 
         let region_meta = Arc::new(build_region_meta());
 
@@ -325,31 +379,48 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn test_region_manifest_checkpoint_compress() {
-        test_region_manifest_checkpoint(true).await
+    async fn test_fs_region_manifest_checkpoint_compress() {
+        let duration = Duration::from_millis(50);
+        let manifest = new_fs_manifest(true, Some(duration)).await;
+
+        test_region_manifest_checkpoint(&manifest, duration).await
     }
 
     #[tokio::test]
-    async fn test_region_manifest_checkpoint_uncompress() {
-        test_region_manifest_checkpoint(false).await
+    async fn test_fs_region_manifest_checkpoint_uncompress() {
+        let duration = Duration::from_millis(50);
+        let manifest = new_fs_manifest(false, Some(duration)).await;
+
+        test_region_manifest_checkpoint(&manifest, duration).await
     }
 
-    async fn test_region_manifest_checkpoint(compress: bool) {
-        common_telemetry::init_default_ut_logging();
-        let tmp_dir = create_temp_dir("test_region_manifest_checkpoint");
-        let mut builder = Fs::default();
-        builder.root(&tmp_dir.path().to_string_lossy());
-        let object_store = ObjectStore::new(builder).unwrap().finish();
+    #[tokio::test]
+    async fn test_s3_region_manifest_checkpoint_compress() {
+        if s3_test_config().is_some() {
+            let duration = Duration::from_millis(50);
+            let (manifest, temp_dir) = new_s3_manifest(true, Some(duration)).await;
 
-        let test_gc_duration = Duration::from_millis(50);
-        let manifest = RegionManifest::with_checkpointer(
-            "/manifest/",
-            object_store,
-            manifest_compress_type(compress),
-            None,
-            Some(test_gc_duration),
-        );
-        manifest.start().await.unwrap();
+            test_region_manifest_checkpoint(&manifest, duration).await;
+            temp_dir.remove_all().await.unwrap();
+        }
+    }
+
+    #[tokio::test]
+    async fn test_s3_region_manifest_checkpoint_uncompress() {
+        if s3_test_config().is_some() {
+            let duration = Duration::from_millis(50);
+            let (manifest, temp_dir) = new_s3_manifest(false, Some(duration)).await;
+
+            test_region_manifest_checkpoint(&manifest, duration).await;
+            temp_dir.remove_all().await.unwrap();
+        }
+    }
+
+    async fn test_region_manifest_checkpoint(
+        manifest: &RegionManifest,
+        test_gc_duration: Duration,
+    ) {
+        common_telemetry::init_default_ut_logging();
 
         let region_meta = Arc::new(build_region_meta());
         let new_region_meta = Arc::new(build_altered_region_meta());
@@ -376,7 +447,7 @@ mod tests {
             manifest.update(action).await.unwrap();
         }
         assert!(manifest.last_checkpoint().await.unwrap().is_none());
-        assert_scan(&manifest, 0, 3).await;
+        assert_scan(manifest, 0, 3).await;
         // update flushed manifest version for doing checkpoint
         manifest.set_flushed_manifest_version(2);
 
@@ -435,7 +506,7 @@ mod tests {
             manifest.update(action).await.unwrap();
         }
 
-        assert_scan(&manifest, 3, 2).await;
+        assert_scan(manifest, 3, 2).await;
 
         // do another checkpoints
         // compacted RegionChange
@@ -459,7 +530,7 @@ mod tests {
                          files.contains_key(&file_ids[1]) &&
                          *metadata == RawRegionMetadata::from(region_meta.as_ref())));
 
-        assert_scan(&manifest, 4, 1).await;
+        assert_scan(manifest, 4, 1).await;
         // compacted RegionEdit
         manifest.set_flushed_manifest_version(4);
         let checkpoint = manifest.do_checkpoint().await.unwrap().unwrap();