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

Redis lookup backend implementation

Browse Source
This commit is contained in:
mdecimus
2023-12-12 18:45:52 +01:00
parent 78afc703f5
commit b7869901ee
15 changed files with 586 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
crates/main/Cargo.toml
View File

@@ -31,8 +31,8 @@ tracing = "0.1"
jemallocator = "0.5.0"
[features]
#default = ["sqlite", "foundationdb", "postgres", "mysql", "rocks", "elastic", "s3"]
default = ["sqlite", "postgres", "mysql", "foundationdb", "rocks"]
#default = ["sqlite", "foundationdb", "postgres", "mysql", "rocks", "elastic", "s3", "redis"]
default = ["sqlite", "postgres", "mysql", "redis"]
sqlite = ["store/sqlite"]
foundationdb = ["store/foundation"]
postgres = ["store/postgres"]
@@ -40,3 +40,4 @@ mysql = ["store/mysql"]
rocks = ["store/rocks"]
elastic = ["store/elastic"]
s3 = ["store/s3"]
redis = ["store/redis"]

3
crates/store/Cargo.toml
View File

@@ -42,6 +42,8 @@ regex = "1.7.0"
reqwest = { version = "0.11", default-features = false, features = ["rustls-tls-webpki-roots", "blocking"] }
flate2 = "1.0"
async-trait = "0.1.68"
redis = { version = "0.24.0", features = [ "tokio-comp", "tokio-rustls-comp", "tls-rustls-insecure", "tls-rustls-webpki-roots", "cluster-async"], optional = true }
deadpool = { version = "0.10.0", features = ["managed"], optional = true }
[dev-dependencies]
tokio = { version = "1.23", features = ["full"] }
@@ -55,6 +57,7 @@ mysql = ["mysql_async"]
s3 = ["rust-s3"]
foundation = ["foundationdb", "futures"]
fdb-chunked-bm = []
redis = ["dep:redis", "deadpool"]
test_mode = []

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

@@ -31,6 +31,8 @@ pub mod memory;
pub mod mysql;
#[cfg(feature = "postgres")]
pub mod postgres;
#[cfg(feature = "redis")]
pub mod redis;
#[cfg(feature = "rocks")]
pub mod rocksdb;
#[cfg(feature = "s3")]

90
crates/store/src/backend/redis/lookup.rs Normal file
View File

@@ -0,0 +1,90 @@
/*
* Copyright (c) 2023 Stalwart Labs Ltd.
*
* This file is part of Stalwart Mail Server.
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of
* the License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
* in the LICENSE file at the top-level directory of this distribution.
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*
* You can be released from the requirements of the AGPLv3 license by
* purchasing a commercial license. Please contact licensing@stalw.art
* for more details.
*/
use redis::AsyncCommands;
use crate::{Deserialize, LookupKey, LookupValue};
use super::{RedisPool, RedisStore};
impl RedisStore {
pub async fn key_set(&self, key: Vec<u8>, value: LookupValue<Vec<u8>>) -> crate::Result<()> {
match &self.pool {
RedisPool::Single(pool) => self.key_set_(pool.get().await?.as_mut(), key, value).await,
RedisPool::Cluster(pool) => self.key_set_(pool.get().await?.as_mut(), key, value).await,
}
}
pub async fn key_get<T: Deserialize + std::fmt::Debug + 'static>(
&self,
key: LookupKey,
) -> crate::Result<LookupValue<T>> {
match &self.pool {
RedisPool::Single(pool) => self.key_get_(pool.get().await?.as_mut(), key).await,
RedisPool::Cluster(pool) => self.key_get_(pool.get().await?.as_mut(), key).await,
}
}
async fn key_get_<T: Deserialize + std::fmt::Debug + 'static>(
&self,
conn: &mut impl AsyncCommands,
key: LookupKey,
) -> crate::Result<LookupValue<T>> {
match key {
LookupKey::Key(key) => {
if let Some(value) = conn.get::<_, Option<Vec<u8>>>(key).await? {
T::deserialize(&value).map(|value| LookupValue::Value { value, expires: 0 })
} else {
Ok(LookupValue::None)
}
}
LookupKey::Counter(key) => {
let value: Option<i64> = conn.get(key).await?;
Ok(LookupValue::Counter {
num: value.unwrap_or(0),
})
}
}
}
async fn key_set_(
&self,
conn: &mut impl AsyncCommands,
key: Vec<u8>,
value: LookupValue<Vec<u8>>,
) -> crate::Result<()> {
match value {
LookupValue::Value { value, expires } => {
if expires > 0 {
conn.set_ex(key, value, expires).await?;
} else {
conn.set(key, value).await?;
}
}
LookupValue::Counter { num } => conn.incr(key, num).await?,
LookupValue::None => (),
}
Ok(())
}
}

159
crates/store/src/backend/redis/mod.rs Normal file
View File

