diff options
Diffstat (limited to 'src/rpc')
-rw-r--r-- | src/rpc/Cargo.toml | 6 | ||||
-rw-r--r-- | src/rpc/layout.rs | 579 | ||||
-rw-r--r-- | src/rpc/lib.rs | 1 | ||||
-rw-r--r-- | src/rpc/ring.rs | 197 | ||||
-rw-r--r-- | src/rpc/rpc_helper.rs | 4 | ||||
-rw-r--r-- | src/rpc/system.rs | 100 |
6 files changed, 674 insertions, 213 deletions
diff --git a/src/rpc/Cargo.toml b/src/rpc/Cargo.toml index ac7c2a2e..d8ebb71e 100644 --- a/src/rpc/Cargo.toml +++ b/src/rpc/Cargo.toml @@ -1,11 +1,12 @@ [package] name = "garage_rpc" -version = "0.4.0" +version = "0.5.0" authors = ["Alex Auvolat <alex@adnab.me>"] edition = "2018" license = "AGPL-3.0" description = "Cluster membership management and RPC protocol for the Garage object store" repository = "https://git.deuxfleurs.fr/Deuxfleurs/garage" +readme = "../../README.md" [lib] path = "lib.rs" @@ -13,7 +14,7 @@ path = "lib.rs" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] -garage_util = { version = "0.4.0", path = "../util" } +garage_util = { version = "0.5.0", path = "../util" } arc-swap = "1.0" bytes = "1.0" @@ -26,6 +27,7 @@ sodiumoxide = { version = "0.2.5-0", package = "kuska-sodiumoxide" } async-trait = "0.1.7" rmp-serde = "0.15" serde = { version = "1.0", default-features = false, features = ["derive", "rc"] } +serde_bytes = "0.11" serde_json = "1.0" futures = "0.3" diff --git a/src/rpc/layout.rs b/src/rpc/layout.rs new file mode 100644 index 00000000..895dbf1c --- /dev/null +++ b/src/rpc/layout.rs @@ -0,0 +1,579 @@ +use std::cmp::Ordering; +use std::collections::{HashMap, HashSet}; + +use serde::{Deserialize, Serialize}; + +use garage_util::crdt::{AutoCrdt, Crdt, LwwMap}; +use garage_util::data::*; + +use crate::ring::*; + +/// The layout of the cluster, i.e. the list of roles +/// which are assigned to each cluster node +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct ClusterLayout { + pub version: u64, + + pub replication_factor: usize, + pub roles: LwwMap<Uuid, NodeRoleV>, + + /// node_id_vec: a vector of node IDs with a role assigned + /// in the system (this includes gateway nodes). + /// The order here is different than the vec stored by `roles`, because: + /// 1. non-gateway nodes are first so that they have lower numbers + /// 2. nodes that don't have a role are excluded (but they need to + /// stay in the CRDT as tombstones) + pub node_id_vec: Vec<Uuid>, + /// the assignation of data partitions to node, the values + /// are indices in node_id_vec + #[serde(with = "serde_bytes")] + pub ring_assignation_data: Vec<CompactNodeType>, + + /// Role changes which are staged for the next version of the layout + pub staging: LwwMap<Uuid, NodeRoleV>, + pub staging_hash: Hash, +} + +#[derive(PartialEq, Eq, PartialOrd, Ord, Clone, Debug, Serialize, Deserialize)] +pub struct NodeRoleV(pub Option<NodeRole>); + +impl AutoCrdt for NodeRoleV { + const WARN_IF_DIFFERENT: bool = true; +} + +/// The user-assigned roles of cluster nodes +#[derive(PartialEq, Eq, PartialOrd, Ord, Clone, Debug, Serialize, Deserialize)] +pub struct NodeRole { + /// Datacenter at which this entry belong. This information might be used to perform a better + /// geodistribution + pub zone: String, + /// The (relative) capacity of the node + /// If this is set to None, the node does not participate in storing data for the system + /// and is only active as an API gateway to other nodes + pub capacity: Option<u32>, + /// A set of tags to recognize the node + pub tags: Vec<String>, +} + +impl NodeRole { + pub fn capacity_string(&self) -> String { + match self.capacity { + Some(c) => format!("{}", c), + None => "gateway".to_string(), + } + } +} + +impl ClusterLayout { + pub fn new(replication_factor: usize) -> Self { + let empty_lwwmap = LwwMap::new(); + let empty_lwwmap_hash = blake2sum(&rmp_to_vec_all_named(&empty_lwwmap).unwrap()[..]); + + ClusterLayout { + version: 0, + replication_factor, + roles: LwwMap::new(), + node_id_vec: Vec::new(), + ring_assignation_data: Vec::new(), + staging: empty_lwwmap, + staging_hash: empty_lwwmap_hash, + } + } + + pub fn merge(&mut self, other: &ClusterLayout) -> bool { + match other.version.cmp(&self.version) { + Ordering::Greater => { + *self = other.clone(); + true + } + Ordering::Equal => { + self.staging.merge(&other.staging); + + let new_staging_hash = blake2sum(&rmp_to_vec_all_named(&self.staging).unwrap()[..]); + let changed = new_staging_hash != self.staging_hash; + + self.staging_hash = new_staging_hash; + + changed + } + Ordering::Less => false, + } + } + + /// Returns a list of IDs of nodes that currently have + /// a role in the cluster + pub fn node_ids(&self) -> &[Uuid] { + &self.node_id_vec[..] + } + + pub fn num_nodes(&self) -> usize { + self.node_id_vec.len() + } + + /// Returns the role of a node in the layout + pub fn node_role(&self, node: &Uuid) -> Option<&NodeRole> { + match self.roles.get(node) { + Some(NodeRoleV(Some(v))) => Some(v), + _ => None, + } + } + + /// Check a cluster layout for internal consistency + /// returns true if consistent, false if error + pub fn check(&self) -> bool { + // Check that the hash of the staging data is correct + let staging_hash = blake2sum(&rmp_to_vec_all_named(&self.staging).unwrap()[..]); + if staging_hash != self.staging_hash { + return false; + } + + // Check that node_id_vec contains the correct list of nodes + let mut expected_nodes = self + .roles + .items() + .iter() + .filter(|(_, _, v)| v.0.is_some()) + .map(|(id, _, _)| *id) + .collect::<Vec<_>>(); + expected_nodes.sort(); + let mut node_id_vec = self.node_id_vec.clone(); + node_id_vec.sort(); + if expected_nodes != node_id_vec { + return false; + } + + // Check that the assignation data has the correct length + if self.ring_assignation_data.len() != (1 << PARTITION_BITS) * self.replication_factor { + return false; + } + + // Check that the assigned nodes are correct identifiers + // of nodes that are assigned a role + // and that role is not the role of a gateway nodes + for x in self.ring_assignation_data.iter() { + if *x as usize >= self.node_id_vec.len() { + return false; + } + let node = self.node_id_vec[*x as usize]; + match self.roles.get(&node) { + Some(NodeRoleV(Some(x))) if x.capacity.is_some() => (), + _ => return false, + } + } + + true + } + + /// Calculate an assignation of partitions to nodes + pub fn calculate_partition_assignation(&mut self) -> bool { + let (configured_nodes, zones) = self.configured_nodes_and_zones(); + let n_zones = zones.len(); + + println!("Calculating updated partition assignation, this may take some time..."); + println!(); + + let old_partitions = self.parse_assignation_data(); + + let mut partitions = old_partitions.clone(); + for part in partitions.iter_mut() { + part.nodes + .retain(|(_, info)| info.map(|x| x.capacity.is_some()).unwrap_or(false)); + } + + // When nodes are removed, or when bootstraping an assignation from + // scratch for a new cluster, the old partitions will have holes (or be empty). + // Here we add more nodes to make a complete (sub-optimal) assignation, + // using an initial partition assignation that is calculated using the multi-dc maglev trick + match self.initial_partition_assignation() { + Some(initial_partitions) => { + for (part, ipart) in partitions.iter_mut().zip(initial_partitions.iter()) { + for (id, info) in ipart.nodes.iter() { + if part.nodes.len() < self.replication_factor { + part.add(part.nodes.len() + 1, n_zones, id, info.unwrap()); + } + } + assert!(part.nodes.len() == self.replication_factor); + } + } + None => { + return false; + } + } + + // Calculate how many partitions each node should ideally store, + // and how many partitions they are storing with the current assignation + // This defines our target for which we will optimize in the following loop. + let total_capacity = configured_nodes + .iter() + .map(|(_, info)| info.capacity.unwrap_or(0)) + .sum::<u32>() as usize; + let total_partitions = self.replication_factor * (1 << PARTITION_BITS); + let target_partitions_per_node = configured_nodes + .iter() + .map(|(id, info)| { + ( + *id, + info.capacity.unwrap_or(0) as usize * total_partitions / total_capacity, + ) + }) + .collect::<HashMap<&Uuid, usize>>(); + + let mut partitions_per_node = self.partitions_per_node(&partitions[..]); + + println!("Target number of partitions per node:"); + for (node, npart) in target_partitions_per_node.iter() { + println!("{:?}\t{}", node, npart); + } + println!(); + + // Shuffle partitions between nodes so that nodes will reach (or better approach) + // their target number of stored partitions + loop { + let mut option = None; + for (i, part) in partitions.iter_mut().enumerate() { + for (irm, (idrm, _)) in part.nodes.iter().enumerate() { + let suprm = partitions_per_node.get(*idrm).cloned().unwrap_or(0) as i32 + - target_partitions_per_node.get(*idrm).cloned().unwrap_or(0) as i32; + + for (idadd, infoadd) in configured_nodes.iter() { + // skip replacing a node by itself + // and skip replacing by gateway nodes + if idadd == idrm || infoadd.capacity.is_none() { + continue; + } + + let supadd = partitions_per_node.get(*idadd).cloned().unwrap_or(0) as i32 + - target_partitions_per_node.get(*idadd).cloned().unwrap_or(0) as i32; + + // We want to try replacing node idrm by node idadd + // if that brings us close to our goal. + let square = |i: i32| i * i; + let oldcost = square(suprm) + square(supadd); + let newcost = square(suprm - 1) + square(supadd + 1); + if newcost >= oldcost { + // not closer to our goal + continue; + } + let gain = oldcost - newcost; + + let mut newpart = part.clone(); + + newpart.nodes.remove(irm); + if !newpart.add(newpart.nodes.len() + 1, n_zones, idadd, infoadd) { + continue; + } + assert!(newpart.nodes.len() == self.replication_factor); + + if !old_partitions[i] + .is_valid_transition_to(&newpart, self.replication_factor) + { + continue; + } + + if option + .as_ref() + .map(|(old_gain, _, _, _, _)| gain > *old_gain) + .unwrap_or(true) + { + option = Some((gain, i, idadd, idrm, newpart)); + } + } + } + } + if let Some((_gain, i, idadd, idrm, newpart)) = option { + *partitions_per_node.entry(idadd).or_insert(0) += 1; + *partitions_per_node.get_mut(idrm).unwrap() -= 1; + partitions[i] = newpart; + } else { + break; + } + } + + // Check we completed the assignation correctly + // (this is a set of checks for the algorithm's consistency) + assert!(partitions.len() == (1 << PARTITION_BITS)); + assert!(partitions + .iter() + .all(|p| p.nodes.len() == self.replication_factor)); + + let new_partitions_per_node = self.partitions_per_node(&partitions[..]); + assert!(new_partitions_per_node == partitions_per_node); + + // Show statistics + println!("New number of partitions per node:"); + for (node, npart) in partitions_per_node.iter() { + println!("{:?}\t{}", node, npart); + } + println!(); + + let mut diffcount = HashMap::new(); + for (oldpart, newpart) in old_partitions.iter().zip(partitions.iter()) { + let nminus = oldpart.txtplus(newpart); + let nplus = newpart.txtplus(oldpart); + if nminus != "[...]" || nplus != "[...]" { + let tup = (nminus, nplus); + *diffcount.entry(tup).or_insert(0) += 1; + } + } + if diffcount.is_empty() { + println!("No data will be moved between nodes."); + } else { + let mut diffcount = diffcount.into_iter().collect::<Vec<_>>(); + diffcount.sort(); + println!("Number of partitions that move:"); + for ((nminus, nplus), npart) in diffcount { + println!("\t{}\t{} -> {}", npart, nminus, nplus); + } + } + println!(); + + // Calculate and save new assignation data + let (nodes, assignation_data) = + self.compute_assignation_data(&configured_nodes[..], &partitions[..]); + + self.node_id_vec = nodes; + self.ring_assignation_data = assignation_data; + + true + } + + fn initial_partition_assignation(&self) -> Option<Vec<PartitionAss<'_>>> { + let (configured_nodes, zones) = self.configured_nodes_and_zones(); + let n_zones = zones.len(); + + // Create a vector of partition indices (0 to 2**PARTITION_BITS-1) + let partitions_idx = (0usize..(1usize << PARTITION_BITS)).collect::<Vec<_>>(); + + // Prepare ring + let mut partitions: Vec<PartitionAss> = partitions_idx + .iter() + .map(|_i| PartitionAss::new()) + .collect::<Vec<_>>(); + + // Create MagLev priority queues for each node + let mut queues = configured_nodes + .iter() + .filter(|(_id, info)| info.capacity.is_some()) + .map(|(node_id, node_info)| { + let mut parts = partitions_idx + .iter() + .map(|i| { + let part_data = + [&u16::to_be_bytes(*i as u16)[..], node_id.as_slice()].concat(); + (*i, fasthash(&part_data[..])) + }) + .collect::<Vec<_>>(); + parts.sort_by_key(|(_i, h)| *h); + let parts_i = parts.iter().map(|(i, _h)| *i).collect::<Vec<_>>(); + (node_id, node_info, parts_i, 0) + }) + .collect::<Vec<_>>(); + + let max_capacity = configured_nodes + .iter() + .filter_map(|(_, node_info)| node_info.capacity) + .fold(0, std::cmp::max); + + // Fill up ring + for rep in 0..self.replication_factor { + queues.sort_by_key(|(ni, _np, _q, _p)| { + let queue_data = [&u16::to_be_bytes(rep as u16)[..], ni.as_slice()].concat(); + fasthash(&queue_data[..]) + }); + + for (_, _, _, pos) in queues.iter_mut() { + *pos = 0; + } + + let mut remaining = partitions_idx.len(); + while remaining > 0 { + let remaining0 = remaining; + for i_round in 0..max_capacity { + for (node_id, node_info, q, pos) in queues.iter_mut() { + if i_round >= node_info.capacity.unwrap() { + continue; + } + for (pos2, &qv) in q.iter().enumerate().skip(*pos) { + if partitions[qv].add(rep + 1, n_zones, node_id, node_info) { + remaining -= 1; + *pos = pos2 + 1; + break; + } + } + } + } + if remaining == remaining0 { + // No progress made, exit + return None; + } + } + } + + Some(partitions) + } + + fn configured_nodes_and_zones(&self) -> (Vec<(&Uuid, &NodeRole)>, HashSet<&str>) { + let configured_nodes = self + .roles + .items() + .iter() + .filter(|(_id, _, info)| info.0.is_some()) + .map(|(id, _, info)| (id, info.0.as_ref().unwrap())) + .collect::<Vec<(&Uuid, &NodeRole)>>(); + + let zones = configured_nodes + .iter() + .filter(|(_id, info)| info.capacity.is_some()) + .map(|(_id, info)| info.zone.as_str()) + .collect::<HashSet<&str>>(); + + (configured_nodes, zones) + } + + fn compute_assignation_data<'a>( + &self, + configured_nodes: &[(&'a Uuid, &'a NodeRole)], + partitions: &[PartitionAss<'a>], + ) -> (Vec<Uuid>, Vec<CompactNodeType>) { + assert!(partitions.len() == (1 << PARTITION_BITS)); + + // Make a canonical order for nodes + let mut nodes = configured_nodes + .iter() + .filter(|(_id, info)| info.capacity.is_some()) + .map(|(id, _)| **id) + .collect::<Vec<_>>(); + let nodes_rev = nodes + .iter() + .enumerate() + .map(|(i, id)| (*id, i as CompactNodeType)) + .collect::<HashMap<Uuid, CompactNodeType>>(); + + let mut assignation_data = vec![]; + for partition in partitions.iter() { + assert!(partition.nodes.len() == self.replication_factor); + for (id, _) in partition.nodes.iter() { + assignation_data.push(*nodes_rev.get(id).unwrap()); + } + } + + nodes.extend( + configured_nodes + .iter() + .filter(|(_id, info)| info.capacity.is_none()) + .map(|(id, _)| **id), + ); + + (nodes, assignation_data) + } + + fn parse_assignation_data(&self) -> Vec<PartitionAss<'_>> { + if self.ring_assignation_data.len() == self.replication_factor * (1 << PARTITION_BITS) { + // If the previous assignation data is correct, use that + let mut partitions = vec![]; + for i in 0..(1 << PARTITION_BITS) { + let mut part = PartitionAss::new(); + for node_i in self.ring_assignation_data + [i * self.replication_factor..(i + 1) * self.replication_factor] + .iter() + { + let node_id = &self.node_id_vec[*node_i as usize]; + + if let Some(NodeRoleV(Some(info))) = self.roles.get(node_id) { + part.nodes.push((node_id, Some(info))); + } else { + part.nodes.push((node_id, None)); + } + } + partitions.push(part); + } + partitions + } else { + // Otherwise start fresh + (0..(1 << PARTITION_BITS)) + .map(|_| PartitionAss::new()) + .collect() + } + } + + fn partitions_per_node<'a>(&self, partitions: &[PartitionAss<'a>]) -> HashMap<&'a Uuid, usize> { + let mut partitions_per_node = HashMap::<&Uuid, usize>::new(); + for p in partitions.iter() { + for (id, _) in p.nodes.iter() { + *partitions_per_node.entry(*id).or_insert(0) += 1; + } + } + partitions_per_node + } +} + +// ---- Internal structs for partition assignation in layout ---- + +#[derive(Clone)] +struct PartitionAss<'a> { + nodes: Vec<(&'a Uuid, Option<&'a NodeRole>)>, +} + +impl<'a> PartitionAss<'a> { + fn new() -> Self { + Self { nodes: Vec::new() } + } + + fn nplus(&self, other: &PartitionAss<'a>) -> usize { + self.nodes + .iter() + .filter(|x| !other.nodes.contains(x)) + .count() + } + + fn txtplus(&self, other: &PartitionAss<'a>) -> String { + let mut nodes = self + .nodes + .iter() + .filter(|x| !other.nodes.contains(x)) + .map(|x| format!("{:?}", x.0)) + .collect::<Vec<_>>(); + nodes.sort(); + if self.nodes.iter().any(|x| other.nodes.contains(x)) { + nodes.push("...".into()); + } + format!("[{}]", nodes.join(" ")) + } + + fn is_valid_transition_to(&self, other: &PartitionAss<'a>, replication_factor: usize) -> bool { + let min_keep_nodes_per_part = (replication_factor + 1) / 2; + let n_removed = self.nplus(other); + + if self.nodes.len() <= min_keep_nodes_per_part { + n_removed == 0 + } else { + n_removed <= self.nodes.len() - min_keep_nodes_per_part + } + } + + fn add( + &mut self, + target_len: usize, + n_zones: usize, + node: &'a Uuid, + role: &'a NodeRole, + ) -> bool { + if self.nodes.len() != target_len - 1 { + return false; + } + + let p_zns = self + .nodes + .iter() + .map(|(_id, info)| info.unwrap().zone.as_str()) + .collect::<HashSet<&str>>(); + if (p_zns.len() < n_zones && !p_zns.contains(&role.zone.as_str())) + || (p_zns.len() == n_zones && !self.nodes.iter().any(|(id, _)| *id == node)) + { + self.nodes.push((node, Some(role))); + true + } else { + false + } + } +} diff --git a/src/rpc/lib.rs b/src/rpc/lib.rs index ea3f1139..b72392ab 100644 --- a/src/rpc/lib.rs +++ b/src/rpc/lib.rs @@ -5,6 +5,7 @@ extern crate log; mod consul; +pub mod layout; pub mod ring; pub mod system; diff --git a/src/rpc/ring.rs b/src/rpc/ring.rs index 3cb0d233..73a126a2 100644 --- a/src/rpc/ring.rs +++ b/src/rpc/ring.rs @@ -1,12 +1,11 @@ //! Module containing types related to computing nodes which should receive a copy of data blocks //! and metadata -use std::collections::{HashMap, HashSet}; use std::convert::TryInto; -use serde::{Deserialize, Serialize}; - use garage_util::data::*; +use crate::layout::ClusterLayout; + /// A partition id, which is stored on 16 bits /// i.e. we have up to 2**16 partitions. /// (in practice we have exactly 2**PARTITION_BITS partitions) @@ -22,47 +21,6 @@ pub const PARTITION_BITS: usize = 8; const PARTITION_MASK_U16: u16 = ((1 << PARTITION_BITS) - 1) << (16 - PARTITION_BITS); -/// The user-defined configuration of the cluster's nodes -#[derive(Clone, Debug, Serialize, Deserialize)] -pub struct NetworkConfig { - /// Map of each node's id to it's configuration - pub members: HashMap<Uuid, NetworkConfigEntry>, - /// Version of this config - pub version: u64, -} - -impl NetworkConfig { - pub(crate) fn new() -> Self { - Self { - members: HashMap::new(), - version: 0, - } - } -} - -/// The overall configuration of one (possibly remote) node -#[derive(Clone, Debug, Serialize, Deserialize)] -pub struct NetworkConfigEntry { - /// Datacenter at which this entry belong. This infromation might be used to perform a better - /// geodistribution - pub zone: String, - /// The (relative) capacity of the node - /// If this is set to None, the node does not participate in storing data for the system - /// and is only active as an API gateway to other nodes - pub capacity: Option<u32>, - /// A tag to recognize the entry, not used for other things than display - pub tag: String, -} - -impl NetworkConfigEntry { - pub fn capacity_string(&self) -> String { - match self.capacity { - Some(c) => format!("{}", c), - None => "gateway".to_string(), - } - } -} - /// A ring distributing fairly objects to nodes #[derive(Clone)] pub struct Ring { @@ -70,7 +28,7 @@ pub struct Ring { pub replication_factor: usize, /// The network configuration used to generate this ring - pub config: NetworkConfig, + pub layout: ClusterLayout, // Internal order of nodes used to make a more compact representation of the ring nodes: Vec<Uuid>, @@ -81,7 +39,7 @@ pub struct Ring { // Type to store compactly the id of a node in the system // Change this to u16 the day we want to have more than 256 nodes in a cluster -type CompactNodeType = u8; +pub type CompactNodeType = u8; // The maximum number of times an object might get replicated // This must be at least 3 because Garage supports 3-way replication @@ -102,132 +60,26 @@ struct RingEntry { } impl Ring { - // TODO this function MUST be refactored, it's 100 lines long, with a 50 lines loop, going up to 6 - // levels of imbrication. It is basically impossible to test, maintain, or understand for an - // outsider. - pub(crate) fn new(config: NetworkConfig, replication_factor: usize) -> Self { - // Create a vector of partition indices (0 to 2**PARTITION_BITS-1) - let partitions_idx = (0usize..(1usize << PARTITION_BITS)).collect::<Vec<_>>(); - - let zones = config - .members - .iter() - .filter(|(_id, info)| info.capacity.is_some()) - .map(|(_id, info)| info.zone.as_str()) - .collect::<HashSet<&str>>(); - let n_zones = zones.len(); - - // Prepare ring - let mut partitions: Vec<Vec<(&Uuid, &NetworkConfigEntry)>> = partitions_idx - .iter() - .map(|_i| Vec::new()) - .collect::<Vec<_>>(); - - // Create MagLev priority queues for each node - let mut queues = config - .members - .iter() - .filter(|(_id, info)| info.capacity.is_some()) - .map(|(node_id, node_info)| { - let mut parts = partitions_idx - .iter() - .map(|i| { - let part_data = - [&u16::to_be_bytes(*i as u16)[..], node_id.as_slice()].concat(); - (*i, fasthash(&part_data[..])) - }) - .collect::<Vec<_>>(); - parts.sort_by_key(|(_i, h)| *h); - let parts_i = parts.iter().map(|(i, _h)| *i).collect::<Vec<_>>(); - (node_id, node_info, parts_i, 0) - }) - .collect::<Vec<_>>(); - - let max_capacity = config - .members - .iter() - .filter_map(|(_, node_info)| node_info.capacity) - .fold(0, std::cmp::max); - - assert!(replication_factor <= MAX_REPLICATION); - - // Fill up ring - for rep in 0..replication_factor { - queues.sort_by_key(|(ni, _np, _q, _p)| { - let queue_data = [&u16::to_be_bytes(rep as u16)[..], ni.as_slice()].concat(); - fasthash(&queue_data[..]) - }); - - for (_, _, _, pos) in queues.iter_mut() { - *pos = 0; - } - - let mut remaining = partitions_idx.len(); - while remaining > 0 { - let remaining0 = remaining; - for i_round in 0..max_capacity { - for (node_id, node_info, q, pos) in queues.iter_mut() { - if i_round >= node_info.capacity.unwrap() { - continue; - } - for (pos2, &qv) in q.iter().enumerate().skip(*pos) { - if partitions[qv].len() != rep { - continue; - } - let p_zns = partitions[qv] - .iter() - .map(|(_id, info)| info.zone.as_str()) - .collect::<HashSet<&str>>(); - if (p_zns.len() < n_zones && !p_zns.contains(&node_info.zone.as_str())) - || (p_zns.len() == n_zones - && !partitions[qv].iter().any(|(id, _i)| id == node_id)) - { - partitions[qv].push((node_id, node_info)); - remaining -= 1; - *pos = pos2 + 1; - break; - } - } - } - } - if remaining == remaining0 { - // No progress made, exit - warn!("Could not build ring, not enough nodes configured."); - return Self { - replication_factor, - config, - nodes: vec![], - ring: vec![], - }; - } - } + pub(crate) fn new(layout: ClusterLayout, replication_factor: usize) -> Self { + if replication_factor != layout.replication_factor { + warn!("Could not build ring: replication factor does not match between local configuration and network role assignation."); + return Self::empty(layout, replication_factor); } - // Make a canonical order for nodes - let nodes = config - .members - .iter() - .filter(|(_id, info)| info.capacity.is_some()) - .map(|(id, _)| *id) - .collect::<Vec<_>>(); - let nodes_rev = nodes - .iter() - .enumerate() - .map(|(i, id)| (*id, i as CompactNodeType)) - .collect::<HashMap<Uuid, CompactNodeType>>(); + if layout.ring_assignation_data.len() != replication_factor * (1 << PARTITION_BITS) { + warn!("Could not build ring: network role assignation data has invalid length"); + return Self::empty(layout, replication_factor); + } - let ring = partitions - .iter() - .enumerate() - .map(|(i, nodes)| { + let nodes = layout.node_id_vec.clone(); + let ring = (0..(1 << PARTITION_BITS)) + .map(|i| { let top = (i as u16) << (16 - PARTITION_BITS); - let nodes = nodes - .iter() - .map(|(id, _info)| *nodes_rev.get(id).unwrap()) - .collect::<Vec<CompactNodeType>>(); - assert!(nodes.len() == replication_factor); let mut nodes_buf = [0u8; MAX_REPLICATION]; - nodes_buf[..replication_factor].copy_from_slice(&nodes[..]); + nodes_buf[..replication_factor].copy_from_slice( + &layout.ring_assignation_data + [replication_factor * i..replication_factor * (i + 1)], + ); RingEntry { hash_prefix: top, nodes_buf, @@ -237,12 +89,21 @@ impl Ring { Self { replication_factor, - config, + layout, nodes, ring, } } + fn empty(layout: ClusterLayout, replication_factor: usize) -> Self { + Self { + replication_factor, + layout, + nodes: vec![], + ring: vec![], + } + } + /// Get the partition in which data would fall on pub fn partition_of(&self, position: &Hash) -> Partition { let top = u16::from_be_bytes(position.as_slice()[0..2].try_into().unwrap()); diff --git a/src/rpc/rpc_helper.rs b/src/rpc/rpc_helper.rs index df0e94f8..68bdfc4f 100644 --- a/src/rpc/rpc_helper.rs +++ b/src/rpc/rpc_helper.rs @@ -225,7 +225,7 @@ impl RpcHelper { // Retrieve some status variables that we will use to sort requests let peer_list = self.0.fullmesh.get_peer_list(); let ring: Arc<Ring> = self.0.ring.borrow().clone(); - let our_zone = match ring.config.members.get(&self.0.our_node_id) { + let our_zone = match ring.layout.node_role(&self.0.our_node_id) { Some(pc) => &pc.zone, None => "", }; @@ -238,7 +238,7 @@ impl RpcHelper { // and within a same zone we priorize nodes with the lowest latency. let mut requests = requests .map(|(to, fut)| { - let peer_zone = match ring.config.members.get(&to) { + let peer_zone = match ring.layout.node_role(&to) { Some(pc) => &pc.zone, None => "", }; diff --git a/src/rpc/system.rs b/src/rpc/system.rs index 3f5f7fb1..aa8947ea 100644 --- a/src/rpc/system.rs +++ b/src/rpc/system.rs @@ -23,12 +23,13 @@ use netapp::{NetApp, NetworkKey, NodeID, NodeKey}; use garage_util::background::BackgroundRunner; use garage_util::config::Config; -use garage_util::data::Uuid; +use garage_util::data::*; use garage_util::error::*; use garage_util::persister::Persister; use garage_util::time::*; use crate::consul::*; +use crate::layout::*; use crate::ring::*; use crate::rpc_helper::*; @@ -48,13 +49,13 @@ pub enum SystemRpc { Ok, /// Request to connect to a specific node (in <pubkey>@<host>:<port> format) Connect(String), - /// Ask other node its config. Answered with AdvertiseConfig - PullConfig, + /// Ask other node its cluster layout. Answered with AdvertiseClusterLayout + PullClusterLayout, /// Advertise Garage status. Answered with another AdvertiseStatus. /// Exchanged with every node on a regular basis. AdvertiseStatus(NodeStatus), - /// Advertisement of nodes config. Sent spontanously or in response to PullConfig - AdvertiseConfig(NetworkConfig), + /// Advertisement of cluster layout. Sent spontanously or in response to PullClusterLayout + AdvertiseClusterLayout(ClusterLayout), /// Get known nodes states GetKnownNodes, /// Return known nodes @@ -70,7 +71,7 @@ pub struct System { /// The id of this node pub id: Uuid, - persist_config: Persister<NetworkConfig>, + persist_cluster_layout: Persister<ClusterLayout>, persist_peer_list: Persister<Vec<(Uuid, SocketAddr)>>, local_status: ArcSwap<NodeStatus>, @@ -103,8 +104,10 @@ pub struct NodeStatus { pub hostname: String, /// Replication factor configured on the node pub replication_factor: usize, - /// Configuration version - pub config_version: u64, + /// Cluster layout version + pub cluster_layout_version: u64, + /// Hash of cluster layout staging data + pub cluster_layout_staging_hash: Hash, } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -187,17 +190,17 @@ impl System { gen_node_key(&config.metadata_dir).expect("Unable to read or generate node ID"); info!("Node public key: {}", hex::encode(&node_key.public_key())); - let persist_config = Persister::new(&config.metadata_dir, "network_config"); + let persist_cluster_layout = Persister::new(&config.metadata_dir, "cluster_layout"); let persist_peer_list = Persister::new(&config.metadata_dir, "peer_list"); - let net_config = match persist_config.load() { + let cluster_layout = match persist_cluster_layout.load() { Ok(x) => x, Err(e) => { info!( - "No valid previous network configuration stored ({}), starting fresh.", + "No valid previous cluster layout stored ({}), starting fresh.", e ); - NetworkConfig::new() + ClusterLayout::new(replication_factor) } }; @@ -206,10 +209,11 @@ impl System { .into_string() .unwrap_or_else(|_| "<invalid utf-8>".to_string()), replication_factor, - config_version: net_config.version, + cluster_layout_version: cluster_layout.version, + cluster_layout_staging_hash: cluster_layout.staging_hash, }; - let ring = Ring::new(net_config, replication_factor); + let ring = Ring::new(cluster_layout, replication_factor); let (update_ring, ring) = watch::channel(Arc::new(ring)); if let Some(addr) = config.rpc_public_addr { @@ -229,7 +233,7 @@ impl System { let sys = Arc::new(System { id: netapp.id.into(), - persist_config, + persist_cluster_layout, persist_peer_list, local_status: ArcSwap::new(Arc::new(local_status)), node_status: RwLock::new(HashMap::new()), @@ -292,12 +296,12 @@ impl System { } /// Save network configuration to disc - async fn save_network_config(self: Arc<Self>) -> Result<(), Error> { + async fn save_cluster_layout(self: Arc<Self>) -> Result<(), Error> { let ring: Arc<Ring> = self.ring.borrow().clone(); - self.persist_config - .save_async(&ring.config) + self.persist_cluster_layout + .save_async(&ring.layout) .await - .expect("Cannot save current cluster configuration"); + .expect("Cannot save current cluster layout"); Ok(()) } @@ -305,7 +309,8 @@ impl System { let mut new_si: NodeStatus = self.local_status.load().as_ref().clone(); let ring = self.ring.borrow(); - new_si.config_version = ring.config.version; + new_si.cluster_layout_version = ring.layout.version; + new_si.cluster_layout_staging_hash = ring.layout.staging_hash; self.local_status.swap(Arc::new(new_si)); } @@ -337,9 +342,9 @@ impl System { ))); } - fn handle_pull_config(&self) -> SystemRpc { + fn handle_pull_cluster_layout(&self) -> SystemRpc { let ring = self.ring.borrow().clone(); - SystemRpc::AdvertiseConfig(ring.config.clone()) + SystemRpc::AdvertiseClusterLayout(ring.layout.clone()) } fn handle_get_known_nodes(&self) -> SystemRpc { @@ -360,7 +365,8 @@ impl System { .unwrap_or(NodeStatus { hostname: "?".to_string(), replication_factor: 0, - config_version: 0, + cluster_layout_version: 0, + cluster_layout_staging_hash: Hash::from([0u8; 32]), }), }) .collect::<Vec<_>>(); @@ -381,10 +387,12 @@ impl System { std::process::exit(1); } - if info.config_version > local_info.config_version { + if info.cluster_layout_version > local_info.cluster_layout_version + || info.cluster_layout_staging_hash != local_info.cluster_layout_staging_hash + { let self2 = self.clone(); self.background.spawn_cancellable(async move { - self2.pull_config(from).await; + self2.pull_cluster_layout(from).await; Ok(()) }); } @@ -397,32 +405,39 @@ impl System { Ok(SystemRpc::Ok) } - async fn handle_advertise_config( + async fn handle_advertise_cluster_layout( self: Arc<Self>, - adv: &NetworkConfig, + adv: &ClusterLayout, ) -> Result<SystemRpc, Error> { let update_ring = self.update_ring.lock().await; - let ring: Arc<Ring> = self.ring.borrow().clone(); + let mut layout: ClusterLayout = self.ring.borrow().layout.clone(); + + let prev_layout_check = layout.check(); + if layout.merge(adv) { + if prev_layout_check && !layout.check() { + error!("New cluster layout is invalid, discarding."); + return Err(Error::Message( + "New cluster layout is invalid, discarding.".into(), + )); + } - if adv.version > ring.config.version { - let ring = Ring::new(adv.clone(), self.replication_factor); + let ring = Ring::new(layout.clone(), self.replication_factor); update_ring.send(Arc::new(ring))?; drop(update_ring); let self2 = self.clone(); - let adv = adv.clone(); self.background.spawn_cancellable(async move { self2 .rpc .broadcast( &self2.system_endpoint, - SystemRpc::AdvertiseConfig(adv), + SystemRpc::AdvertiseClusterLayout(layout), RequestStrategy::with_priority(PRIO_HIGH), ) .await; Ok(()) }); - self.background.spawn(self.clone().save_network_config()); + self.background.spawn(self.clone().save_cluster_layout()); } Ok(SystemRpc::Ok) @@ -456,14 +471,15 @@ impl System { }; while !*stop_signal.borrow() { - let not_configured = self.ring.borrow().config.members.is_empty(); + let not_configured = !self.ring.borrow().layout.check(); let no_peers = self.fullmesh.get_peer_list().len() < self.replication_factor; + let expected_n_nodes = self.ring.borrow().layout.num_nodes(); let bad_peers = self .fullmesh .get_peer_list() .iter() .filter(|p| p.is_up()) - .count() != self.ring.borrow().config.members.len(); + .count() != expected_n_nodes; if not_configured || no_peers || bad_peers { info!("Doing a bootstrap/discovery step (not_configured: {}, no_peers: {}, bad_peers: {})", not_configured, no_peers, bad_peers); @@ -533,18 +549,18 @@ impl System { self.persist_peer_list.save_async(&peer_list).await } - async fn pull_config(self: Arc<Self>, peer: Uuid) { + async fn pull_cluster_layout(self: Arc<Self>, peer: Uuid) { let resp = self .rpc .call( &self.system_endpoint, peer, - SystemRpc::PullConfig, + SystemRpc::PullClusterLayout, RequestStrategy::with_priority(PRIO_HIGH).with_timeout(PING_TIMEOUT), ) .await; - if let Ok(SystemRpc::AdvertiseConfig(config)) = resp { - let _: Result<_, _> = self.handle_advertise_config(&config).await; + if let Ok(SystemRpc::AdvertiseClusterLayout(layout)) = resp { + let _: Result<_, _> = self.handle_advertise_cluster_layout(&layout).await; } } } @@ -554,9 +570,11 @@ impl EndpointHandler<SystemRpc> for System { async fn handle(self: &Arc<Self>, msg: &SystemRpc, from: NodeID) -> Result<SystemRpc, Error> { match msg { SystemRpc::Connect(node) => self.handle_connect(node).await, - SystemRpc::PullConfig => Ok(self.handle_pull_config()), + SystemRpc::PullClusterLayout => Ok(self.handle_pull_cluster_layout()), SystemRpc::AdvertiseStatus(adv) => self.handle_advertise_status(from.into(), adv).await, - SystemRpc::AdvertiseConfig(adv) => self.clone().handle_advertise_config(adv).await, + SystemRpc::AdvertiseClusterLayout(adv) => { + self.clone().handle_advertise_cluster_layout(adv).await + } SystemRpc::GetKnownNodes => Ok(self.handle_get_known_nodes()), _ => Err(Error::BadRpc("Unexpected RPC message".to_string())), } |