Command to retrieve object

Author: iwanbk

README.md

@@ -18,6 +18,12 @@ cd s3-cas
 cargo build --release --features binary
 ```
 
+## Running
+
+```console
+s3-cas server --access-key=MY_KEY --secret-key=MY_SECRET --fs-root=/tmp/s3/fs --meta-root=/tmp/s3/meta
+```
+
 ## Inline metadata
 
 Objects smaller than or equal to a configurable threshold can be stored directly in their metadata records,

src/cas/fs.rs

@@ -155,6 +155,10 @@ impl CasFS {
         self.meta_store.get_path_tree()
     }
 
+    pub fn fs_root(&self) -> &PathBuf {
+        &self.root
+    }
+
     pub fn max_inlined_data_length(&self) -> usize {
         self.meta_store.max_inlined_data_length()
     }

src/lib.rs

@@ -5,4 +5,5 @@ pub mod cas;
 pub mod inspect;
 pub mod metastore;
 pub mod metrics;
+pub mod retrieve;
 pub mod s3fs;

src/main.rs

@@ -11,6 +11,7 @@ use tracing_subscriber::FmtSubscriber;
 use s3_cas::cas::{CasFS, StorageEngine};
 use s3_cas::inspect::{disk_space, num_keys};
 use s3_cas::metastore::Durability;
+use s3_cas::retrieve::{retrieve, RetrieveConfig};
 
 #[derive(Parser)]
 #[command(version)]
@@ -81,6 +82,8 @@ pub enum Command {
         command: InspectCommand,
     },
 
+    Retrieve(RetrieveConfig),
+
     /// Start S3-cas server
     Server(ServerConfig),
 }
@@ -120,6 +123,7 @@ fn main() -> Result<()> {
                 println!("Disk space: {}", disk_space);
             }
         },
+        Command::Retrieve(config) => retrieve(config)?,
         Command::Server(config) => {
             run(config)?;
         }

src/retrieve.rs

@@ -0,0 +1,101 @@
+use std::path::PathBuf;
+
+use anyhow::Result;
+use bytes::Bytes;
+use clap::Parser;
+use futures::StreamExt;
+use tokio::io::AsyncWriteExt;
+
+use crate::cas::block_stream::BlockStream;
+use crate::cas::range_request::RangeRequest;
+use crate::cas::CasFS;
+use crate::cas::StorageEngine;
+use crate::metrics::SharedMetrics;
+
+#[derive(Parser, Debug)]
+pub struct RetrieveConfig {
+    #[arg(long, default_value = ".")]
+    pub meta_root: PathBuf,
+
+    #[arg(long, default_value = ".")]
+    pub fs_root: PathBuf,
+
+    #[arg(
+        long,
+        default_value = "fjall",
+        help = "Metadata DB  (fjall, fjall_notx)"
+    )]
+    pub metadata_db: StorageEngine,
+
+    #[arg(required = true, help = "Bucket name")]
+    pub bucket: String,
+
+    #[arg(required = true, help = "Object key")]
+    pub key: String,
+
+    #[arg(required = true, help = "Destination file path")]
+    pub dest: String,
+}
+
+#[tokio::main]
+pub async fn retrieve(args: RetrieveConfig) -> Result<()> {
+    tracing::info!(
+        "Retrieving object from bucket: {}, key: {}",
+        args.bucket,
+        args.key
+    );
+    let storage_engine = args.metadata_db;
+    let metrics = SharedMetrics::new();
+    let casfs = CasFS::new(
+        args.fs_root.clone(),
+        args.meta_root.clone(),
+        metrics.clone(),
+        storage_engine,
+        None,
+        None,
+    );
+
+    tracing::info!("get_object_blocks");
+    let (obj_meta, blocks) = match casfs.get_object_blocks(&args.bucket, &args.key)? {
+        Some((obj, blocks)) => (obj, blocks),
+        None => {
+            println!("Object not found");
+            return Ok(());
+        }
+    };
+
+    tracing::info!("Object found, size: {}", obj_meta.size());
+    if let Some(data) = obj_meta.inlined() {
+        let mut file = tokio::fs::File::create(&args.dest).await?;
+        file.write_all(data).await?;
+        return Ok(());
+    }
+
+    let mut paths = Vec::with_capacity(blocks.len());
+    let mut block_size = 0;
+    for block in blocks {
+        block_size += block.size();
+        paths.push((block.disk_path(casfs.fs_root().clone()), block.size()));
+        tracing::info!("block path: {:?}", block.disk_path(casfs.fs_root().clone()));
+    }
+
+    debug_assert!(obj_meta.size() as usize == block_size);
+    tracing::info!("creating block stream");
+    let mut block_stream = BlockStream::new(paths, block_size, RangeRequest::All, metrics);
+
+    // Create the destination file
+    tracing::info!("creating destination file: {}", args.dest);
+    let mut file = tokio::fs::File::create(&args.dest).await?;
+
+    // Read from block stream and write to file
+    while let Some(chunk_result) = block_stream.next().await {
+        let chunk: Bytes = chunk_result?;
+        tracing::info!("writing chunk of size: {}", chunk.len());
+        file.write_all(&chunk).await?;
+    }
+
+    // Ensure all data is written to disk
+    file.flush().await?;
+
+    Ok(())
+}