diff --git a/Cargo.lock b/Cargo.lock index 997d6a92..dc9d50cb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1321,6 +1321,7 @@ dependencies = [ "opentelemetry-prometheus", "parse_duration", "serde", + "serde_bytes", "serde_json", "sha1", "sha2", diff --git a/src/garage/Cargo.toml b/src/garage/Cargo.toml index 7d60313e..1d64d838 100644 --- a/src/garage/Cargo.toml +++ b/src/garage/Cargo.toml @@ -49,6 +49,8 @@ structopt.workspace = true git-version.workspace = true serde.workspace = true +serde_json.workspace = true +serde_bytes.workspace = true futures.workspace = true tokio.workspace = true diff --git a/src/garage/cli/structs.rs b/src/garage/cli/structs.rs index 3652ef6b..5ac16eb7 100644 --- a/src/garage/cli/structs.rs +++ b/src/garage/cli/structs.rs @@ -71,6 +71,10 @@ pub enum NodeOperation { /// Connect to Garage node that is currently isolated from the system #[structopt(name = "connect", version = garage_version())] Connect(ConnectNodeOpt), + + /// Dump the content of a metadata table as JSON lines + #[structopt(name = "dump", version = garage_version())] + Dump(DumpNodeOpt), } #[derive(StructOpt, Debug)] @@ -88,6 +92,12 @@ pub struct ConnectNodeOpt { pub(crate) node: String, } +#[derive(StructOpt, Debug)] +pub struct DumpNodeOpt { + /// Name of the data table to dump + pub(crate) what: String, +} + #[derive(StructOpt, Debug)] pub enum LayoutOperation { /// Assign role to Garage node diff --git a/src/garage/main.rs b/src/garage/main.rs index 2703bedd..289a0b12 100644 --- a/src/garage/main.rs +++ b/src/garage/main.rs @@ -145,11 +145,14 @@ async fn main() { let res = match opt.cmd { Command::Server => server::run_server(opt.config_file, opt.secrets).await, Command::OfflineRepair(repair_opt) => { - repair::offline::offline_repair(opt.config_file, opt.secrets, repair_opt).await + repair::offline::offline_repair(opt.config_file, opt.secrets, repair_opt) } Command::ConvertDb(conv_opt) => { cli::convert_db::do_conversion(conv_opt).map_err(From::from) } + Command::Node(NodeOperation::Dump(dump_opt)) => { + repair::offline::dump(opt.config_file, opt.secrets, dump_opt) + } Command::Node(NodeOperation::NodeId(node_id_opt)) => { node_id_command(opt.config_file, node_id_opt.quiet) } diff --git a/src/garage/repair/offline.rs b/src/garage/repair/offline.rs index 45024e71..ebe27a9b 100644 --- a/src/garage/repair/offline.rs +++ b/src/garage/repair/offline.rs @@ -1,14 +1,18 @@ +use std::io::Write; use std::path::PathBuf; +use serde::Serialize; + use garage_util::config::*; use garage_util::error::*; use garage_model::garage::Garage; +use garage_table::{replication::TableReplication, *}; use crate::cli::structs::*; use crate::secrets::{fill_secrets, Secrets}; -pub async fn offline_repair( +pub fn offline_repair( config_file: PathBuf, secrets: Secrets, opt: OfflineRepairOpt, @@ -45,3 +49,100 @@ pub async fn offline_repair( Ok(()) } + +pub fn dump(config_file: PathBuf, secrets: Secrets, opt: DumpNodeOpt) -> Result<(), Error> { + let what = opt.what.as_str(); + + info!("Loading configuration..."); + let config = fill_secrets(read_config(config_file)?, secrets)?; + + info!("Initializing Garage main data store..."); + let garage = Garage::new(config)?; + + match what { + "bucket" | "buckets" => dump_table_inner(&garage.bucket_table), + "bucket_alias" | "bucket_aliases" => dump_table_inner(&garage.bucket_alias_table), + "key" | "keys" => dump_table_inner(&garage.key_table), + "object" | "objects" => dump_table_inner(&garage.object_table), + "object_counter" | "object_counters" => Err(Error::Message( + "object_counters cannot be JSON-serialized".into(), + )), + "mpu" => dump_table_inner(&garage.mpu_table), + "mpu_counter" | "mpu_counters" => Err(Error::Message( + "mpu_counters cannot be JSON-serialized".into(), + )), + "version" | "versions" => dump_table_inner(&garage.version_table), + "block_ref" | "block_refs" => dump_table_inner(&garage.block_ref_table), + #[cfg(feature = "k2v")] + "k2v_item" | "k2v_items" => dump_table_inner(&garage.k2v.item_table), + //#[cfg(feature = "k2v")] + "k2v_counter" | "k2v_counters" => Err(Error::Message( + "k2v_counters cannot be JSON-serialized".into(), + )), + other => { + let mut stdout = std::io::stdout().lock(); + match other { + "cluster_layout" => Err(Error::Message( + "cluster_layout cannot be JSON-serialized".into(), + )), + _ => Err(Error::Message(format!("invalid thing to dump: {}", what))), + } + } + } +} + +#[derive(Serialize)] +struct DumpEntry<'a, T: Serialize> { + #[serde(with = "serde_bytes")] + partition_key: &'a [u8], + #[serde(with = "serde_bytes")] + sort_key: &'a [u8], + entry: &'a T, +} + +fn dump_table_inner(table: &Table) -> Result<(), Error> +where + F: TableSchema, + R: TableReplication, +{ + eprintln!("Dumping table {}...", F::TABLE_NAME); + + let mut stdout = std::io::stdout().lock(); + + for line in table.data.store.iter()? { + let (_k, v) = line?; + let v_dec = table.data.decode_entry(&v)?; + let pkh = v_dec.partition_key().hash(); + let dump_entry = DumpEntry { + partition_key: pkh.as_slice(), + sort_key: v_dec.sort_key().sort_key(), + entry: &v_dec, + }; + dump_line(&mut stdout, dump_entry)?; + } + stdout.flush()?; + + Ok(()) +} + +fn dump_line( + mut stdout: &mut std::io::StdoutLock<'static>, + dump_entry: T, +) -> Result<(), Error> { + let mut ser = serde_json::ser::Serializer::with_formatter(&mut stdout, DumpFormatter); + dump_entry.serialize(&mut ser)?; + stdout.write_all(b"\n")?; + Ok(()) +} + +struct DumpFormatter; +impl serde_json::ser::Formatter for DumpFormatter { + fn write_byte_array(&mut self, writer: &mut W, value: &[u8]) -> std::io::Result<()> + where + W: ?Sized + std::io::Write, + { + writer.write_all(b"\"")?; + writer.write_all(hex::encode(&value).as_bytes())?; + writer.write_all(b"\"") + } +}