@@ -0,0 +1,159 @@
/*
* Copyright (c) 2023 Stalwart Labs Ltd.
*
* This file is part of Stalwart Mail Server.
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of
* the License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
* in the LICENSE file at the top-level directory of this distribution.
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*
* You can be released from the requirements of the AGPLv3 license by
* purchasing a commercial license. Please contact licensing@stalw.art
* for more details.
*/
use std::time::Duration;
use deadpool::{
managed::{Manager, Pool, PoolError},
Runtime,
};
use redis::{
cluster::{ClusterClient, ClusterClientBuilder},
Client, RedisError,
};
use utils::config::{utils::AsKey, Config};
pub mod lookup;
pub mod pool;
pub struct RedisStore {
pool: RedisPool,
}
struct RedisConnectionManager {
client: Client,
timeout: Duration,
}
struct RedisClusterConnectionManager {
client: ClusterClient,
timeout: Duration,
}
enum RedisPool {
Single(Pool<RedisConnectionManager>),
Cluster(Pool<RedisClusterConnectionManager>),
}
impl RedisStore {
pub async fn open(config: &Config, prefix: impl AsKey) -> crate::Result<Self> {
let prefix = prefix.as_key();
let db = if let Some(url) = config.value((&prefix, "url")) {
Self {
pool: RedisPool::Single(build_pool(
config,
&prefix,
RedisConnectionManager {
client: Client::open(url)?,
timeout: config.property_or_static((&prefix, "timeout"), "10s")?,
},
)?),
}
} else {
let addresses = config
.values((&prefix, "urls"))
.map(|(_, v)| v.to_string())
.collect::<Vec<_>>();
if addresses.is_empty() {
return Err(crate::Error::InternalError(format!(
"No Redis cluster URLs specified for {prefix:?}"
)));
}
let mut builder = ClusterClientBuilder::new(addresses.into_iter());
if let Some(value) = config.property((&prefix, "username"))? {
builder = builder.username(value);
}
if let Some(value) = config.property((&prefix, "password"))? {
builder = builder.password(value);
}
if let Some(value) = config.property((&prefix, "retries"))? {
builder = builder.retries(value);
}
if let Some(value) = config.property::<Duration>((&prefix, "max-retry-wait"))? {
builder = builder.max_retry_wait(value.as_secs());
}
if let Some(value) = config.property::<Duration>((&prefix, "min-retry-wait"))? {
builder = builder.min_retry_wait(value.as_secs());
}
if let Some(true) = config.property::<bool>((&prefix, "read-from-replicas"))? {
builder = builder.read_from_replicas();
}
Self {
pool: RedisPool::Cluster(build_pool(
config,
&prefix,
RedisClusterConnectionManager {
client: builder.build()?,
timeout: config.property_or_static((&prefix, "timeout"), "10s")?,
},
)?),
}
};
Ok(db)
}
}
fn build_pool<M: Manager>(
config: &Config,
prefix: &str,
manager: M,
) -> utils::config::Result<Pool<M>> {
Pool::builder(manager)
.runtime(Runtime::Tokio1)
.max_size(config.property_or_static((prefix, "pool.max-connections"), "10")?)
.create_timeout(
config
.property_or_static::<Duration>((prefix, "pool.create-timeout"), "30s")?
.into(),
)
.wait_timeout(config.property_or_static((prefix, "pool.wait-timeout"), "30s")?)
.recycle_timeout(config.property_or_static((prefix, "pool.recycle-timeout"), "30s")?)
.build()
.map_err(|err| {
format!(
"Failed to build pool for {prefix:?}: {err}",
prefix = prefix,
err = err
)
})
}
impl From<PoolError<RedisError>> for crate::Error {
fn from(value: PoolError<RedisError>) -> Self {
crate::Error::InternalError(format!("Redis pool error: {}", value))
}
}
impl From<PoolError<crate::Error>> for crate::Error {
fn from(value: PoolError<crate::Error>) -> Self {
crate::Error::InternalError(format!("Connection pool {}", value))
}
}
impl From<RedisError> for crate::Error {
fn from(value: RedisError) -> Self {
crate::Error::InternalError(format!("Redis error: {}", value))
}
}

83
crates/store/src/backend/redis/pool.rs Normal file
View File

