aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--src/main.rs71
-rw-r--r--src/server.rs9
-rw-r--r--src/table_sync.rs75
3 files changed, 142 insertions, 13 deletions
diff --git a/src/main.rs b/src/main.rs
index 8b124bff..ebf97a29 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -61,6 +61,10 @@ pub enum Command {
/// Configure Garage node
#[structopt(name = "configure")]
Configure(ConfigureOpt),
+
+ /// Remove Garage node from cluster
+ #[structopt(name = "remove")]
+ Remove(RemoveOpt),
}
#[derive(StructOpt, Debug)]
@@ -82,6 +86,16 @@ pub struct ConfigureOpt {
n_tokens: u32,
}
+#[derive(StructOpt, Debug)]
+pub struct RemoveOpt {
+ /// Node to configure (prefix of hexadecimal node id)
+ node_id: String,
+
+ /// If this flag is not given, the node won't be removed
+ #[structopt(long = "yes")]
+ yes: bool,
+}
+
#[tokio::main]
async fn main() {
let opt = Opt::from_args();
@@ -102,11 +116,20 @@ async fn main() {
let rpc_cli = RpcClient::new(&tls_config).expect("Could not create RPC client");
let resp = match opt.cmd {
- Command::Server(server_opt) => server::run_server(server_opt.config_file).await,
+ Command::Server(server_opt) => {
+ // Abort on panic (same behavior as in Go)
+ std::panic::set_hook(Box::new(|panic_info| {
+ eprintln!("{}", panic_info.to_string());
+ std::process::abort();
+ }));
+
+ server::run_server(server_opt.config_file).await
+ }
Command::Status => cmd_status(rpc_cli, opt.rpc_host).await,
Command::Configure(configure_opt) => {
cmd_configure(rpc_cli, opt.rpc_host, configure_opt).await
}
+ Command::Remove(remove_opt) => cmd_remove(rpc_cli, opt.rpc_host, remove_opt).await,
};
if let Err(e) = resp {
@@ -224,3 +247,49 @@ async fn cmd_configure(
.await?;
Ok(())
}
+
+async fn cmd_remove(
+ rpc_cli: RpcClient,
+ rpc_host: SocketAddr,
+ args: RemoveOpt,
+) -> Result<(), Error> {
+ let mut config = match rpc_cli
+ .call(&rpc_host, &Message::PullConfig, DEFAULT_TIMEOUT)
+ .await?
+ {
+ Message::AdvertiseConfig(cfg) => cfg,
+ resp => return Err(Error::Message(format!("Invalid RPC response: {:?}", resp))),
+ };
+
+ let mut candidates = vec![];
+ for (key, _) in config.members.iter() {
+ if hex::encode(key).starts_with(&args.node_id) {
+ candidates.push(key.clone());
+ }
+ }
+ if candidates.len() != 1 {
+ return Err(Error::Message(format!(
+ "{} matching nodes",
+ candidates.len()
+ )));
+ }
+
+ if !args.yes {
+ return Err(Error::Message(format!(
+ "Add the flag --yes to really remove {:?} from the cluster",
+ candidates[0]
+ )));
+ }
+
+ config.members.remove(&candidates[0]);
+ config.version += 1;
+
+ rpc_cli
+ .call(
+ &rpc_host,
+ &Message::AdvertiseConfig(config),
+ DEFAULT_TIMEOUT,
+ )
+ .await?;
+ Ok(())
+}
diff --git a/src/server.rs b/src/server.rs
index af58ded1..78b992f5 100644
--- a/src/server.rs
+++ b/src/server.rs
@@ -93,7 +93,8 @@ impl Garage {
&db,
"block_ref".to_string(),
data_rep_param.clone(),
- ).await;
+ )
+ .await;
let version_table = Table::new(
VersionTable {
background: background.clone(),
@@ -103,7 +104,8 @@ impl Garage {
&db,
"version".to_string(),
meta_rep_param.clone(),
- ).await;
+ )
+ .await;
let object_table = Table::new(
ObjectTable {
background: background.clone(),
@@ -113,7 +115,8 @@ impl Garage {
&db,
"object".to_string(),
meta_rep_param.clone(),
- ).await;
+ )
+ .await;
let mut garage = Self {
db,
diff --git a/src/table_sync.rs b/src/table_sync.rs
index 5097c1b0..039dab6d 100644
--- a/src/table_sync.rs
+++ b/src/table_sync.rs
@@ -1,7 +1,7 @@
use rand::Rng;
+use std::collections::BTreeSet;
use std::sync::Arc;
use std::time::Duration;
-use std::collections::BTreeSet;
use futures::{pin_mut, select};
use futures_util::future::*;
@@ -10,7 +10,7 @@ use tokio::sync::Mutex;
use crate::data::*;
use crate::error::Error;
-use crate::membership::{Ring, System};
+use crate::membership::Ring;
use crate::table::*;
const SCAN_INTERVAL: Duration = Duration::from_secs(3600);
@@ -29,6 +29,7 @@ pub struct SyncTodo {
pub struct Partition {
pub begin: Hash,
pub end: Hash,
+ pub retain: bool,
}
impl<F: TableSchema + 'static> TableSyncer<F> {
@@ -124,7 +125,8 @@ impl<F: TableSchema + 'static> TableSyncer<F> {
}
async fn sync_partition(self: &Arc<Self>, partition: &Partition) -> Result<(), Error> {
- unimplemented!()
+ eprintln!("NOT IMPLEMENTED: SYNC PARTITION {:?}", partition);
+ Ok(())
}
}
@@ -135,14 +137,17 @@ impl SyncTodo {
self.todo.clear();
let ring: Arc<Ring> = table.system.ring.borrow().clone();
+
for i in 0..ring.ring.len() {
let nodes = ring.walk_ring_from_pos(i, table.param.replication_factor);
let begin = ring.ring[i].location.clone();
+ if i == 0 {
+ self.add_full_scan_aux(table, [0u8; 32].into(), begin.clone(), &nodes[..], &my_id);
+ }
+
if i == ring.ring.len() - 1 {
- let end = ring.ring[0].location.clone();
self.add_full_scan_aux(table, begin, [0xffu8; 32].into(), &nodes[..], &my_id);
- self.add_full_scan_aux(table, [0u8; 32].into(), end, &nodes[..], &my_id);
} else {
let end = ring.ring[i + 1].location.clone();
self.add_full_scan_aux(table, begin, end, &nodes[..], &my_id);
@@ -158,23 +163,75 @@ impl SyncTodo {
nodes: &[UUID],
my_id: &UUID,
) {
- if !nodes.contains(my_id) {
+ let retain = nodes.contains(my_id);
+ if !retain {
// Check if we have some data to send, otherwise skip
if table
.store
.range(begin.clone()..end.clone())
.next()
.is_none()
- {}
+ {
+ return;
+ }
}
- self.todo.push(Partition { begin, end });
+ self.todo.push(Partition { begin, end, retain });
}
fn add_ring_difference<F: TableSchema>(&mut self, table: &Table<F>, old: &Ring, new: &Ring) {
+ let my_id = table.system.id.clone();
+
let old_ring = ring_points(old);
let new_ring = ring_points(new);
- unimplemented!()
+ let both_ring = old_ring.union(&new_ring).cloned().collect::<BTreeSet<_>>();
+
+ let prev_todo_begin = self
+ .todo
+ .iter()
+ .map(|x| x.begin.clone())
+ .collect::<BTreeSet<_>>();
+ let prev_todo_end = self
+ .todo
+ .iter()
+ .map(|x| x.end.clone())
+ .collect::<BTreeSet<_>>();
+ let prev_todo = prev_todo_begin
+ .union(&prev_todo_end)
+ .cloned()
+ .collect::<BTreeSet<_>>();
+
+ let all_points = both_ring.union(&prev_todo).cloned().collect::<Vec<_>>();
+
+ self.todo.sort_by(|x, y| x.begin.cmp(&y.begin));
+ let mut new_todo = vec![];
+ for i in 0..all_points.len() - 1 {
+ let begin = all_points[i].clone();
+ let end = all_points[i + 1].clone();
+ let was_ours = old
+ .walk_ring(&begin, table.param.replication_factor)
+ .contains(&my_id);
+ let is_ours = new
+ .walk_ring(&begin, table.param.replication_factor)
+ .contains(&my_id);
+ let was_todo = match self.todo.binary_search_by(|x| x.begin.cmp(&begin)) {
+ Ok(_) => true,
+ Err(j) => {
+ (j > 0 && self.todo[j - 1].begin < end && begin < self.todo[j - 1].end)
+ || (j < self.todo.len()
+ && self.todo[j].begin < end && begin < self.todo[j].end)
+ }
+ };
+ if was_todo || (is_ours && !was_ours) || (was_ours && !is_ours) {
+ new_todo.push(Partition {
+ begin,
+ end,
+ retain: is_ours,
+ });
+ }
+ }
+
+ self.todo = new_todo;
}
fn pop_task(&mut self) -> Option<Partition> {
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-05 23:17:55 +0000