DB inspect:

- add num_keys and disk_space to MetaStore trait
- refactor get_object to be reusable
- use clap to support sub command

Author: iwanbk

Cargo.lock

@@ -23,7 +23,6 @@ async-fs = "2.1"
 faster-hex = "0.10.0"
 uuid = { version = "1.12", features = ["v4"] }
 chrono = "0.4"
-structopt = { version = "0.3.26" }
 anyhow = { version = "1.0.95" }
 dotenv = { version = "0.15" }
 openssl = { version = "0.10.68", features = ["vendored"], optional = true }
@@ -42,6 +41,7 @@ hyper-util = { version = "0.1.9", features = [
 rusoto_core = "0.48.0"
 hyper = { version = "1.6.0" }
 http-body-util = "0.1.2"
+clap = { version = "4.5.32", features = ["derive"] }
 
 
 [profile.release]

Cargo.toml

@@ -200,6 +200,32 @@ impl CasFS {
         self.meta_store.get_meta(bucket_name, key)
     }
 
+    pub fn get_object_blocks(
+        &self,
+        bucket_name: &str,
+        key: &str,
+    ) -> Result<Option<(Object, Vec<crate::metastore::Block>)>, MetaError> {
+        let obj_meta = self.get_object_meta(bucket_name, key)?;
+        let Some(obj_meta) = obj_meta else {
+            return Ok(None);
+        };
+
+        if obj_meta.is_inlined() {
+            Ok(Some((obj_meta, vec![])))
+        } else {
+            let blocks = obj_meta.blocks();
+            let block_map = self.block_tree()?;
+            let mut block_vec = Vec::with_capacity(blocks.len());
+            for block in blocks {
+                let block_meta = block_map
+                    .get_block(block)?
+                    .ok_or(MetaError::BlockNotFound)?;
+                block_vec.push(block_meta);
+            }
+            Ok(Some((obj_meta, block_vec)))
+        }
+    }
+
     // create and insert a new  bucket
     pub fn create_bucket(&self, bucket_name: &str) -> Result<(), MetaError> {
         let bm = BucketMeta::new(bucket_name.to_string());

src/cas/fs.rs

@@ -2,57 +2,67 @@ use std::path::PathBuf;
 
 use anyhow::Result;
 use bytes::Bytes;
+use clap::{Parser, Subcommand};
 use http_body_util::Full;
 use prometheus::Encoder;
-use structopt::StructOpt;
 use tracing::{info, Level};
 use tracing_subscriber::FmtSubscriber;
 
 use s3_cas::cas::{CasFS, StorageEngine};
 use s3_cas::metastore::Durability;
 
-#[derive(StructOpt)]
-struct Args {
-    #[structopt(long, default_value = ".")]
+#[derive(Parser)]
+#[command(version)]
+struct Cli {
+    #[arg(long, default_value = ".")]
     fs_root: PathBuf,
 
-    #[structopt(long, default_value = ".")]
+    #[arg(long, default_value = ".")]
     meta_root: PathBuf,
 
-    #[structopt(long, default_value = "localhost")]
+    #[arg(long, default_value = "localhost")]
     host: String,
 
-    #[structopt(long, default_value = "8014")]
+    #[arg(long, default_value = "8014")]
     port: u16,
 
-    #[structopt(long, default_value = "localhost")]
+    #[arg(long, default_value = "localhost")]
     metric_host: String,
 
-    #[structopt(long, default_value = "9100")]
+    #[arg(long, default_value = "9100")]
     metric_port: u16,
 
-    #[structopt(long, help = "leave empty to disable it")]
+    #[arg(long, help = "leave empty to disable it")]
     inline_metadata_size: Option<usize>,
 
-    #[structopt(long, requires("secret-key"), display_order = 1000)]
+    #[arg(long, required = true, display_order = 1000)]
     access_key: Option<String>,
 
-    #[structopt(long, requires("access-key"), display_order = 1000)]
+    #[arg(long, required = true, display_order = 1000)]
     secret_key: Option<String>,
 
-    #[structopt(
+    #[arg(
         long,
         default_value = "fjall",
-        help = "Metadata DB (fjall, fjall_notx)"
+        help = "Metadata DB  (fjall, fjall_notx)"
     )]
     metadata_db: StorageEngine,
 
-    #[structopt(
+    #[arg(
         long,
         default_value = "fdatasync",
         help = "Durability level (buffer, fsync, fdatasync)"
     )]
     durability: Durability,
+
+    #[command(subcommand)]
+    command: Option<Command>,
+}
+
+#[derive(Debug, Subcommand)]
+pub enum Command {
+    // TODO: inspect DB
+    Inspect,
 }
 
 fn setup_tracing() {
@@ -67,18 +77,24 @@ fn main() -> Result<()> {
     dotenv::dotenv().ok();
 
     setup_tracing();
-
-    let args: Args = Args::from_args();
-
-    run(args)
+    let cli = Cli::parse();
+    match cli.command {
+        Some(Command::Inspect) => {
+            println!("Inspecting");
+        }
+        None => {
+            run(cli)?;
+        }
+    }
+    Ok(())
 }
 
 use hyper_util::rt::{TokioExecutor, TokioIo};
 use hyper_util::server::conn::auto::Builder as ConnBuilder;
 use s3s::service::S3ServiceBuilder;
 
 #[tokio::main]
-async fn run(args: Args) -> anyhow::Result<()> {
+async fn run(args: Cli) -> anyhow::Result<()> {
     let storage_engine = args.metadata_db;
 
     // provider

src/main.rs

@@ -40,6 +40,7 @@ pub enum MetaError {
     NotMetaTree(String),
     TransactionError(String),
     PersistError(String),
+    BlockNotFound,
     OtherDBError(String),
 }
 
@@ -59,6 +60,7 @@ impl fmt::Display for MetaError {
             MetaError::NotMetaTree(ref s) => write!(f, "Not a meta tree: {}", s),
             MetaError::TransactionError(ref s) => write!(f, "Transaction error: {}", s),
             MetaError::PersistError(ref s) => write!(f, "Persist error: {}", s),
+            MetaError::BlockNotFound => write!(f, "Block not found"),
             MetaError::OtherDBError(ref s) => write!(f, "Other DB error: {}", s),
         }
     }

src/metastore/errors.rs

@@ -143,6 +143,9 @@ impl Object {
         }
     }
 
+    pub fn is_inlined(&self) -> bool {
+        matches!(&self.data, ObjectData::Inline { .. })
+    }
     pub fn inlined(&self) -> Option<&Vec<u8>> {
         match &self.data {
             ObjectData::Inline { data } => Some(data),

src/metastore/object.rs

@@ -254,6 +254,14 @@ impl MetaStore for FjallStore {
 
         Box::new(FjallTransaction::new(tx, Arc::new(self.clone())))
     }
+
+    fn num_keys(&self) -> (usize, usize, usize) {
+        unimplemented!();
+    }
+
+    fn disk_space(&self) -> u64 {
+        self.keyspace.disk_space()
+    }
 }
 
 pub struct FjallTransaction {

src/metastore/stores/fjall.rs

@@ -209,6 +209,18 @@ impl MetaStore for FjallStoreNotx {
     fn begin_transaction(&self) -> Box<dyn Transaction> {
         Box::new(FjallNoTransaction::new(Arc::new(self.clone())))
     }
+
+    fn num_keys(&self) -> (usize, usize, usize) {
+        (
+            self.bucket_partition.approximate_len(),
+            self.block_partition.approximate_len(),
+            self.path_partition.approximate_len(),
+        )
+    }
+
+    fn disk_space(&self) -> u64 {
+        self.keyspace.disk_space()
+    }
 }
 
 // FjallNoTransaction is fjall without real transaction support.

src/metastore/stores/fjall_notx.rs

@@ -72,6 +72,12 @@ pub trait MetaStore: Send + Sync + Debug + 'static {
     fn delete_object(&self, bucket: &str, key: &str) -> Result<Vec<Block>, MetaError>;
 
     fn begin_transaction(&self) -> Box<dyn Transaction>;
+
+    // returns the number of keys of the bucket, block, and path trees.
+    fn num_keys(&self) -> (usize, usize, usize);
+
+    // returns the disk space used by the metadata store.
+    fn disk_space(&self) -> u64;
 }
 
 pub trait Transaction: Send + Sync {

src/metastore/traits.rs

@@ -347,8 +347,9 @@ impl S3 for S3FS {
         }
 
         // load metadata
-        let obj_meta = match self.casfs.get_object_meta(&bucket, &key) {
-            Ok(Some(obj_meta)) => obj_meta,
+
+        let (obj_meta, blocks) = match self.casfs.get_object_blocks(&bucket, &key) {
+            Ok(Some((obj_meta, blocks))) => (obj_meta, blocks),
             Ok(None) => {
                 return Err(s3_error!(NoSuchKey, "Object does not exist"));
             }
@@ -387,31 +388,13 @@ impl S3 for S3FS {
         };
 
         // load the data
-        let block_map = try_!(self.casfs.block_tree());
-        let mut paths = Vec::with_capacity(obj_meta.blocks().len());
+        let mut paths = Vec::with_capacity(blocks.len());
         let mut block_size = 0;
-        for block in obj_meta.blocks() {
-            // unwrap here is safe as we only add blocks to the list of an object if they are
-            // corectly inserted in the block map
-            let block_meta = match block_map.get_block(block) {
-                Ok(Some(block_meta)) => block_meta,
-                Ok(None) => {
-                    return Err(s3_error!(
-                        InternalError,
-                        "Could not find block in block map"
-                    ));
-                }
-                Err(e) => {
-                    return Err(s3_error!(
-                        InternalError,
-                        "Could not get block from block map: {}",
-                        e
-                    ));
-                }
-            };
-            block_size += block_meta.size();
-            paths.push((block_meta.disk_path(self.root.clone()), block_meta.size()));
+        for block in blocks {
+            block_size += block.size();
+            paths.push((block.disk_path(self.root.clone()), block.size()));
         }
+
         debug_assert!(obj_meta.size() as usize == block_size);
         let block_stream = BlockStream::new(paths, block_size, range, self.metrics.clone());
         let stream = StreamingBlob::wrap(block_stream);