@@ -0,0 +1,83 @@
/*
* Copyright (c) 2023 Stalwart Labs Ltd.
*
* This file is part of Stalwart Mail Server.
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of
* the License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
* in the LICENSE file at the top-level directory of this distribution.
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*
* You can be released from the requirements of the AGPLv3 license by
* purchasing a commercial license. Please contact licensing@stalw.art
* for more details.
*/
use async_trait::async_trait;
use deadpool::managed;
use redis::{
aio::{Connection, ConnectionLike},
cluster_async::ClusterConnection,
};
use super::{RedisClusterConnectionManager, RedisConnectionManager};
#[async_trait]
impl managed::Manager for RedisConnectionManager {
type Type = Connection;
type Error = crate::Error;
async fn create(&self) -> Result<Connection, crate::Error> {
match tokio::time::timeout(self.timeout, self.client.get_tokio_connection()).await {
Ok(conn) => conn.map_err(Into::into),
Err(_) => Err(crate::Error::InternalError(
"Redis connection timeout".into(),
)),
}
}
async fn recycle(
&self,
conn: &mut Connection,
_: &managed::Metrics,
) -> managed::RecycleResult<crate::Error> {
conn.req_packed_command(&redis::cmd("PING"))
.await
.map(|_| ())
.map_err(|err| managed::RecycleError::Backend(err.into()))
}
}
#[async_trait]
impl managed::Manager for RedisClusterConnectionManager {
type Type = ClusterConnection;
type Error = crate::Error;
async fn create(&self) -> Result<ClusterConnection, crate::Error> {
match tokio::time::timeout(self.timeout, self.client.get_async_connection()).await {
Ok(conn) => conn.map_err(Into::into),
Err(_) => Err(crate::Error::InternalError(
"Redis connection timeout".into(),
)),
}
}
async fn recycle(
&self,
conn: &mut ClusterConnection,
_: &managed::Metrics,
) -> managed::RecycleResult<crate::Error> {
conn.req_packed_command(&redis::cmd("PING"))
.await
.map(|_| ())
.map_err(|err| managed::RecycleError::Backend(err.into()))
}
}

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

@@ -53,6 +53,9 @@ use crate::backend::rocksdb::RocksDbStore;
#[cfg(feature = "elastic")]
use crate::backend::elastic::ElasticSearchStore;
#[cfg(feature = "redis")]
use crate::backend::redis::RedisStore;
#[async_trait]
pub trait ConfigStore {
async fn parse_stores(&self) -> utils::config::Result<Stores>;
@@ -161,6 +164,13 @@ impl ConfigStore for Config {
);
continue;
}
#[cfg(feature = "redis")]
"redis" => {
config
.lookup_stores
.insert(store_id, RedisStore::open(self, prefix).await?.into());
continue;
}
"memory" => {
let prefix = prefix.as_key();
for lookup_id in self.sub_keys((&prefix, "lookup")) {

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

@@ -49,6 +49,10 @@ impl LookupStore {
)),
},
LookupStore::Memory(store) => store.query(query, params),
#[cfg(feature = "redis")]
LookupStore::Redis(_) => Err(crate::Error::InternalError(
"Redis does not support queries".into(),
)),
};
tracing::trace!( context = "store", event = "query", query = query, result = ?result);
@@ -81,6 +85,8 @@ impl LookupStore {
batch.ops.push(Operation::Value { class, op });
store.write(batch.build()).await
}
#[cfg(feature = "redis")]
LookupStore::Redis(store) => store.key_set(key, value).await,
LookupStore::Memory(_) => unimplemented!(),
}
}
@@ -110,6 +116,8 @@ impl LookupStore {
.await
.map(|num| LookupValue::Counter { num }),
},
#[cfg(feature = "redis")]
LookupStore::Redis(store) => store.key_get(key).await,
LookupStore::Memory(_) => unimplemented!(),
}
}
@@ -137,7 +145,7 @@ impl LookupStore {
store
.iterate(IterateParams::new(from_key, to_key), |key, value| {
if value.deserialize_be_u64(0)? < current_time {
expired_keys.push(key.to_vec());
expired_keys.push(key.get(1..).unwrap_or_default().to_vec());
}
Ok(true)
})
@@ -159,6 +167,8 @@ impl LookupStore {
}
}
}
#[cfg(feature = "redis")]
LookupStore::Redis(store) => {}
LookupStore::Memory(_) => {}
}

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

@@ -434,7 +434,7 @@ impl Store {
value
);
}
SUBSPACE_INDEX_VALUES if key[0] >= 2 => {
SUBSPACE_INDEX_VALUES if key[0] >= 3 => {
// Ignore named keys
return Ok(true);
}

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

@@ -60,6 +60,9 @@ use backend::rocksdb::RocksDbStore;
#[cfg(feature = "elastic")]
use backend::elastic::ElasticSearchStore;
#[cfg(feature = "redis")]
use backend::redis::RedisStore;
pub trait Deserialize: Sized + Sync + Send {
fn deserialize(bytes: &[u8]) -> crate::Result<Self>;
}
@@ -236,6 +239,8 @@ pub enum FtsStore {
pub enum LookupStore {
Store(Store),
Memory(Arc<MemoryStore>),
#[cfg(feature = "redis")]
Redis(Arc<RedisStore>),
}
#[cfg(feature = "sqlite")]
@@ -293,6 +298,13 @@ impl From<ElasticSearchStore> for FtsStore {
}
}
#[cfg(feature = "redis")]
impl From<RedisStore> for LookupStore {
fn from(store: RedisStore) -> Self {
Self::Redis(Arc::new(store))
}
}
impl From<Store> for FtsStore {
fn from(store: Store) -> Self {
Self::Store(store)