Cargo.lock

@@ -403,10 +403,49 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a0610544180c38b88101fecf2dd634b174a62eef6946f84dfc6a7127512b381c"
 dependencies = [
  "bitflags",
- "textwrap",
+ "textwrap 0.11.0",
  "unicode-width",
 ]
 
+[[package]]
+name = "clap"
+version = "3.1.18"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d2dbdf4bdacb33466e854ce889eee8dfd5729abf7ccd7664d0a2d60cd384440b"
+dependencies = [
+ "atty",
+ "bitflags",
+ "clap_derive",
+ "clap_lex",
+ "indexmap",
+ "lazy_static",
+ "strsim",
+ "termcolor",
+ "textwrap 0.15.0",
+]
+
+[[package]]
+name = "clap_derive"
+version = "3.1.18"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "25320346e922cffe59c0bbc5410c8d8784509efb321488971081313cb1e1a33c"
+dependencies = [
+ "heck 0.4.0",
+ "proc-macro-error",
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
+[[package]]
+name = "clap_lex"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a37c35f1112dad5e6e0b1adaff798507497a18fceeb30cceb3bae7d1427b9213"
+dependencies = [
+ "os_str_bytes",
+]
+
 [[package]]
 name = "cloudabi"
 version = "0.0.3"
@@ -504,6 +543,16 @@ dependencies = [
  "subtle",
 ]
 
+[[package]]
+name = "crypto-mac"
+version = "0.11.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b1d1a86f49236c215f271d40892d5fc950490551400b02ef360692c29815c714"
+dependencies = [
+ "generic-array",
+ "subtle",
+]
+
 [[package]]
 name = "ct-logs"
 version = "0.8.0"
@@ -848,7 +897,7 @@ dependencies = [
  "garage_web",
  "git-version",
  "hex",
- "hmac",
+ "hmac 0.10.1",
  "http",
  "hyper",
  "kuska-sodiumoxide",
@@ -904,7 +953,7 @@ dependencies = [
  "garage_table 0.7.0",
  "garage_util 0.7.0",
  "hex",
- "hmac",
+ "hmac 0.10.1",
  "http",
  "http-range",
  "httpdate 0.3.2",
@@ -1261,6 +1310,12 @@ dependencies = [
  "unicode-segmentation",
 ]
 
+[[package]]
+name = "heck"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2540771e65fc8cb83cd6e8a237f70c319bd5c29f78ed1084ba5d50eeac86f7f9"
+
 [[package]]
 name = "hermit-abi"
 version = "0.1.19"
@@ -1296,6 +1351,16 @@ dependencies = [
  "digest",
 ]
 
+[[package]]
+name = "hmac"
+version = "0.11.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2a2a2320eb7ec0ebe8da8f744d7812d9fc4cb4d09344ac01898dbcb6a20ae69b"
+dependencies = [
+ "crypto-mac 0.11.1",
+ "digest",
+]
+
 [[package]]
 name = "http"
 version = "0.2.6"
@@ -1523,6 +1588,23 @@ dependencies = [
  "serde_json",
 ]
 
+[[package]]
+name = "k2v-client"
+version = "0.1.0"
+dependencies = [
+ "base64",
+ "clap 3.1.18",
+ "garage_util 0.7.0",
+ "http",
+ "rusoto_core",
+ "rusoto_credential",
+ "rusoto_signature",
+ "serde",
+ "serde_json",
+ "thiserror",
+ "tokio",
+]
+
 [[package]]
 name = "k8s-openapi"
 version = "0.13.1"
@@ -2057,6 +2139,12 @@ dependencies = [
  "num-traits",
 ]
 
+[[package]]
+name = "os_str_bytes"
+version = "6.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "029d8d0b2f198229de29dca79676f2738ff952edf3fde542eb8bf94d8c21b435"
+
 [[package]]
 name = "parking_lot"
 version = "0.11.2"
@@ -2306,7 +2394,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "62941722fb675d463659e49c4f3fe1fe792ff24fe5bbaa9c08cd3b98a1c354f5"
 dependencies = [
  "bytes 1.1.0",
- "heck",
+ "heck 0.3.3",
  "itertools 0.10.3",
  "lazy_static",
  "log",
@@ -2533,6 +2621,75 @@ dependencies = [
  "xmlparser",
 ]
 
+[[package]]
+name = "rusoto_core"
+version = "0.48.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1db30db44ea73551326269adcf7a2169428a054f14faf9e1768f2163494f2fa2"
+dependencies = [
+ "async-trait",
+ "base64",
+ "bytes 1.1.0",
+ "crc32fast",
+ "futures",
+ "http",
+ "hyper",
+ "hyper-tls",
+ "lazy_static",
+ "log",
+ "rusoto_credential",
+ "rusoto_signature",
+ "rustc_version",
+ "serde",
+ "serde_json",
+ "tokio",
+ "xml-rs",
+]
+
+[[package]]
+name = "rusoto_credential"
+version = "0.48.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ee0a6c13db5aad6047b6a44ef023dbbc21a056b6dab5be3b79ce4283d5c02d05"
+dependencies = [
+ "async-trait",
+ "chrono",
+ "dirs-next",
+ "futures",
+ "hyper",
+ "serde",
+ "serde_json",
+ "shlex",
+ "tokio",
+ "zeroize",
+]
+
+[[package]]
+name = "rusoto_signature"
+version = "0.48.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a5ae95491c8b4847931e291b151127eccd6ff8ca13f33603eb3d0035ecb05272"
+dependencies = [
+ "base64",
+ "bytes 1.1.0",
+ "chrono",
+ "digest",
+ "futures",
+ "hex",
+ "hmac 0.11.0",
+ "http",
+ "hyper",
+ "log",
+ "md-5",
+ "percent-encoding",
+ "pin-project-lite",
+ "rusoto_credential",
+ "rustc_version",
+ "serde",
+ "sha2",
+ "tokio",
+]
+
 [[package]]
 name = "rustc_version"
 version = "0.4.0"
@@ -2669,9 +2826,9 @@ checksum = "a4a3381e03edd24287172047536f20cabde766e2cd3e65e6b00fb3af51c4f38d"
 
 [[package]]
 name = "serde"
-version = "1.0.136"
+version = "1.0.137"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ce31e24b01e1e524df96f1c2fdd054405f8d7376249a5110886fb4b658484789"
+checksum = "61ea8d54c77f8315140a05f4c7237403bf38b72704d031543aa1d16abbf517d1"
 dependencies = [
  "serde_derive",
 ]
@@ -2697,9 +2854,9 @@ dependencies = [
 
 [[package]]
 name = "serde_derive"
-version = "1.0.136"
+version = "1.0.137"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "08597e7152fcd306f41838ed3e37be9eaeed2b61c42e2117266a554fab4662f9"
+checksum = "1f26faba0c3959972377d3b2d306ee9f71faee9714294e41bb777f83f88578be"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -2719,9 +2876,9 @@ dependencies = [
 
 [[package]]
 name = "serde_json"
-version = "1.0.79"
+version = "1.0.81"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8e8d9fa5c3b304765ce1fd9c4c8a3de2c8db365a5b91be52f186efc675681d95"
+checksum = "9b7ce2b32a1aed03c558dc61a5cd328f15aff2dbc17daad8fb8af04d2100e15c"
 dependencies = [
  "indexmap",
  "itoa",
@@ -2754,6 +2911,12 @@ dependencies = [
  "opaque-debug",
 ]
 
+[[package]]
+name = "shlex"
+version = "1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "43b2853a4d09f215c24cc5489c992ce46052d359b5109343cbafbf26bc62f8a3"
+
 [[package]]
 name = "signal-hook-registry"
 version = "1.4.0"
@@ -2876,7 +3039,7 @@ version = "0.3.26"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0c6b5c64445ba8094a6ab0c3cd2ad323e07171012d9c98b0b15651daf1787a10"
 dependencies = [
- "clap",
+ "clap 2.34.0",
  "lazy_static",
  "structopt-derive",
 ]
@@ -2887,7 +3050,7 @@ version = "0.4.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "dcb5ae327f9cc13b68763b5749770cb9e048a99bd9dfdfa58d0cf05d5f64afe0"
 dependencies = [
- "heck",
+ "heck 0.3.3",
  "proc-macro-error",
  "proc-macro2",
  "quote",
@@ -2902,9 +3065,9 @@ checksum = "6bdef32e8150c2a081110b42772ffe7d7c9032b606bc226c8260fd97e0976601"
 
 [[package]]
 name = "syn"
-version = "1.0.89"
+version = "1.0.94"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ea297be220d52398dcc07ce15a209fce436d361735ac1db700cab3b6cdfb9f54"
+checksum = "a07e33e919ebcd69113d5be0e4d70c5707004ff45188910106854f38b960df4a"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -2956,19 +3119,25 @@ dependencies = [
 ]
 
 [[package]]
-name = "thiserror"
+name = "textwrap"
-version = "1.0.30"
+version = "0.15.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "854babe52e4df1653706b98fcfc05843010039b406875930a70e4d9644e5c417"
+checksum = "b1141d4d61095b28419e22cb0bbf02755f5e54e0526f97f1e3d1d160e60885fb"
+
+[[package]]
+name = "thiserror"
+version = "1.0.31"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bd829fe32373d27f76265620b5309d0340cb8550f523c1dda251d6298069069a"
 dependencies = [
  "thiserror-impl",
 ]
 
 [[package]]
 name = "thiserror-impl"
-version = "1.0.30"
+version = "1.0.31"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "aa32fd3f627f367fe16f893e2597ae3c05020f8bba2666a4e6ea73d377e5714b"
+checksum = "0396bc89e626244658bef819e22d0cc459e795a5ebe878e6ec336d1674a8d79a"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -3535,6 +3704,12 @@ version = "0.32.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "504a2476202769977a040c6364301a3f65d0cc9e3fb08600b2bda150a0488316"
 
+[[package]]
+name = "xml-rs"
+version = "0.8.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d2d7d3948613f75c98fd9328cfdcc45acc4d360655289d0a7d4ec931392200a3"
+
 [[package]]
 name = "xmlparser"
 version = "0.13.3"

Cargo.toml

@@ -8,9 +8,12 @@ members = [
 	"src/admin",
 	"src/api",
 	"src/web",
-	"src/garage"
+	"src/garage",
+	"src/k2v-client",
 ]
 
+default-members = ["src/garage"]
+
 [profile.dev]
 lto = "off"
 

shell.nix

@@ -79,6 +79,8 @@ function refresh_toolchain {
      pkgs.rustfmt
      pkgs.perl
      pkgs.protobuf
+     pkgs.pkg-config
+     pkgs.openssl
      cargo2nix.packages.x86_64-linux.cargo2nix
     ] else [])
    ++

src/garage/cli/cmd.rs

@@ -1,6 +1,7 @@
 use std::collections::HashSet;
 
 use garage_util::error::*;
+use garage_util::formater::format_table;
 
 use garage_rpc::layout::*;
 use garage_rpc::system::*;

src/garage/cli/layout.rs

@@ -1,6 +1,7 @@
 use garage_util::crdt::Crdt;
 use garage_util::data::*;
 use garage_util::error::*;
+use garage_util::formater::format_table;
 
 use garage_rpc::layout::*;
 use garage_rpc::system::*;

src/garage/cli/util.rs

@@ -3,6 +3,7 @@ use std::collections::HashMap;
 use garage_util::crdt::*;
 use garage_util::data::Uuid;
 use garage_util::error::*;
+use garage_util::formater::format_table;
 
 use garage_model::bucket_table::*;
 use garage_model::key_table::*;
@@ -173,35 +174,6 @@ pub fn print_bucket_info(bucket: &Bucket, relevant_keys: &HashMap<String, Key>)
 	};
 }
 
-pub fn format_table(data: Vec<String>) {
-	let data = data
-		.iter()
-		.map(|s| s.split('\t').collect::<Vec<_>>())
-		.collect::<Vec<_>>();
-
-	let columns = data.iter().map(|row| row.len()).fold(0, std::cmp::max);
-	let mut column_size = vec![0; columns];
-
-	let mut out = String::new();
-
-	for row in data.iter() {
-		for (i, col) in row.iter().enumerate() {
-			column_size[i] = std::cmp::max(column_size[i], col.chars().count());
-		}
-	}
-
-	for row in data.iter() {
-		for (col, col_len) in row[..row.len() - 1].iter().zip(column_size.iter()) {
-			out.push_str(col);
-			(0..col_len - col.chars().count() + 2).for_each(|_| out.push(' '));
-		}
-		out.push_str(row[row.len() - 1]);
-		out.push('\n');
-	}
-
-	print!("{}", out);
-}
-
 pub fn find_matching_node(
 	cand: impl std::iter::Iterator<Item = Uuid>,
 	pattern: &str,

src/k2v-client/Cargo.toml

@@ -0,0 +1,27 @@
+[package]
+name = "k2v-client"
+version = "0.1.0"
+edition = "2018"
+
+[dependencies]
+base64 = "0.13.0"
+http = "0.2.6"
+rusoto_core = "0.48.0"
+rusoto_credential = "0.48.0"
+rusoto_signature = "0.48.0"
+serde = "1.0.137"
+serde_json = "1.0.81"
+thiserror = "1.0.31"
+tokio = "1.17.0"
+
+# cli deps
+clap = { version = "3.1.18", optional = true, features = ["derive", "env"] }
+garage_util = { path = "../util", optional = true }
+
+
+[features]
+cli = ["clap", "tokio/fs", "tokio/io-std", "garage_util"]
+
+[[bin]]
+name = "k2v-cli"
+required-features = ["cli"]

src/k2v-client/README.md

@@ -0,0 +1,25 @@
+Example usage:
+```sh
+# all these values can be provided on the cli instead
+export AWS_ACCESS_KEY_ID=GK123456
+export AWS_SECRET_ACCESS_KEY=0123..789
+export AWS_REGION=garage
+export K2V_ENDPOINT=http://172.30.2.1:3903
+export K2V_BUCKET=my-bucket
+
+cargo run --features=cli -- read-range my-partition-key --all
+
+cargo run --features=cli -- insert my-partition-key my-sort-key --text "my string1"
+cargo run --features=cli -- insert my-partition-key my-sort-key --text "my string2"
+cargo run --features=cli -- insert my-partition-key my-sort-key2 --text "my string"
+
+cargo run --features=cli -- read-range my-partition-key --all
+
+causality=$(cargo run --features=cli -- read my-partition-key my-sort-key2 -b | head -n1)
+cargo run --features=cli -- delete my-partition-key my-sort-key2 -c $causality
+
+causality=$(cargo run --features=cli -- read my-partition-key my-sort-key -b | head -n1)
+cargo run --features=cli -- insert my-partition-key my-sort-key --text "my string3" -c $causality
+
+cargo run --features=cli -- read-range my-partition-key --all
+```

src/k2v-client/src/bin/k2v-cli.rs

@@ -0,0 +1,466 @@
+use k2v_client::*;
+
+use garage_util::formater::format_table;
+
+use rusoto_core::credential::AwsCredentials;
+use rusoto_core::Region;
+
+use clap::{Parser, Subcommand};
+
+/// K2V command line interface
+#[derive(Parser, Debug)]
+#[clap(author, version, about, long_about = None)]
+struct Args {
+	/// Name of the region to use
+	#[clap(short, long, env = "AWS_REGION", default_value = "garage")]
+	region: String,
+	/// Url of the endpoint to connect to
+	#[clap(short, long, env = "K2V_ENDPOINT")]
+	endpoint: String,
+	/// Access key ID
+	#[clap(short, long, env = "AWS_ACCESS_KEY_ID")]
+	key_id: String,
+	/// Access key ID
+	#[clap(short, long, env = "AWS_SECRET_ACCESS_KEY")]
+	secret: String,
+	/// Bucket name
+	#[clap(short, long, env = "K2V_BUCKET")]
+	bucket: String,
+	#[clap(subcommand)]
+	command: Command,
+}
+
+#[derive(Subcommand, Debug)]
+enum Command {
+	/// Insert a single value
+	Insert {
+		/// Partition key to insert to
+		partition_key: String,
+		/// Sort key to insert to
+		sort_key: String,
+		/// Causality of the insertion
+		#[clap(short, long)]
+		causality: Option<String>,
+		/// Value to insert
+		#[clap(flatten)]
+		value: Value,
+	},
+	/// Read a single value
+	Read {
+		/// Partition key to read from
+		partition_key: String,
+		/// Sort key to read from
+		sort_key: String,
+		/// Output formating
+		#[clap(flatten)]
+		output_kind: ReadOutputKind,
+	},
+	/// Delete a single value
+	Delete {
+		/// Partition key to delete from
+		partition_key: String,
+		/// Sort key to delete from
+		sort_key: String,
+		/// Causality information
+		#[clap(short, long)]
+		causality: String,
+	},
+	/// List partition keys
+	ReadIndex {
+		/// Output formating
+		#[clap(flatten)]
+		output_kind: BatchOutputKind,
+		/// Output only partition keys matching this filter
+		#[clap(flatten)]
+		filter: Filter,
+	},
+	/// Read a range of sort keys
+	ReadRange {
+		/// Partition key to read from
+		partition_key: String,
+		/// Output formating
+		#[clap(flatten)]
+		output_kind: BatchOutputKind,
+		/// Output only sort keys matching this filter
+		#[clap(flatten)]
+		filter: Filter,
+	},
+	/// Delete a range of sort keys
+	DeleteRange {
+		/// Partition key to delete from
+		partition_key: String,
+		/// Output formating
+		#[clap(flatten)]
+		output_kind: BatchOutputKind,
+		/// Delete only sort keys matching this filter
+		#[clap(flatten)]
+		filter: Filter,
+	},
+}
+
+/// Where to read a value from
+#[derive(Parser, Debug)]
+#[clap(group = clap::ArgGroup::new("value").multiple(false).required(true))]
+struct Value {
+	/// Read value from a file. use - to read from stdin
+	#[clap(short, long, group = "value")]
+	file: Option<String>,
+	/// Read a base64 value from commandline
+	#[clap(short, long, group = "value")]
+	b64: Option<String>,
+	/// Read a raw (UTF-8) value from the commandline
+	#[clap(short, long, group = "value")]
+	text: Option<String>,
+}
+
+impl Value {
+	async fn to_data(&self) -> Result<Vec<u8>, Error> {
+		if let Some(ref text) = self.text {
+			Ok(text.as_bytes().to_vec())
+		} else if let Some(ref b64) = self.b64 {
+			base64::decode(b64).map_err(|_| Error::Message("invalid base64 input".into()))
+		} else if let Some(ref path) = self.file {
+			use tokio::io::AsyncReadExt;
+			if path == "-" {
+				let mut file = tokio::io::stdin();
+				let mut vec = Vec::new();
+				file.read_to_end(&mut vec).await?;
+				Ok(vec)
+			} else {
+				let mut file = tokio::fs::File::open(path).await?;
+				let mut vec = Vec::new();
+				file.read_to_end(&mut vec).await?;
+				Ok(vec)
+			}
+		} else {
+			unreachable!("Value must have one option set")
+		}
+	}
+}
+
+#[derive(Parser, Debug)]
+#[clap(group = clap::ArgGroup::new("output-kind").multiple(false).required(false))]
+struct ReadOutputKind {
+	/// Base64 output. Conflicts are line separated, first line is causality token
+	#[clap(short, long, group = "output-kind")]
+	b64: bool,
+	/// Raw output. Conflicts generate error, causality token is not returned
+	#[clap(short, long, group = "output-kind")]
+	raw: bool,
+	/// Human formated output
+	#[clap(short = 'H', long, group = "output-kind")]
+	human: bool,
+	/// JSON formated output
+	#[clap(short, long, group = "output-kind")]
+	json: bool,
+}
+
+impl ReadOutputKind {
+	fn display_output(&self, val: CausalValue) -> ! {
+		use std::io::Write;
+		use std::process::exit;
+
+		if self.json {
+			let stdout = std::io::stdout();
+			serde_json::to_writer_pretty(stdout, &val).unwrap();
+			exit(0);
+		}
+
+		if self.raw {
+			let mut val = val.value;
+			if val.len() != 1 {
+				eprintln!(
+					"Raw mode can only read non-concurent values, found {} values, expected 1",
+					val.len()
+				);
+				exit(1);
+			}
+			let val = val.pop().unwrap();
+			match val {
+				K2vValue::Value(v) => {
+					std::io::stdout().write_all(&v).unwrap();
+					exit(0);
+				}
+				K2vValue::Tombstone => {
+					eprintln!("Expected value, found tombstone");
+					exit(2);
+				}
+			}
+		}
+
+		let causality: String = val.causality.into();
+		if self.b64 {
+			println!("{}", causality);
+			for val in val.value {
+				match val {
+					K2vValue::Value(v) => {
+						println!("{}", base64::encode(&v))
+					}
+					K2vValue::Tombstone => {
+						println!();
+					}
+				}
+			}
+			exit(0);
+		}
+
+		// human
+		println!("causality: {}", causality);
+		println!("values:");
+		for val in val.value {
+			match val {
+				K2vValue::Value(v) => {
+					if let Ok(string) = std::str::from_utf8(&v) {
+						println!("  utf-8: {}", string);
+					} else {
+						println!("  base64: {}", base64::encode(&v));
+					}
+				}
+				K2vValue::Tombstone => {
+					println!("  tombstone");
+				}
+			}
+		}
+		exit(0);
+	}
+}
+
+#[derive(Parser, Debug)]
+#[clap(group = clap::ArgGroup::new("output-kind").multiple(false).required(false))]
+struct BatchOutputKind {
+	/// Human formated output
+	#[clap(short = 'H', long, group = "output-kind")]
+	human: bool,
+	/// JSON formated output
+	#[clap(short, long, group = "output-kind")]
+	json: bool,
+}
+
+/// Filter for batch operations
+#[derive(Parser, Debug)]
+#[clap(group = clap::ArgGroup::new("filter").multiple(true).required(true))]
+struct Filter {
+	/// Match only keys starting with this prefix
+	#[clap(short, long, group = "filter")]
+	prefix: Option<String>,
+	/// Match only keys lexicographically after this key (including this key itself)
+	#[clap(short, long, group = "filter")]
+	start: Option<String>,
+	/// Match only keys lexicographically before this key (excluding this key)
+	#[clap(short, long, group = "filter")]
+	end: Option<String>,
+	/// Only match the first X keys
+	#[clap(short, long)]
+	limit: Option<u64>,
+	/// Return keys in reverse order
+	#[clap(short, long)]
+	reverse: bool,
+	/// Return only keys where conflict happened
+	#[clap(short, long)]
+	conflicts_only: bool,
+	/// Also include keys storing only tombstones
+	#[clap(short, long)]
+	tombstones: bool,
+	/// Return any key
+	#[clap(short, long, group = "filter")]
+	all: bool,
+}
+
+impl Filter {
+	fn k2v_filter(&self) -> k2v_client::Filter<'_> {
+		k2v_client::Filter {
+			start: self.start.as_deref(),
+			end: self.end.as_deref(),
+			prefix: self.prefix.as_deref(),
+			limit: self.limit,
+			reverse: self.reverse,
+		}
+	}
+}
+
+#[tokio::main]
+async fn main() -> Result<(), Error> {
+	let args = Args::parse();
+
+	let region = Region::Custom {
+		name: args.region,
+		endpoint: args.endpoint,
+	};
+
+	let creds = AwsCredentials::new(args.key_id, args.secret, None, None);
+
+	let client = K2vClient::new(region, args.bucket, creds, None)?;
+
+	match args.command {
+		Command::Insert {
+			partition_key,
+			sort_key,
+			causality,
+			value,
+		} => {
+			client
+				.insert_item(
+					&partition_key,
+					&sort_key,
+					value.to_data().await?,
+					causality.map(Into::into),
+				)
+				.await?;
+		}
+		Command::Delete {
+			partition_key,
+			sort_key,
+			causality,
+		} => {
+			client
+				.delete_item(&partition_key, &sort_key, causality.into())
+				.await?;
+		}
+		Command::Read {
+			partition_key,
+			sort_key,
+			output_kind,
+		} => {
+			let res = client.read_item(&partition_key, &sort_key).await?;
+			output_kind.display_output(res);
+		}
+		Command::ReadIndex {
+			output_kind,
+			filter,
+		} => {
+			if filter.conflicts_only || filter.tombstones {
+				return Err(Error::Message(
+					"conlicts-only and tombstones are invalid for read-index".into(),
+				));
+			}
+			let res = client.read_index(filter.k2v_filter()).await?;
+			if output_kind.json {
+				let values = res
+					.items
+					.into_iter()
+					.map(|(k, v)| {
+						let mut value = serde_json::to_value(v).unwrap();
+						value
+							.as_object_mut()
+							.unwrap()
+							.insert("sort_key".to_owned(), k.into());
+						value
+					})
+					.collect::<Vec<_>>();
+				let json = serde_json::json!({
+					"next_key": res.next_start,
+					"values": values,
+				});
+
+				let stdout = std::io::stdout();
+				serde_json::to_writer_pretty(stdout, &json).unwrap();
+			} else {
+				if let Some(next) = res.next_start {
+					println!("next key: {}", next);
+				}
+
+				let mut to_print = Vec::new();
+				to_print.push(format!("key:\tentries\tconflicts\tvalues\tbytes"));
+				for (k, v) in res.items {
+					to_print.push(format!(
+						"{}\t{}\t{}\t{}\t{}",
+						k, v.entries, v.conflicts, v.values, v.bytes
+					));
+				}
+				format_table(to_print);
+			}
+		}
+		Command::ReadRange {
+			partition_key,
+			output_kind,
+			filter,
+		} => {
+			let op = BatchReadOp {
+				partition_key: &partition_key,
+				filter: filter.k2v_filter(),
+				conflicts_only: filter.conflicts_only,
+				tombstones: filter.tombstones,
+				single_item: false,
+			};
+			let mut res = client.read_batch(&[op]).await?;
+			let res = res.pop().unwrap();
+			if output_kind.json {
+				let values = res
+					.items
+					.into_iter()
+					.map(|(k, v)| {
+						let mut value = serde_json::to_value(v).unwrap();
+						value
+							.as_object_mut()
+							.unwrap()
+							.insert("sort_key".to_owned(), k.into());
+						value
+					})
+					.collect::<Vec<_>>();
+				let json = serde_json::json!({
+					"next_key": res.next_start,
+					"values": values,
+				});
+
+				let stdout = std::io::stdout();
+				serde_json::to_writer_pretty(stdout, &json).unwrap();
+			} else {
+				if let Some(next) = res.next_start {
+					println!("next key: {}", next);
+				}
+				for (key, values) in res.items {
+					println!("key: {}", key);
+					let causality: String = values.causality.into();
+					println!("causality: {}", causality);
+					for value in values.value {
+						match value {
+							K2vValue::Value(v) => {
+								if let Ok(string) = std::str::from_utf8(&v) {
+									println!("  value(utf-8): {}", string);
+								} else {
+									println!("  value(base64): {}", base64::encode(&v));
+								}
+							}
+							K2vValue::Tombstone => {
+								println!("  tombstone");
+							}
+						}
+					}
+				}
+			}
+		}
+		Command::DeleteRange {
+			partition_key,
+			output_kind,
+			filter,
+		} => {
+			let op = BatchDeleteOp {
+				partition_key: &partition_key,
+				prefix: filter.prefix.as_deref(),
+				start: filter.start.as_deref(),
+				end: filter.end.as_deref(),
+				single_item: false,
+			};
+			if filter.reverse
+				|| filter.conflicts_only
+				|| filter.tombstones
+				|| filter.limit.is_some()
+			{
+				return Err(Error::Message(
+					"limit, conlicts-only, reverse and tombstones are invalid for delete-range"
+						.into(),
+				));
+			}
+
+			let res = client.delete_batch(&[op]).await?;
+
+			if output_kind.json {
+				println!("{}", res[0]);
+			} else {
+				println!("deleted {} keys", res[0]);
+			}
+		}
+	}
+
+	Ok(())
+}

src/k2v-client/src/error.rs

@@ -0,0 +1,22 @@
+use std::borrow::Cow;
+
+use thiserror::Error;
+
+/// Errors returned by this crate
+#[derive(Error, Debug)]
+pub enum Error {
+	#[error("received invalid response: {0}")]
+	InvalidResponse(Cow<'static, str>),
+	#[error("not found")]
+	NotFound,
+	#[error("io error: {0}")]
+	IoError(#[from] std::io::Error),
+	#[error("rusoto tls error: {0}")]
+	RusotoTls(#[from] rusoto_core::request::TlsError),
+	#[error("rusoto http error: {0}")]
+	RusotoHttp(#[from] rusoto_core::HttpDispatchError),
+	#[error("deserialization error: {0}")]
+	Deserialization(#[from] serde_json::Error),
+	#[error("{0}")]
+	Message(Cow<'static, str>),
+}

src/k2v-client/src/lib.rs

@@ -0,0 +1,566 @@
+use std::collections::BTreeMap;
+use std::time::Duration;
+
+use http::header::{ACCEPT, CONTENT_LENGTH, CONTENT_TYPE};
+use http::status::StatusCode;
+use http::HeaderMap;
+
+use rusoto_core::{ByteStream, DispatchSignedRequest, HttpClient};
+use rusoto_credential::AwsCredentials;
+use rusoto_signature::region::Region;
+use rusoto_signature::signature::SignedRequest;
+use serde::de::Error as DeError;
+use serde::{Deserialize, Deserializer, Serialize, Serializer};
+
+use tokio::io::AsyncReadExt;
+
+mod error;
+
+pub use error::Error;
+
+const DEFAULT_TIMEOUT: Duration = Duration::from_secs(5);
+const DEFAULT_POLL_TIMEOUT: Duration = Duration::from_secs(300);
+const SERVICE: &str = "k2v";
+const GARAGE_CAUSALITY_TOKEN: &str = "X-Garage-Causality-Token";
+
+/// Client used to query a K2V server.
+pub struct K2vClient {
+	region: Region,
+	bucket: String,
+	creds: AwsCredentials,
+	client: HttpClient,
+}
+
+impl K2vClient {
+	/// Create a new K2V client.
+	pub fn new(
+		region: Region,
+		bucket: String,
+		creds: AwsCredentials,
+		user_agent: Option<String>,
+	) -> Result<Self, Error> {
+		let mut client = HttpClient::new()?;
+		if let Some(ua) = user_agent {
+			client.local_agent_prepend(ua);
+		} else {
+			client.local_agent_prepend(format!("k2v/{}", env!("CARGO_PKG_VERSION")));
+		}
+		Ok(K2vClient {
+			region,
+			bucket,
+			creds,
+			client,
+		})
+	}
+
+	/// Perform a ReadItem request, reading the value(s) stored for a single pk+sk.
+	pub async fn read_item(
+		&self,
+		partition_key: &str,
+		sort_key: &str,
+	) -> Result<CausalValue, Error> {
+		let mut req = SignedRequest::new(
+			"GET",
+			SERVICE,
+			&self.region,
+			&format!("/{}/{}", self.bucket, partition_key),
+		);
+		req.add_param("sort_key", sort_key);
+		req.add_header(ACCEPT, "application/octet-stream, application/json");
+
+		let res = self.dispatch(req, None).await?;
+
+		let causality = res
+			.causality_token
+			.ok_or_else(|| Error::InvalidResponse("missing causality token".into()))?;
+
+		if res.status == StatusCode::NO_CONTENT {
+			return Ok(CausalValue {
+				causality,
+				value: vec![K2vValue::Tombstone],
+			});
+		}
+
+		match res.content_type.as_deref() {
+			Some("application/octet-stream") => Ok(CausalValue {
+				causality,
+				value: vec![K2vValue::Value(res.body)],
+			}),
+			Some("application/json") => {
+				let value = serde_json::from_slice(&res.body)?;
+				Ok(CausalValue { causality, value })
+			}
+			Some(ct) => Err(Error::InvalidResponse(
+				format!("invalid content type: {}", ct).into(),
+			)),
+			None => Err(Error::InvalidResponse("missing content type".into())),
+		}
+	}
+
+	/// Perform a PollItem request, waiting for the value(s) stored for a single pk+sk to be
+	/// updated.
+	pub async fn poll_item(
+		&self,
+		partition_key: &str,
+		sort_key: &str,
+		causality: CausalityToken,
+		timeout: Option<Duration>,
+	) -> Result<Option<CausalValue>, Error> {
+		let timeout = timeout.unwrap_or(DEFAULT_POLL_TIMEOUT);
+
+		let mut req = SignedRequest::new(
+			"GET",
+			SERVICE,
+			&self.region,
+			&format!("/{}/{}", self.bucket, partition_key),
+		);
+		req.add_param("sort_key", sort_key);
+		req.add_param("causality_token", &causality.0);
+		req.add_param("timeout", &timeout.as_secs().to_string());
+		req.add_header(ACCEPT, "application/octet-stream, application/json");
+
+		let res = self.dispatch(req, Some(timeout + DEFAULT_TIMEOUT)).await?;
+
+		let causality = res
+			.causality_token
+			.ok_or_else(|| Error::InvalidResponse("missing causality token".into()))?;
+
+		if res.status == StatusCode::NOT_MODIFIED {
+			return Ok(None);
+		}
+
+		if res.status == StatusCode::NO_CONTENT {
+			return Ok(Some(CausalValue {
+				causality,
+				value: vec![K2vValue::Tombstone],
+			}));
+		}
+
+		match res.content_type.as_deref() {
+			Some("application/octet-stream") => Ok(Some(CausalValue {
+				causality,
+				value: vec![K2vValue::Value(res.body)],
+			})),
+			Some("application/json") => {
+				let value = serde_json::from_slice(&res.body)?;
+				Ok(Some(CausalValue { causality, value }))
+			}
+			Some(ct) => Err(Error::InvalidResponse(
+				format!("invalid content type: {}", ct).into(),
+			)),
+			None => Err(Error::InvalidResponse("missing content type".into())),
+		}
+	}
+
+	/// Perform an InsertItem request, inserting a value for a single pk+sk.
+	pub async fn insert_item(
+		&self,
+		partition_key: &str,
+		sort_key: &str,
+		value: Vec<u8>,
+		causality: Option<CausalityToken>,
+	) -> Result<(), Error> {
+		let mut req = SignedRequest::new(
+			"PUT",
+			SERVICE,
+			&self.region,
+			&format!("/{}/{}", self.bucket, partition_key),
+		);
+		req.add_param("sort_key", sort_key);
+		req.set_payload(Some(value));
+
+		if let Some(causality) = causality {
+			req.add_header(GARAGE_CAUSALITY_TOKEN, &causality.0);
+		}
+
+		self.dispatch(req, None).await?;
+		Ok(())
+	}
+
+	/// Perform a DeleteItem request, deleting the value(s) stored for a single pk+sk.
+	pub async fn delete_item(
+		&self,
+		partition_key: &str,
+		sort_key: &str,
+		causality: CausalityToken,
+	) -> Result<(), Error> {
+		let mut req = SignedRequest::new(
+			"DELETE",
+			SERVICE,
+			&self.region,
+			&format!("/{}/{}", self.bucket, partition_key),
+		);
+		req.add_param("sort_key", sort_key);
+		req.add_header(GARAGE_CAUSALITY_TOKEN, &causality.0);
+
+		self.dispatch(req, None).await?;
+		Ok(())
+	}
+
+	/// Perform a ReadIndex request, listing partition key which have at least one associated
+	/// sort key, and which matches the filter.
+	pub async fn read_index(
+		&self,
+		filter: Filter<'_>,
+	) -> Result<PaginatedRange<PartitionInfo>, Error> {
+		let mut req =
+			SignedRequest::new("GET", SERVICE, &self.region, &format!("/{}", self.bucket));
+		filter.insert_params(&mut req);
+
+		let res = self.dispatch(req, None).await?;
+
+		let resp: ReadIndexResponse = serde_json::from_slice(&res.body)?;
+
+		let items = resp
+			.partition_keys
+			.into_iter()
+			.map(|ReadIndexItem { pk, info }| (pk, info))
+			.collect();
+
+		Ok(PaginatedRange {
+			items,
+			next_start: resp.next_start,
+		})
+	}
+
+	/// Perform an InsertBatch request, inserting multiple values at once. Note: this operation is
+	/// *not* atomic: it is possible for some sub-operations to fails and others to success. In
+	/// that case, failure is reported.
+	pub async fn insert_batch(&self, operations: &[BatchInsertOp<'_>]) -> Result<(), Error> {
+		let mut req =
+			SignedRequest::new("POST", SERVICE, &self.region, &format!("/{}", self.bucket));
+
+		let payload = serde_json::to_vec(operations)?;
+		req.set_payload(Some(payload));
+		self.dispatch(req, None).await?;
+		Ok(())
+	}
+
+	/// Perform a ReadBatch request, reading multiple values or range of values at once.
+	pub async fn read_batch(
+		&self,
+		operations: &[BatchReadOp<'_>],
+	) -> Result<Vec<PaginatedRange<CausalValue>>, Error> {
+		let mut req =
+			SignedRequest::new("POST", SERVICE, &self.region, &format!("/{}", self.bucket));
+		req.add_param("search", "");
+
+		let payload = serde_json::to_vec(operations)?;
+		req.set_payload(Some(payload));
+		let res = self.dispatch(req, None).await?;
+
+		let resp: Vec<BatchReadResponse> = serde_json::from_slice(&res.body)?;
+
+		Ok(resp
+			.into_iter()
+			.map(|e| PaginatedRange {
+				items: e
+					.items
+					.into_iter()
+					.map(|BatchReadItem { sk, ct, v }| {
+						(
+							sk,
+							CausalValue {
+								causality: ct,
+								value: v,
+							},
+						)
+					})
+					.collect(),
+				next_start: e.next_start,
+			})
+			.collect())
+	}
+
+	/// Perform a DeleteBatch request, deleting mutiple values or range of values at once, without
+	/// providing causality information.
+	pub async fn delete_batch(&self, operations: &[BatchDeleteOp<'_>]) -> Result<Vec<u64>, Error> {
+		let mut req =
+			SignedRequest::new("POST", SERVICE, &self.region, &format!("/{}", self.bucket));
+		req.add_param("delete", "");
+
+		let payload = serde_json::to_vec(operations)?;
+		req.set_payload(Some(payload));
+		let res = self.dispatch(req, None).await?;
+
+		let resp: Vec<BatchDeleteResponse> = serde_json::from_slice(&res.body)?;
+
+		Ok(resp.into_iter().map(|r| r.deleted_items).collect())
+	}
+
+	async fn dispatch(
+		&self,
+		mut req: SignedRequest,
+		timeout: Option<Duration>,
+	) -> Result<Response, Error> {
+		req.sign(&self.creds);
+		let mut res = self
+			.client
+			.dispatch(req, Some(timeout.unwrap_or(DEFAULT_TIMEOUT)))
+			.await?;
+
+		let causality_token = res
+			.headers
+			.remove(GARAGE_CAUSALITY_TOKEN)
+			.map(CausalityToken);
+		let content_type = res.headers.remove(CONTENT_TYPE);
+
+		let body = match res.status {
+			StatusCode::OK => read_body(&mut res.headers, res.body).await?,
+			StatusCode::NO_CONTENT => Vec::new(),
+			StatusCode::NOT_FOUND => return Err(Error::NotFound),
+			StatusCode::NOT_MODIFIED => Vec::new(),
+			_ => {
+				return Err(Error::InvalidResponse(
+					format!("invalid error code: {}", res.status).into(),
+				))
+			}
+		};
+
+		Ok(Response {
+			body,
+			status: res.status,
+			causality_token,
+			content_type,
+		})
+	}
+}
+
+async fn read_body(headers: &mut HeaderMap<String>, body: ByteStream) -> Result<Vec<u8>, Error> {
+	let body_len = headers
+		.get(CONTENT_LENGTH)
+		.and_then(|h| h.parse().ok())
+		.unwrap_or(0);
+	let mut res = Vec::with_capacity(body_len);
+	body.into_async_read().read_to_end(&mut res).await?;
+	Ok(res)
+}
+
+/// An opaque token used to convey causality between operations.
+#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
+#[serde(transparent)]
+pub struct CausalityToken(String);
+
+impl From<String> for CausalityToken {
+	fn from(v: String) -> Self {
+		CausalityToken(v)
+	}
+}
+
+impl From<CausalityToken> for String {
+	fn from(v: CausalityToken) -> Self {
+		v.0
+	}
+}
+
+/// A value in K2V. can be either a binary value, or a tombstone.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum K2vValue {
+	Tombstone,
+	Value(Vec<u8>),
+}
+
+impl From<Vec<u8>> for K2vValue {
+	fn from(v: Vec<u8>) -> Self {
+		K2vValue::Value(v)
+	}
+}
+
+impl From<Option<Vec<u8>>> for K2vValue {
+	fn from(v: Option<Vec<u8>>) -> Self {
+		match v {
+			Some(v) => K2vValue::Value(v),
+			None => K2vValue::Tombstone,
+		}
+	}
+}
+
+impl<'de> Deserialize<'de> for K2vValue {
+	fn deserialize<D>(d: D) -> Result<Self, D::Error>
+	where
+		D: Deserializer<'de>,
+	{
+		let val: Option<&str> = Option::deserialize(d)?;
+		Ok(match val {
+			Some(s) => {
+				K2vValue::Value(base64::decode(s).map_err(|_| DeError::custom("invalid base64"))?)
+			}
+			None => K2vValue::Tombstone,
+		})
+	}
+}
+
+impl Serialize for K2vValue {
+	fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+	where
+		S: Serializer,
+	{
+		match self {
+			K2vValue::Tombstone => serializer.serialize_none(),
+			K2vValue::Value(v) => {
+				let b64 = base64::encode(v);
+				serializer.serialize_str(&b64)
+			}
+		}
+	}
+}
+
+/// A set of K2vValue and associated causality information.
+#[derive(Debug, Clone, Serialize)]
+pub struct CausalValue {
+	pub causality: CausalityToken,
+	pub value: Vec<K2vValue>,
+}
+
+/// Result of paginated requests.
+#[derive(Debug, Clone)]
+pub struct PaginatedRange<V> {
+	pub items: BTreeMap<String, V>,
+	pub next_start: Option<String>,
+}
+
+/// Filter for batch operations.
+#[derive(Debug, Default, Clone, Deserialize, Serialize)]
+pub struct Filter<'a> {
+	pub start: Option<&'a str>,
+	pub end: Option<&'a str>,
+	pub prefix: Option<&'a str>,
+	pub limit: Option<u64>,
+	#[serde(default)]
+	pub reverse: bool,
+}
+
+impl<'a> Filter<'a> {
+	fn insert_params(&self, req: &mut SignedRequest) {
+		if let Some(start) = &self.start {
+			req.add_param("start", start);
+		}
+		if let Some(end) = &self.end {
+			req.add_param("end", end);
+		}
+		if let Some(prefix) = &self.prefix {
+			req.add_param("prefix", prefix);
+		}
+		if let Some(limit) = &self.limit {
+			req.add_param("limit", &limit.to_string());
+		}
+		if self.reverse {
+			req.add_param("reverse", "true");
+		}
+	}
+}
+
+#[derive(Debug, Clone, Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct ReadIndexResponse<'a> {
+	#[serde(flatten, borrow)]
+	#[allow(dead_code)]
+	filter: Filter<'a>,
+	partition_keys: Vec<ReadIndexItem>,
+	#[allow(dead_code)]
+	more: bool,
+	next_start: Option<String>,
+}
+
+#[derive(Debug, Clone, Deserialize)]
+struct ReadIndexItem {
+	pk: String,
+	#[serde(flatten)]
+	info: PartitionInfo,
+}
+
+/// Information about data stored with a given partition key.
+#[derive(Debug, Clone, Deserialize, Serialize)]
+pub struct PartitionInfo {
+	pub entries: u64,
+	pub conflicts: u64,
+	pub values: u64,
+	pub bytes: u64,
+}
+
+/// Single sub-operation of an InsertBatch.
+#[derive(Debug, Clone, Serialize)]
+pub struct BatchInsertOp<'a> {
+	#[serde(rename = "pk")]
+	pub partition_key: &'a str,
+	#[serde(rename = "sk")]
+	pub sort_key: &'a str,
+	#[serde(rename = "ct")]
+	pub causality: Option<CausalityToken>,
+	#[serde(rename = "v")]
+	pub value: K2vValue,
+}
+
+/// Single sub-operation of a ReadBatch.
+#[derive(Debug, Default, Clone, Deserialize, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct BatchReadOp<'a> {
+	pub partition_key: &'a str,
+	#[serde(flatten, borrow)]
+	pub filter: Filter<'a>,
+	#[serde(default)]
+	pub single_item: bool,
+	#[serde(default)]
+	pub conflicts_only: bool,
+	#[serde(default)]
+	pub tombstones: bool,
+}
+
+#[derive(Debug, Clone, Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct BatchReadResponse<'a> {
+	#[serde(flatten, borrow)]
+	#[allow(dead_code)]
+	op: BatchReadOp<'a>,
+	items: Vec<BatchReadItem>,
+	#[allow(dead_code)]
+	more: bool,
+	next_start: Option<String>,
+}
+
+#[derive(Debug, Clone, Deserialize)]
+struct BatchReadItem {
+	sk: String,
+	ct: CausalityToken,
+	v: Vec<K2vValue>,
+}
+
+/// Single sub-operation of a DeleteBatch
+#[derive(Debug, Clone, Deserialize, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct BatchDeleteOp<'a> {
+	pub partition_key: &'a str,
+	pub prefix: Option<&'a str>,
+	pub start: Option<&'a str>,
+	pub end: Option<&'a str>,
+	#[serde(default)]
+	pub single_item: bool,
+}
+
+impl<'a> BatchDeleteOp<'a> {
+	pub fn new(partition_key: &'a str) -> Self {
+		BatchDeleteOp {
+			partition_key,
+			prefix: None,
+			start: None,
+			end: None,
+			single_item: false,
+		}
+	}
+}
+
+#[derive(Debug, Clone, Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct BatchDeleteResponse<'a> {
+	#[serde(flatten, borrow)]
+	#[allow(dead_code)]
+	filter: BatchDeleteOp<'a>,
+	deleted_items: u64,
+}
+
+struct Response {
+	body: Vec<u8>,
+	status: StatusCode,
+	causality_token: Option<CausalityToken>,
+	content_type: Option<String>,
+}

src/util/formater.rs

@@ -0,0 +1,28 @@
+pub fn format_table(data: Vec<String>) {
+	let data = data
+		.iter()
+		.map(|s| s.split('\t').collect::<Vec<_>>())
+		.collect::<Vec<_>>();
+
+	let columns = data.iter().map(|row| row.len()).fold(0, std::cmp::max);
+	let mut column_size = vec![0; columns];
+
+	let mut out = String::new();
+
+	for row in data.iter() {
+		for (i, col) in row.iter().enumerate() {
+			column_size[i] = std::cmp::max(column_size[i], col.chars().count());
+		}
+	}
+
+	for row in data.iter() {
+		for (col, col_len) in row[..row.len() - 1].iter().zip(column_size.iter()) {
+			out.push_str(col);
+			(0..col_len - col.chars().count() + 2).for_each(|_| out.push(' '));
+		}
+		out.push_str(row[row.len() - 1]);
+		out.push('\n');
+	}
+
+	print!("{}", out);
+}

src/util/lib.rs

@@ -8,6 +8,7 @@ pub mod config;
 pub mod crdt;
 pub mod data;
 pub mod error;
+pub mod formater;
 pub mod metrics;
 pub mod persister;
 pub mod sled_counter;