filip/Stalwart
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

SQL Read replicas and Distributed blob storage 💎 (closes #441)

Browse Source
This commit is contained in:
mdecimus
2024-08-11 19:30:00 +02:00
parent 8d3931b76e
commit 6b92961c36
12 changed files with 398 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
crates/common/src/config/mod.rs
View File

@@ -39,7 +39,11 @@ pub(crate) const CONNECTION_VARS: &[u32; 7] = &[
];
impl Core {
pub async fn parse(config: &mut Config, stores: Stores, config_manager: ConfigManager) -> Self {
pub async fn parse(
config: &mut Config,
mut stores: Stores,
config_manager: ConfigManager,
) -> Self {
let mut data = config
.value_require("storage.data")
.map(|id| id.to_string())
@@ -52,6 +56,29 @@ impl Core {
}
})
.unwrap_or_default();
// SPDX-SnippetBegin
// SPDX-FileCopyrightText: 2020 Stalwart Labs Ltd <hello@stalw.art>
// SPDX-License-Identifier: LicenseRef-SEL
#[cfg(feature = "enterprise")]
let enterprise = crate::enterprise::Enterprise::parse(config, &data).await;
#[cfg(feature = "enterprise")]
if enterprise.is_none() {
if matches!(data, Store::SQLReadReplica(_)) {
config
.new_build_error("storage.data", "SQL read replicas is an Enterprise feature");
data = Store::None;
}
stores
.stores
.retain(|_, store| !matches!(store, Store::SQLReadReplica(_)));
stores
.blob_stores
.retain(|_, store| !matches!(store.backend, BlobBackend::Composite(_)));
}
// SPDX-SnippetEnd
let mut blob = config
.value_require("storage.blob")
.map(|id| id.to_string())
@@ -132,7 +159,7 @@ impl Core {
Self {
#[cfg(feature = "enterprise")]
enterprise: crate::enterprise::Enterprise::parse(config, &data).await,
enterprise,
sieve: Scripting::parse(config, &stores).await,
network: Network::parse(config),
smtp: SmtpConfig::parse(config).await,

2
crates/main/Cargo.toml
View File

@@ -44,4 +44,4 @@ rocks = ["store/rocks"]
elastic = ["store/elastic"]
s3 = ["store/s3"]
redis = ["store/redis"]
enterprise = ["jmap/enterprise", "common/enterprise"]
enterprise = ["jmap/enterprise", "common/enterprise", "store/enterprise"]

1
crates/store/Cargo.toml
View File

@@ -59,6 +59,7 @@ s3 = ["rust-s3"]
foundation = ["foundationdb", "futures"]
fdb-chunked-bm = []
redis = ["dep:redis", "deadpool"]
enterprise = []
test_mode = []

137
crates/store/src/backend/composite/distributed_blob.rs Normal file
View File

@@ -0,0 +1,137 @@
/*
* SPDX-FileCopyrightText: 2020 Stalwart Labs Ltd <hello@stalw.art>
*
* SPDX-License-Identifier: LicenseRef-SEL
*
* This file is subject to the Stalwart Enterprise License Agreement (SEL) and
* is NOT open source software.
*
*/
use std::ops::Range;
use utils::config::{utils::AsKey, Config};
use crate::{BlobBackend, Store, Stores};
pub struct CompositeBlob {
pub stores: Vec<BlobBackend>,
}
impl CompositeBlob {
pub fn open(config: &mut Config, prefix: impl AsKey, stores: &Stores) -> Option<Self> {
let prefix = prefix.as_key();
let store_ids = config
.values((&prefix, "stores"))
.map(|(_, v)| v.to_string())
.collect::<Vec<_>>();
let mut blob_stores = Vec::with_capacity(store_ids.len());
for store_id in store_ids {
if let Some(store) = stores.blob_stores.get(&store_id) {
blob_stores.push(store.backend.clone());
} else {
config.new_build_error(
(&prefix, "stores"),
format!("Blob store {store_id} not found"),
);
return None;
}
}
if !blob_stores.is_empty() {
Some(Self {
stores: blob_stores,
})
} else {
config.new_build_error((&prefix, "stores"), "No blob stores specified");
None
}
}
#[inline(always)]
fn get_store(&self, key: &[u8]) -> &BlobBackend {
&self.stores[key.first().copied().unwrap_or_default() as usize % self.stores.len()]
}
pub async fn get_blob(
&self,
key: &[u8],
read_range: Range<usize>,
) -> trc::Result<Option<Vec<u8>>> {
Box::pin(async move {
match self.get_store(key) {
BlobBackend::Store(store) => match store {
#[cfg(feature = "sqlite")]
Store::SQLite(store) => store.get_blob(key, read_range).await,
#[cfg(feature = "foundation")]
Store::FoundationDb(store) => store.get_blob(key, read_range).await,
#[cfg(feature = "postgres")]
Store::PostgreSQL(store) => store.get_blob(key, read_range).await,
#[cfg(feature = "mysql")]
Store::MySQL(store) => store.get_blob(key, read_range).await,
#[cfg(feature = "rocks")]
Store::RocksDb(store) => store.get_blob(key, read_range).await,
Store::SQLReadReplica(store) => store.get_blob(key, read_range).await,
Store::None => Err(trc::StoreEvent::NotConfigured.into()),
},
BlobBackend::Fs(store) => store.get_blob(key, read_range).await,
#[cfg(feature = "s3")]
BlobBackend::S3(store) => store.get_blob(key, read_range).await,
BlobBackend::Composite(_) => unimplemented!(),
}
})
.await
}
pub async fn put_blob(&self, key: &[u8], data: &[u8]) -> trc::Result<()> {
Box::pin(async move {
match self.get_store(key) {
BlobBackend::Store(store) => match store {
#[cfg(feature = "sqlite")]
Store::SQLite(store) => store.put_blob(key, data).await,
#[cfg(feature = "foundation")]
Store::FoundationDb(store) => store.put_blob(key, data).await,
#[cfg(feature = "postgres")]
Store::PostgreSQL(store) => store.put_blob(key, data).await,
#[cfg(feature = "mysql")]
Store::MySQL(store) => store.put_blob(key, data).await,
#[cfg(feature = "rocks")]
Store::RocksDb(store) => store.put_blob(key, data).await,
Store::SQLReadReplica(store) => store.put_blob(key, data).await,
Store::None => Err(trc::StoreEvent::NotConfigured.into()),
},
BlobBackend::Fs(store) => store.put_blob(key, data).await,
#[cfg(feature = "s3")]
BlobBackend::S3(store) => store.put_blob(key, data).await,
BlobBackend::Composite(_) => unimplemented!(),
}
})
.await
}
pub async fn delete_blob(&self, key: &[u8]) -> trc::Result<bool> {
Box::pin(async move {
match self.get_store(key) {
BlobBackend::Store(store) => match store {
#[cfg(feature = "sqlite")]
Store::SQLite(store) => store.delete_blob(key).await,
#[cfg(feature = "foundation")]
Store::FoundationDb(store) => store.delete_blob(key).await,
#[cfg(feature = "postgres")]
Store::PostgreSQL(store) => store.delete_blob(key).await,
#[cfg(feature = "mysql")]
Store::MySQL(store) => store.delete_blob(key).await,
#[cfg(feature = "rocks")]
Store::RocksDb(store) => store.delete_blob(key).await,
Store::SQLReadReplica(store) => store.delete_blob(key).await,
Store::None => Err(trc::StoreEvent::NotConfigured.into()),
},
BlobBackend::Fs(store) => store.delete_blob(key).await,
#[cfg(feature = "s3")]
BlobBackend::S3(store) => store.delete_blob(key).await,
BlobBackend::Composite(_) => unimplemented!(),
}
})
.await
}
}

12
crates/store/src/backend/composite/mod.rs Normal file
View File

@@ -0,0 +1,12 @@
/*
* SPDX-FileCopyrightText: 2020 Stalwart Labs Ltd <hello@stalw.art>
*
* SPDX-License-Identifier: LicenseRef-SEL
*
* This file is subject to the Stalwart Enterprise License Agreement (SEL) and
* is NOT open source software.
*
*/
pub mod distributed_blob;
pub mod read_replica;

129
crates/store/src/backend/composite/read_replica.rs Normal file
View File

@@ -0,0 +1,129 @@
/*
* SPDX-FileCopyrightText: 2020 Stalwart Labs Ltd <hello@stalw.art>
*
* SPDX-License-Identifier: LicenseRef-SEL
*
* This file is subject to the Stalwart Enterprise License Agreement (SEL) and
* is NOT open source software.
*
*/
use std::{
ops::Range,
sync::atomic::{AtomicUsize, Ordering},
};
use roaring::RoaringBitmap;
use utils::config::{utils::AsKey, Config};
use crate::{
write::{AssignedIds, Batch, BitmapClass, ValueClass},
BitmapKey, Deserialize, IterateParams, Key, Store, Stores, ValueKey,
};
pub struct SQLReadReplica {
primary: Store,
replicas: Vec<Store>,
last_used_replica: AtomicUsize,
}
impl SQLReadReplica {
pub fn open(config: &mut Config, prefix: impl AsKey, stores: &Stores) -> Option<Self> {
let prefix = prefix.as_key();
let primary_id = config.value_require((&prefix, "primary"))?.to_string();
let replica_ids = config
.values((&prefix, "replicas"))
.map(|(_, v)| v.to_string())
.collect::<Vec<_>>();
let primary = if let Some(store) = stores.stores.get(&primary_id) {
store.clone()
} else {
config.new_build_error(
(&prefix, "primary"),
format!("Primary store {primary_id} not found"),
);
return None;
};
let mut replicas = Vec::with_capacity(replica_ids.len());
for replica_id in replica_ids {
if let Some(store) = stores.stores.get(&replica_id) {
replicas.push(store.clone());
} else {
config.new_build_error(
(&prefix, "replicas"),
format!("Replica store {replica_id} not found"),
);
return None;
}
}
if !replicas.is_empty() {
Some(Self {
primary,
replicas,
last_used_replica: AtomicUsize::new(0),
})
} else {
config.new_build_error((&prefix, "replicas"), "No replica stores specified");
None
}
}
#[inline(always)]
fn replica(&self) -> &Store {
&self.replicas[self.last_used_replica.fetch_add(1, Ordering::Relaxed) % self.replicas.len()]
}
pub async fn get_blob(&self, key: &[u8], range: Range<usize>) -> trc::Result<Option<Vec<u8>>> {
Box::pin(self.replica().get_blob(key, range)).await
}
pub async fn put_blob(&self, key: &[u8], data: &[u8]) -> trc::Result<()> {
Box::pin(self.primary.put_blob(key, data)).await
}
pub async fn delete_blob(&self, key: &[u8]) -> trc::Result<bool> {
Box::pin(self.primary.delete_blob(key)).await
}
pub async fn get_value<U>(&self, key: impl Key) -> trc::Result<Option<U>>
where
U: Deserialize + 'static,
{
Box::pin(self.replica().get_value(key)).await
}
pub async fn get_bitmap(
&self,
key: BitmapKey<BitmapClass<u32>>,
) -> trc::Result<Option<RoaringBitmap>> {
Box::pin(self.replica().get_bitmap(key)).await
}
pub async fn iterate<T: Key>(
&self,
params: IterateParams<T>,
cb: impl for<'x> FnMut(&'x [u8], &'x [u8]) -> trc::Result<bool> + Sync + Send,
) -> trc::Result<()> {
Box::pin(self.replica().iterate(params, cb)).await
}
pub async fn get_counter(
&self,
key: impl Into<ValueKey<ValueClass<u32>>> + Sync + Send,
) -> trc::Result<i64> {
Box::pin(self.replica().get_counter(key)).await
}
pub async fn write(&self, batch: Batch) -> trc::Result<AssignedIds> {
Box::pin(self.primary.write(batch)).await
}
pub async fn delete_range(&self, from: impl Key, to: impl Key) -> trc::Result<()> {
Box::pin(self.primary.delete_range(from, to)).await
}
pub async fn purge_store(&self) -> trc::Result<()> {
Box::pin(self.primary.purge_store()).await
}
}

2
crates/store/src/backend/mod.rs
View File

@@ -4,6 +4,8 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
#[cfg(feature = "enterprise")]
pub mod composite;
#[cfg(feature = "elastic")]
pub mod elastic;
#[cfg(feature = "foundation")]

47
crates/store/src/config.rs
View File

@@ -53,12 +53,14 @@ impl Stores {
pub async fn parse_stores(&mut self, config: &mut Config) {
let is_reload = !self.stores.is_empty();
for id in config
#[cfg(feature = "enterprise")]
let mut composite_stores = Vec::new();
let store_ids = config
.sub_keys("store", ".type")
.map(|id| id.to_string())
.collect::<Vec<_>>()
{
.collect::<Vec<_>>();
for id in store_ids {
let id = id.as_str();
// Parse store
#[cfg(feature = "test_mode")]
@@ -203,6 +205,10 @@ impl Stores {
self.lookup_stores.insert(store_id, db);
}
}
#[cfg(feature = "enterprise")]
"composite-read" | "composite-blob" => {
composite_stores.push((store_id, protocol));
}
unknown => {
config.new_parse_warning(
("store", id, "type"),
@@ -211,6 +217,39 @@ impl Stores {
}
}
}
#[cfg(feature = "enterprise")]
for (id, protocol) in composite_stores {
let prefix = ("store", id.as_str());
match protocol.as_str() {
"composite-read" => {
if let Some(db) = crate::backend::composite::read_replica::SQLReadReplica::open(
config, prefix, self,
) {
self.stores.insert(id, Store::SQLReadReplica(db.into()));
}
}
"composite-blob" => {
if let Some(db) =
crate::backend::composite::distributed_blob::CompositeBlob::open(
config, prefix, self,
)
{
let store = BlobStore {
backend: crate::BlobBackend::Composite(db.into()),
compression: config
.property_or_default::<CompressionAlgo>(
("store", id.as_str(), "compression"),
"none",
)
.unwrap_or(CompressionAlgo::None),
};
self.blob_stores.insert(id, store);
}
}
_ => (),
}
}
}
pub async fn parse_lookups(&mut self, config: &mut Config) {

12
crates/store/src/dispatch/blob.rs
View File

@@ -30,11 +30,15 @@ impl BlobStore {
Store::MySQL(store) => store.get_blob(key, read_range).await,
#[cfg(feature = "rocks")]
Store::RocksDb(store) => store.get_blob(key, read_range).await,
#[cfg(feature = "enterprise")]
Store::SQLReadReplica(store) => store.get_blob(key, read_range).await,
Store::None => Err(trc::StoreEvent::NotConfigured.into()),
},
BlobBackend::Fs(store) => store.get_blob(key, read_range).await,
#[cfg(feature = "s3")]
BlobBackend::S3(store) => store.get_blob(key, read_range).await,
#[cfg(feature = "enterprise")]
BlobBackend::Composite(store) => store.get_blob(key, read_range).await,
};
trc::event!(
@@ -106,11 +110,15 @@ impl BlobStore {
Store::MySQL(store) => store.put_blob(key, data.as_ref()).await,
#[cfg(feature = "rocks")]
Store::RocksDb(store) => store.put_blob(key, data.as_ref()).await,
#[cfg(feature = "enterprise")]
Store::SQLReadReplica(store) => store.put_blob(key, data.as_ref()).await,
Store::None => Err(trc::StoreEvent::NotConfigured.into()),
},
BlobBackend::Fs(store) => store.put_blob(key, data.as_ref()).await,
#[cfg(feature = "s3")]
BlobBackend::S3(store) => store.put_blob(key, data.as_ref()).await,
#[cfg(feature = "enterprise")]
BlobBackend::Composite(store) => store.put_blob(key, data.as_ref()).await,
}
.caused_by(trc::location!());
@@ -138,11 +146,15 @@ impl BlobStore {
Store::MySQL(store) => store.delete_blob(key).await,
#[cfg(feature = "rocks")]
Store::RocksDb(store) => store.delete_blob(key).await,
#[cfg(feature = "enterprise")]
Store::SQLReadReplica(store) => store.delete_blob(key).await,
Store::None => Err(trc::StoreEvent::NotConfigured.into()),
},
BlobBackend::Fs(store) => store.delete_blob(key).await,
#[cfg(feature = "s3")]
BlobBackend::S3(store) => store.delete_blob(key).await,
#[cfg(feature = "enterprise")]
BlobBackend::Composite(store) => store.delete_blob(key).await,
}
.caused_by(trc::location!());

2
crates/store/src/dispatch/mod.rs
View File

@@ -26,6 +26,8 @@ impl Store {
Self::MySQL(_) => "mysql",
#[cfg(feature = "rocks")]
Self::RocksDb(_) => "rocksdb",
#[cfg(feature = "enterprise")]
Self::SQLReadReplica(_) => "read_replica",
Self::None => "none",
}
}

22
crates/store/src/dispatch/store.rs
View File

@@ -50,6 +50,8 @@ impl Store {
Self::MySQL(store) => store.get_value(key).await,
#[cfg(feature = "rocks")]
Self::RocksDb(store) => store.get_value(key).await,
#[cfg(feature = "enterprise")]
Self::SQLReadReplica(store) => store.get_value(key).await,
Self::None => Err(trc::StoreEvent::NotConfigured.into()),
}
.caused_by(trc::location!())
@@ -70,6 +72,8 @@ impl Store {
Self::MySQL(store) => store.get_bitmap(key).await,
#[cfg(feature = "rocks")]
Self::RocksDb(store) => store.get_bitmap(key).await,
#[cfg(feature = "enterprise")]
Self::SQLReadReplica(store) => store.get_bitmap(key).await,
Self::None => Err(trc::StoreEvent::NotConfigured.into()),
}
.caused_by(trc::location!())
@@ -114,6 +118,8 @@ impl Store {
Self::MySQL(store) => store.iterate(params, cb).await,
#[cfg(feature = "rocks")]
Self::RocksDb(store) => store.iterate(params, cb).await,
#[cfg(feature = "enterprise")]
Self::SQLReadReplica(store) => store.iterate(params, cb).await,
Self::None => Err(trc::StoreEvent::NotConfigured.into()),
}
.caused_by(trc::location!());
@@ -141,6 +147,8 @@ impl Store {
Self::MySQL(store) => store.get_counter(key).await,
#[cfg(feature = "rocks")]
Self::RocksDb(store) => store.get_counter(key).await,
#[cfg(feature = "enterprise")]
Self::SQLReadReplica(store) => store.get_counter(key).await,
Self::None => Err(trc::StoreEvent::NotConfigured.into()),
}
.caused_by(trc::location!())
@@ -204,6 +212,8 @@ impl Store {
Self::MySQL(store) => store.write(batch).await,
#[cfg(feature = "rocks")]
Self::RocksDb(store) => store.write(batch).await,
#[cfg(feature = "enterprise")]
Self::SQLReadReplica(store) => store.write(batch).await,
Self::None => Err(trc::StoreEvent::NotConfigured.into()),
}
.caused_by(trc::location!())?;
@@ -249,6 +259,8 @@ impl Store {
Self::MySQL(store) => store.write(batch).await,
#[cfg(feature = "rocks")]
Self::RocksDb(store) => store.write(batch).await,
#[cfg(feature = "enterprise")]
Self::SQLReadReplica(store) => store.write(batch).await,
Self::None => Err(trc::StoreEvent::NotConfigured.into()),
};
@@ -303,6 +315,8 @@ impl Store {
Self::MySQL(store) => store.purge_store().await,
#[cfg(feature = "rocks")]
Self::RocksDb(store) => store.purge_store().await,
#[cfg(feature = "enterprise")]
Self::SQLReadReplica(store) => store.purge_store().await,
Self::None => Err(trc::StoreEvent::NotConfigured.into()),
}
.caused_by(trc::location!())
@@ -320,6 +334,8 @@ impl Store {
Self::MySQL(store) => store.delete_range(from, to).await,
#[cfg(feature = "rocks")]
Self::RocksDb(store) => store.delete_range(from, to).await,
#[cfg(feature = "enterprise")]
Self::SQLReadReplica(store) => store.delete_range(from, to).await,
Self::None => Err(trc::StoreEvent::NotConfigured.into()),
}
.caused_by(trc::location!())
@@ -473,6 +489,8 @@ impl Store {
Self::MySQL(store) => store.get_blob(key, range).await,
#[cfg(feature = "rocks")]
Self::RocksDb(store) => store.get_blob(key, range).await,
#[cfg(feature = "enterprise")]
Self::SQLReadReplica(store) => store.get_blob(key, range).await,
Self::None => Err(trc::StoreEvent::NotConfigured.into()),
}
.caused_by(trc::location!())
@@ -490,6 +508,8 @@ impl Store {
Self::MySQL(store) => store.put_blob(key, data).await,
#[cfg(feature = "rocks")]
Self::RocksDb(store) => store.put_blob(key, data).await,
#[cfg(feature = "enterprise")]
Self::SQLReadReplica(store) => store.put_blob(key, data).await,
Self::None => Err(trc::StoreEvent::NotConfigured.into()),
}
.caused_by(trc::location!())
@@ -507,6 +527,8 @@ impl Store {
Self::MySQL(store) => store.delete_blob(key).await,
#[cfg(feature = "rocks")]
Self::RocksDb(store) => store.delete_blob(key).await,
#[cfg(feature = "enterprise")]
Self::SQLReadReplica(store) => store.delete_blob(key).await,
Self::None => Err(trc::StoreEvent::NotConfigured.into()),
}
.caused_by(trc::location!())

8
crates/store/src/lib.rs
View File

@@ -183,6 +183,8 @@ pub enum Store {
MySQL(Arc<MysqlStore>),
#[cfg(feature = "rocks")]
RocksDb(Arc<RocksDbStore>),
#[cfg(feature = "enterprise")]
SQLReadReplica(Arc<backend::composite::read_replica::SQLReadReplica>),
#[default]
None,
}
@@ -205,6 +207,8 @@ pub enum BlobBackend {
Fs(Arc<FsStore>),
#[cfg(feature = "s3")]
S3(Arc<S3Store>),
#[cfg(feature = "enterprise")]
Composite(Arc<backend::composite::distributed_blob::CompositeBlob>),
}
#[derive(Clone)]
@@ -664,6 +668,8 @@ impl Store {
Store::PostgreSQL(_) => true,
#[cfg(feature = "mysql")]
Store::MySQL(_) => true,
#[cfg(feature = "enterprise")]
Store::SQLReadReplica(_) => true,
_ => false,
}
}
@@ -682,6 +688,8 @@ impl std::fmt::Debug for Store {
Self::MySQL(_) => f.debug_tuple("MySQL").finish(),
#[cfg(feature = "rocks")]
Self::RocksDb(_) => f.debug_tuple("RocksDb").finish(),
#[cfg(feature = "enterprise")]
Self::SQLReadReplica(_) => f.debug_tuple("SQLReadReplica").finish(),
Self::None => f.debug_tuple("None").finish(),
}
}