From 864e20652458c1a4ae09c882ad3e29d6b0988b06 Mon Sep 17 00:00:00 2001 From: Lars Wirzenius Date: Mon, 28 Sep 2020 09:10:45 +0300 Subject: feat: add rudimentary backup client Also, a bit of logging for server. --- src/bin/obnam-backup.rs | 151 ++++++++++++++++++++++++++++++++++++++++++++++++ src/bin/obnam-server.rs | 13 ++++- 2 files changed, 161 insertions(+), 3 deletions(-) create mode 100644 src/bin/obnam-backup.rs (limited to 'src') diff --git a/src/bin/obnam-backup.rs b/src/bin/obnam-backup.rs new file mode 100644 index 0000000..2b767af --- /dev/null +++ b/src/bin/obnam-backup.rs @@ -0,0 +1,151 @@ +// Read stdin, split into chunks, upload new chunks to chunk server. + +use indicatif::{ProgressBar, ProgressStyle}; +use obnam::chunk::Chunk; +use obnam::chunkmeta::ChunkMeta; +use serde::Deserialize; +use sha2::{Digest, Sha256}; +use std::collections::HashMap; +use std::io::prelude::*; +use std::io::BufReader; +use std::path::{Path, PathBuf}; +use structopt::StructOpt; + +const BUFFER_SIZE: usize = 1024 * 1024; + +#[derive(Debug, StructOpt)] +#[structopt(name = "obnam-backup", about = "Simplistic backup client")] +struct Opt { + #[structopt(parse(from_os_str))] + config: PathBuf, +} + +fn main() -> anyhow::Result<()> { + let opt = Opt::from_args(); + let config = Config::read_config(&opt.config).unwrap(); + let pb = ProgressBar::new_spinner(); + pb.set_style( + ProgressStyle::default_bar() + .template("backing up:\n{bytes} ({bytes_per_sec}) {elapsed} {msg} {spinner}"), + ); + + println!("config: {:?}", config); + + let client = reqwest::blocking::Client::builder() + .danger_accept_invalid_certs(true) + .build()?; + + let stdin = std::io::stdin(); + let mut stdin = BufReader::new(stdin); + let mut dup = 0; + loop { + match read_chunk(&mut stdin)? { + None => break, + Some(chunk) => { + let n = chunk.data().len() as u64; + if !has_chunk(&client, &config, &chunk.meta())? { + pb.inc(n); + upload_chunk(&client, &config, chunk)?; + } else { + dup += n; + } + } + } + } + pb.finish(); + println!( + "read total {} bytes from stdin ({} dup)", + pb.position(), + dup + ); + Ok(()) +} + +#[derive(Debug, Deserialize, Clone)] +pub struct Config { + pub server_name: String, + pub server_port: u16, +} + +impl Config { + pub fn read_config(filename: &Path) -> anyhow::Result { + let config = std::fs::read_to_string(filename)?; + let config: Config = serde_yaml::from_str(&config)?; + Ok(config) + } +} + +fn read_chunk(handle: &mut H) -> anyhow::Result> +where + H: Read + BufRead, +{ + let mut buffer = [0; BUFFER_SIZE]; + let mut used = 0; + + loop { + let n = handle.read(&mut buffer[used..])?; + used += n; + if n == 0 || used == BUFFER_SIZE { + break; + } + } + + if used == 0 { + return Ok(None); + } + + let buffer = &buffer[..used]; + let mut hasher = Sha256::new(); + hasher.update(buffer); + let hash = hasher.finalize(); + let hash = format!("{:x}", hash); + let meta = ChunkMeta::new(&hash); + + let chunk = Chunk::new(meta, buffer.to_vec()); + Ok(Some(chunk)) +} + +fn upload_chunk( + client: &reqwest::blocking::Client, + config: &Config, + chunk: Chunk, +) -> anyhow::Result<()> { + let url = format!( + "http://{}:{}/chunks", + config.server_name, config.server_port + ); + + client + .post(&url) + .header("chunk-meta", chunk.meta().to_json()) + .body(chunk.data().to_vec()) + .send()?; + Ok(()) +} + +fn has_chunk( + client: &reqwest::blocking::Client, + config: &Config, + meta: &ChunkMeta, +) -> anyhow::Result { + let url = format!( + "http://{}:{}/chunks", + config.server_name, config.server_port, + ); + + let req = client + .get(&url) + .query(&[("sha256", meta.sha256())]) + .build()?; + + let res = client.execute(req)?; + let has = if res.status() != 200 { + false + } else { + let text = res.text()?; + let hits: HashMap = serde_json::from_str(&text)?; + !hits.is_empty() + }; + + Ok(has) +} diff --git a/src/bin/obnam-server.rs b/src/bin/obnam-server.rs index 8ad792e..5723b23 100644 --- a/src/bin/obnam-server.rs +++ b/src/bin/obnam-server.rs @@ -1,4 +1,5 @@ use bytes::Bytes; +use log::{debug, error, info}; use obnam::{chunk::Chunk, chunkid::ChunkId, chunkmeta::ChunkMeta, index::Index, store::Store}; use serde::{Deserialize, Serialize}; use std::collections::HashMap; @@ -19,6 +20,8 @@ struct Opt { #[tokio::main] async fn main() -> anyhow::Result<()> { + pretty_env_logger::init(); + let opt = Opt::from_args(); let config = Config::read_config(&opt.config).unwrap(); let config_bare = config.clone(); @@ -28,6 +31,9 @@ async fn main() -> anyhow::Result<()> { let index = Arc::new(Mutex::new(Index::default())); let index = warp::any().map(move || Arc::clone(&index)); + info!("Obnam server starting up"); + debug!("Configuration: {:?}", config_bare); + let create = warp::post() .and(warp::path("chunks")) .and(config.clone()) @@ -58,9 +64,9 @@ async fn main() -> anyhow::Result<()> { let webroot = create.or(fetch).or(search).or(delete); warp::serve(webroot) - .tls() - .key_path(config_bare.tls_key) - .cert_path(config_bare.tls_cert) + // .tls() + // .key_path(config_bare.tls_key) + // .cert_path(config_bare.tls_cert) .run(([127, 0, 0, 1], config_bare.port)) .await; Ok(()) @@ -148,6 +154,7 @@ pub async fn create_chunk( index.insert_generation(id.clone()); } + info!("created chunk {}: {:?}", id, meta); Ok(ChunkResult::Created(id)) } -- cgit v1.2.1