Compare commits

..

6 Commits

Author SHA1 Message Date
Alex Auvolat 5e4e870403 add boto3 test for STREAMING-UNSIGNED-PAYLOAD-TRAILER 2025-05-22 17:44:51 +02:00
Alex 38ca35eb0f Merge pull request 'refactor: make TableShardedReplication a thin wrapper around LayoutManager' (#820) from yuka/garage:refactor-sharded-table into next-v2
Reviewed-on: https://git.deuxfleurs.fr/Deuxfleurs/garage/pulls/820
2025-04-28 10:43:35 +00:00
Yureka a2d87a012d refactor: use replication factor of the layout versions in calculate_sync_map_min_with_quorum 2025-04-28 11:51:01 +02:00
Yureka 899292ee28 refactor: make TableShardedReplication a thin wrapper around LayoutManager 2025-04-28 11:51:01 +02:00
Yureka c8e9c45889 refactor: Use ReplicationFactor type in more places
- Remove the replication_factor.replication_factor() in favor of
  usize::from(replication_factor) to make the conversion more explicit.

- Implement Display on ReplicationFactor so that it can be formatted
  without converting to usize

- Use ReplicationFactor in the constructor of LayoutVersion and add a
  method to get a ReplicationFactor from a LayoutVersion, despite
  LayoutVersion still storing it as usize internally.
2025-04-21 19:47:14 +02:00
Alex Auvolat e79b485aa8 fix panic in ListAdminTokens 2025-04-17 17:38:20 +02:00
13 changed files with 105 additions and 81 deletions
Generated
+4 -4
View File
@@ -50,17 +50,17 @@
}, },
"nixpkgs": { "nixpkgs": {
"locked": { "locked": {
"lastModified": 1736692550, "lastModified": 1747825515,
"narHash": "sha256-7tk8xH+g0sJkKLTJFOxphJxxOjMDFMWv24nXslaU2ro=", "narHash": "sha256-BWpMQymVI73QoKZdcVCxUCCK3GNvr/xa2Dc4DM1o2BE=",
"owner": "NixOS", "owner": "NixOS",
"repo": "nixpkgs", "repo": "nixpkgs",
"rev": "7c4869c47090dd7f9f1bdfb49a22aea026996815", "rev": "cd2812de55cf87df88a9e09bf3be1ce63d50c1a6",
"type": "github" "type": "github"
}, },
"original": { "original": {
"owner": "NixOS", "owner": "NixOS",
"repo": "nixpkgs", "repo": "nixpkgs",
"rev": "7c4869c47090dd7f9f1bdfb49a22aea026996815", "rev": "cd2812de55cf87df88a9e09bf3be1ce63d50c1a6",
"type": "github" "type": "github"
} }
}, },
+2 -2
View File
@@ -2,9 +2,9 @@
description = description =
"Garage, an S3-compatible distributed object store for self-hosted deployments"; "Garage, an S3-compatible distributed object store for self-hosted deployments";
# Nixpkgs 24.11 as of 2025-01-12 # Nixpkgs 25.05 as of 2025-05-22
inputs.nixpkgs.url = inputs.nixpkgs.url =
"github:NixOS/nixpkgs/7c4869c47090dd7f9f1bdfb49a22aea026996815"; "github:NixOS/nixpkgs/cd2812de55cf87df88a9e09bf3be1ce63d50c1a6";
# Rust overlay as of 2025-02-03 # Rust overlay as of 2025-02-03
inputs.rust-overlay.url = inputs.rust-overlay.url =
+13
View File
@@ -112,6 +112,19 @@ if [ -z "$SKIP_S3CMD" ]; then
done done
fi fi
# BOTO3
if [ -z "$SKIP_BOTO3" ]; then
echo "🛠️ Testing with boto3 for STREAMING-UNSIGNED-PAYLOAD-TRAILER"
source ${SCRIPT_FOLDER}/dev-env-aws.sh
AWS_ENDPOINT_URL=https://localhost:4443 python <<EOF
import boto3
client = boto3.client('s3', verify=False)
client.put_object(Body=b'hello world', Bucket='eprouvette', Key='test.s3.txt')
client.delete_object(Bucket='eprouvette', Key='test.s3.txt')
print("OK!")
EOF
fi
# Minio Client # Minio Client
if [ -z "$SKIP_MC" ]; then if [ -z "$SKIP_MC" ]; then
echo "🛠️ Testing with mc (minio client)" echo "🛠️ Testing with mc (minio client)"
+2
View File
@@ -26,6 +26,8 @@ in
s3cmd s3cmd
minio-client minio-client
rclone rclone
(python312.withPackages (ps: [ ps.boto3 ]))
socat socat
psmisc psmisc
which which
+14 -14
View File
@@ -36,6 +36,20 @@ impl RequestHandler for ListAdminTokensRequest {
.map(|t| admin_token_info_results(t, now)) .map(|t| admin_token_info_results(t, now))
.collect::<Vec<_>>(); .collect::<Vec<_>>();
if garage.config.admin.metrics_token.is_some() {
res.insert(
0,
GetAdminTokenInfoResponse {
id: None,
created: None,
name: "metrics_token (from daemon configuration)".into(),
expiration: None,
expired: false,
scope: vec!["Metrics".into()],
},
);
}
if garage.config.admin.admin_token.is_some() { if garage.config.admin.admin_token.is_some() {
res.insert( res.insert(
0, 0,
@@ -50,20 +64,6 @@ impl RequestHandler for ListAdminTokensRequest {
); );
} }
if garage.config.admin.metrics_token.is_some() {
res.insert(
1,
GetAdminTokenInfoResponse {
id: None,
created: None,
name: "metrics_token (from daemon configuration)".into(),
expiration: None,
expired: false,
scope: vec!["Metrics".into()],
},
);
}
Ok(ListAdminTokensResponse(res)) Ok(ListAdminTokensResponse(res))
} }
} }
+2 -4
View File
@@ -155,10 +155,8 @@ impl Garage {
let system = System::new(network_key, replication_factor, consistency_mode, &config)?; let system = System::new(network_key, replication_factor, consistency_mode, &config)?;
let meta_rep_param = TableShardedReplication { let meta_rep_param = TableShardedReplication {
system: system.clone(), layout_manager: system.layout_manager.clone(),
replication_factor: replication_factor.into(), consistency_mode,
write_quorum: replication_factor.write_quorum(consistency_mode),
read_quorum: replication_factor.read_quorum(consistency_mode),
}; };
let control_rep_param = TableFullReplication { let control_rep_param = TableFullReplication {
+1 -6
View File
@@ -28,7 +28,6 @@ pub struct SyncLayoutDigest {
} }
pub struct LayoutHelper { pub struct LayoutHelper {
replication_factor: ReplicationFactor,
consistency_mode: ConsistencyMode, consistency_mode: ConsistencyMode,
layout: Option<LayoutHistory>, layout: Option<LayoutHistory>,
@@ -51,7 +50,6 @@ pub struct LayoutHelper {
impl LayoutHelper { impl LayoutHelper {
pub fn new( pub fn new(
replication_factor: ReplicationFactor,
consistency_mode: ConsistencyMode, consistency_mode: ConsistencyMode,
mut layout: LayoutHistory, mut layout: LayoutHistory,
mut ack_lock: HashMap<u64, AtomicUsize>, mut ack_lock: HashMap<u64, AtomicUsize>,
@@ -97,8 +95,7 @@ impl LayoutHelper {
// consistency on those). // consistency on those).
// This value is calculated using quorums to allow progress even // This value is calculated using quorums to allow progress even
// if not all nodes have successfully completed a sync. // if not all nodes have successfully completed a sync.
let sync_map_min = let sync_map_min = layout.calculate_sync_map_min_with_quorum(&all_nongateway_nodes);
layout.calculate_sync_map_min_with_quorum(replication_factor, &all_nongateway_nodes);
let trackers_hash = layout.calculate_trackers_hash(); let trackers_hash = layout.calculate_trackers_hash();
let staging_hash = layout.calculate_staging_hash(); let staging_hash = layout.calculate_staging_hash();
@@ -111,7 +108,6 @@ impl LayoutHelper {
let is_check_ok = layout.check().is_ok(); let is_check_ok = layout.check().is_ok();
LayoutHelper { LayoutHelper {
replication_factor,
consistency_mode, consistency_mode,
layout: Some(layout), layout: Some(layout),
ack_map_min, ack_map_min,
@@ -134,7 +130,6 @@ impl LayoutHelper {
let changed = f(self.layout.as_mut().unwrap()); let changed = f(self.layout.as_mut().unwrap());
if changed { if changed {
*self = Self::new( *self = Self::new(
self.replication_factor,
self.consistency_mode, self.consistency_mode,
self.layout.take().unwrap(), self.layout.take().unwrap(),
std::mem::take(&mut self.ack_lock), std::mem::take(&mut self.ack_lock),
+15 -11
View File
@@ -123,13 +123,9 @@ impl LayoutHistory {
} }
} }
pub(crate) fn calculate_sync_map_min_with_quorum( /// This function calculates the minimum layout version from which
&self, /// it is safe to read if we want to maintain read-after-write consistency.
replication_factor: ReplicationFactor, pub(crate) fn calculate_sync_map_min_with_quorum(&self, all_nongateway_nodes: &[Uuid]) -> u64 {
all_nongateway_nodes: &[Uuid],
) -> u64 {
// This function calculates the minimum layout version from which
// it is safe to read if we want to maintain read-after-write consistency.
// In the general case the computation can be a bit expensive so // In the general case the computation can be a bit expensive so
// we try to optimize it in several ways. // we try to optimize it in several ways.
@@ -139,8 +135,6 @@ impl LayoutHistory {
return self.current().version; return self.current().version;
} }
let quorum = replication_factor.write_quorum(ConsistencyMode::Consistent);
let min_version = self.min_stored(); let min_version = self.min_stored();
let global_min = self let global_min = self
.update_trackers .update_trackers
@@ -153,7 +147,16 @@ impl LayoutHistory {
// This is represented by reading from the layout with version // This is represented by reading from the layout with version
// number global_min, the smallest layout version for which all nodes // number global_min, the smallest layout version for which all nodes
// have completed a sync. // have completed a sync.
if quorum == self.current().replication_factor { //
// While we currently do not support changing the replication factor
// between layout versions, this calculation is future-proofing for the
// case where this might be possible.
if self
.versions
.iter()
.filter(|v| v.version >= global_min)
.all(|v| v.write_quorum(ConsistencyMode::Consistent) == v.replication_factor)
{
return global_min; return global_min;
} }
@@ -195,7 +198,8 @@ impl LayoutHistory {
.map(|x| self.update_trackers.sync_map.get(x, min_version)) .map(|x| self.update_trackers.sync_map.get(x, min_version))
.collect::<Vec<_>>(); .collect::<Vec<_>>();
sync_values.sort(); sync_values.sort();
let set_min = sync_values[sync_values.len() - quorum]; let set_min =
sync_values[sync_values.len() - v.write_quorum(ConsistencyMode::Consistent)];
if set_min < current_min { if set_min < current_min {
current_min = set_min; current_min = set_min;
} }
+8 -12
View File
@@ -46,11 +46,11 @@ impl LayoutManager {
let cluster_layout = match persist_cluster_layout.load() { let cluster_layout = match persist_cluster_layout.load() {
Ok(x) => { Ok(x) => {
if x.current().replication_factor != replication_factor.replication_factor() { if x.current().replication_factor() != replication_factor {
return Err(Error::Message(format!( return Err(Error::Message(format!(
"Previous cluster layout has replication factor {}, which is different than the one specified in the config file ({}). The previous cluster layout can be purged, if you know what you are doing, simply by deleting the `cluster_layout` file in your metadata directory.", "Previous cluster layout has replication factor {}, which is different than the one specified in the config file ({}). The previous cluster layout can be purged, if you know what you are doing, simply by deleting the `cluster_layout` file in your metadata directory.",
x.current().replication_factor, x.current().replication_factor(),
replication_factor.replication_factor() replication_factor,
))); )));
} }
x x
@@ -64,12 +64,8 @@ impl LayoutManager {
} }
}; };
let mut cluster_layout = LayoutHelper::new( let mut cluster_layout =
replication_factor, LayoutHelper::new(consistency_mode, cluster_layout, Default::default());
consistency_mode,
cluster_layout,
Default::default(),
);
cluster_layout.update_update_trackers(node_id.into()); cluster_layout.update_update_trackers(node_id.into());
let layout = Arc::new(RwLock::new(cluster_layout)); let layout = Arc::new(RwLock::new(cluster_layout));
@@ -301,11 +297,11 @@ impl LayoutManager {
adv.update_trackers adv.update_trackers
); );
if adv.current().replication_factor != self.replication_factor.replication_factor() { if adv.current().replication_factor() != self.replication_factor {
let msg = format!( let msg = format!(
"Received a cluster layout from another node with replication factor {}, which is different from what we have in our configuration ({}). Discarding the cluster layout we received.", "Received a cluster layout from another node with replication factor {}, which is different from what we have in our configuration ({}). Discarding the cluster layout we received.",
adv.current().replication_factor, adv.current().replication_factor(),
self.replication_factor.replication_factor() self.replication_factor,
); );
error!("{}", msg); error!("{}", msg);
return Err(Error::Message(msg)); return Err(Error::Message(msg));
+15 -2
View File
@@ -11,12 +11,13 @@ use garage_util::error::*;
use super::graph_algo::*; use super::graph_algo::*;
use super::*; use super::*;
use crate::replication_mode::*;
// The Message type will be used to collect information on the algorithm. // The Message type will be used to collect information on the algorithm.
pub type Message = Vec<String>; pub type Message = Vec<String>;
impl LayoutVersion { impl LayoutVersion {
pub fn new(replication_factor: usize) -> Self { pub fn new(replication_factor: ReplicationFactor) -> Self {
// We set the default zone redundancy to be Maximum, meaning that the maximum // We set the default zone redundancy to be Maximum, meaning that the maximum
// possible value will be used depending on the cluster topology // possible value will be used depending on the cluster topology
let parameters = LayoutParameters { let parameters = LayoutParameters {
@@ -25,7 +26,7 @@ impl LayoutVersion {
LayoutVersion { LayoutVersion {
version: 0, version: 0,
replication_factor, replication_factor: usize::from(replication_factor),
partition_size: 0, partition_size: 0,
roles: LwwMap::new(), roles: LwwMap::new(),
node_id_vec: Vec::new(), node_id_vec: Vec::new(),
@@ -132,6 +133,18 @@ impl LayoutVersion {
.map(move |i| self.node_id_vec[*i as usize]) .map(move |i| self.node_id_vec[*i as usize])
} }
pub fn replication_factor(&self) -> ReplicationFactor {
ReplicationFactor::new(self.replication_factor).unwrap()
}
pub fn read_quorum(&self, consistency_mode: ConsistencyMode) -> usize {
self.replication_factor().read_quorum(consistency_mode)
}
pub fn write_quorum(&self, consistency_mode: ConsistencyMode) -> usize {
self.replication_factor().write_quorum(consistency_mode)
}
// ===================== internal information extractors ====================== // ===================== internal information extractors ======================
pub(crate) fn expect_get_node_capacity(&self, uuid: &Uuid) -> u64 { pub(crate) fn expect_get_node_capacity(&self, uuid: &Uuid) -> u64 {
+8 -6
View File
@@ -38,14 +38,10 @@ impl ReplicationFactor {
} }
} }
pub fn replication_factor(&self) -> usize {
self.0
}
pub fn read_quorum(&self, consistency_mode: ConsistencyMode) -> usize { pub fn read_quorum(&self, consistency_mode: ConsistencyMode) -> usize {
match consistency_mode { match consistency_mode {
ConsistencyMode::Dangerous | ConsistencyMode::Degraded => 1, ConsistencyMode::Dangerous | ConsistencyMode::Degraded => 1,
ConsistencyMode::Consistent => self.replication_factor().div_ceil(2), ConsistencyMode::Consistent => usize::from(*self).div_ceil(2),
} }
} }
@@ -53,7 +49,7 @@ impl ReplicationFactor {
match consistency_mode { match consistency_mode {
ConsistencyMode::Dangerous => 1, ConsistencyMode::Dangerous => 1,
ConsistencyMode::Degraded | ConsistencyMode::Consistent => { ConsistencyMode::Degraded | ConsistencyMode::Consistent => {
(self.replication_factor() + 1) - self.read_quorum(ConsistencyMode::Consistent) (usize::from(*self) + 1) - self.read_quorum(ConsistencyMode::Consistent)
} }
} }
} }
@@ -65,6 +61,12 @@ impl std::convert::From<ReplicationFactor> for usize {
} }
} }
impl std::fmt::Display for ReplicationFactor {
fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
self.0.fmt(f)
}
}
pub fn parse_replication_mode( pub fn parse_replication_mode(
config: &Config, config: &Config,
) -> Result<(ReplicationFactor, ConsistencyMode), Error> { ) -> Result<(ReplicationFactor, ConsistencyMode), Error> {
+1 -1
View File
@@ -68,7 +68,7 @@ impl SystemMetrics {
let replication_factor = system.replication_factor; let replication_factor = system.replication_factor;
meter meter
.u64_value_observer("garage_replication_factor", move |observer| { .u64_value_observer("garage_replication_factor", move |observer| {
observer.observe(replication_factor.replication_factor() as u64, &[]) observer.observe(usize::from(replication_factor) as u64, &[])
}) })
.with_description("Garage replication factor setting") .with_description("Garage replication factor setting")
.init() .init()
+20 -19
View File
@@ -2,9 +2,10 @@ use std::sync::Arc;
use std::time::Duration; use std::time::Duration;
use garage_rpc::layout::*; use garage_rpc::layout::*;
use garage_rpc::system::System; use garage_rpc::replication_mode::ConsistencyMode;
use garage_util::data::*; use garage_util::data::*;
use crate::replication::sharded::manager::LayoutManager;
use crate::replication::*; use crate::replication::*;
/// Sharded replication schema: /// Sharded replication schema:
@@ -16,13 +17,8 @@ use crate::replication::*;
#[derive(Clone)] #[derive(Clone)]
pub struct TableShardedReplication { pub struct TableShardedReplication {
/// The membership manager of this node /// The membership manager of this node
pub system: Arc<System>, pub layout_manager: Arc<LayoutManager>,
/// How many time each data should be replicated pub consistency_mode: ConsistencyMode,
pub replication_factor: usize,
/// How many nodes to contact for a read, should be at most `replication_factor`
pub read_quorum: usize,
/// How many nodes to contact for a write, should be at most `replication_factor`
pub write_quorum: usize,
} }
impl TableReplication for TableShardedReplication { impl TableReplication for TableShardedReplication {
@@ -32,9 +28,8 @@ impl TableReplication for TableShardedReplication {
type WriteSets = WriteLock<Vec<Vec<Uuid>>>; type WriteSets = WriteLock<Vec<Vec<Uuid>>>;
fn storage_nodes(&self, hash: &Hash) -> Vec<Uuid> { fn storage_nodes(&self, hash: &Hash) -> Vec<Uuid> {
let layout = self.system.cluster_layout();
let mut ret = vec![]; let mut ret = vec![];
for version in layout.versions().iter() { for version in self.layout_manager.layout().versions().iter() {
ret.extend(version.nodes_of(hash)); ret.extend(version.nodes_of(hash));
} }
ret.sort(); ret.sort();
@@ -43,31 +38,37 @@ impl TableReplication for TableShardedReplication {
} }
fn read_nodes(&self, hash: &Hash) -> Vec<Uuid> { fn read_nodes(&self, hash: &Hash) -> Vec<Uuid> {
self.system self.layout_manager
.cluster_layout() .layout()
.read_version() .read_version()
.nodes_of(hash) .nodes_of(hash)
.collect() .collect()
} }
fn read_quorum(&self) -> usize { fn read_quorum(&self) -> usize {
self.read_quorum self.layout_manager
.layout()
.read_version()
.read_quorum(self.consistency_mode)
} }
fn write_sets(&self, hash: &Hash) -> Self::WriteSets { fn write_sets(&self, hash: &Hash) -> Self::WriteSets {
self.system self.layout_manager.write_lock_with(|l| write_sets(l, hash))
.layout_manager
.write_lock_with(|l| write_sets(l, hash))
} }
fn write_quorum(&self) -> usize { fn write_quorum(&self) -> usize {
self.write_quorum self.layout_manager
.layout()
.current()
.write_quorum(self.consistency_mode)
} }
fn partition_of(&self, hash: &Hash) -> Partition { fn partition_of(&self, hash: &Hash) -> Partition {
self.system.cluster_layout().current().partition_of(hash) self.layout_manager.layout().current().partition_of(hash)
} }
fn sync_partitions(&self) -> SyncPartitions { fn sync_partitions(&self) -> SyncPartitions {
let layout = self.system.cluster_layout(); let layout = self.layout_manager.layout();
let layout_version = layout.ack_map_min(); let layout_version = layout.ack_map_min();
let mut partitions = layout let mut partitions = layout