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

Minio/S3 blob storage support.

Browse Source
This commit is contained in:
Mauro D
2023-06-06 16:53:29 +00:00
parent c6e45a21e3
commit ab895b2fae
10 changed files with 628 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
crates/jmap/src/api/admin.rs
View File

@@ -42,6 +42,12 @@ impl JMAP {
document_id: 0,
})
.await?;
self.store
.bulk_delete_blob(&store::BlobKind::LinkedMaildir {
account_id,
document_id: 0,
})
.await?;
// Delete mailboxes
let mut batch = BatchBuilder::new();

2
crates/jmap/src/blob/upload.rs
View File

@@ -74,7 +74,7 @@ impl JMAP {
}
}
pub async fn put_blob(&self, kind: &BlobKind, data: &[u8]) -> Result<bool, MethodError> {
pub async fn put_blob(&self, kind: &BlobKind, data: &[u8]) -> Result<(), MethodError> {
self.store.put_blob(kind, data).await.map_err(|err| {
tracing::error!(
event = "error",

1
crates/store/Cargo.toml
View File

@@ -10,6 +10,7 @@ maybe-async = { path = "../maybe-async" }
rocksdb = { version = "0.20.1", optional = true }
foundationdb = { version = "0.7.0", optional = true }
rusqlite = { version = "0.29.0", features = ["bundled"], optional = true }
rust-s3 = { version = "0.33.0", default-features = false, features = ["tokio-rustls-tls"] }
tokio = { version = "1.23", features = ["sync", "fs", "io-util"] }
r2d2 = { version = "0.8.10", optional = true }
futures = { version = "0.3", optional = true }

174
crates/store/src/blob/mod.rs
View File

@@ -24,22 +24,81 @@
pub mod read;
pub mod write;
use std::path::{Path, PathBuf};
use std::{path::PathBuf, time::Duration};
use s3::{
creds::{error::CredentialsError, Credentials},
error::S3Error,
Bucket, Region,
};
use utils::config::Config;
use crate::BlobKind;
pub enum BlobStore {
Local(PathBuf),
Remote(String),
Local(BlobPaths),
Remote(Bucket),
}
pub struct BlobPaths {
path_email: PathBuf,
path_temporary: PathBuf,
path_other: PathBuf,
}
impl BlobStore {
pub async fn new(config: &Config) -> crate::Result<Self> {
Ok(BlobStore::Local(
config.value_require("store.blob.path")?.into(),
))
match config.value_require("store.blob.type")? {
"s3" | "minio" | "gcs" => {
// Obtain region and endpoint from config
let region = config.value_require("store.blob.s3.region")?;
let region = if let Some(endpoint) = config.value("store.blob.s3.endpoint") {
Region::Custom {
region: region.to_string(),
endpoint: endpoint.to_string(),
}
} else {
region.parse().unwrap()
};
let credentials = Credentials::new(
config.value("store.blob.s3.access-key"),
config.value("store.blob.s3.secret-key"),
config.value("store.blob.s3.security-token"),
config.value("store.blob.s3.session-token"),
config.value("store.blob.s3.profile"),
)?;
let timeout =
config.property_or_static::<Duration>("store.blob.s3.timeout", "30s")?;
Ok(BlobStore::Remote(
Bucket::new(
config.value_require("store.blob.s3.bucket")?,
region,
credentials,
)?
.with_path_style()
.with_request_timeout(timeout),
))
}
"local" => {
let path = config.property_require::<PathBuf>("store.blob.local.path")?;
let mut path_email = path.clone();
path_email.push("emails");
let mut path_temporary = path.clone();
path_temporary.push("tmp");
let mut path_other = path;
path_other.push("blobs");
Ok(BlobStore::Local(BlobPaths {
path_email,
path_temporary,
path_other,
}))
}
unknown => Err(crate::Error::InternalError(format!(
"Unknown blob store type: {unknown}",
))),
}
}
}
@@ -49,26 +108,41 @@ impl From<std::io::Error> for crate::Error {
}
}
fn get_path(base_path: &Path, kind: &BlobKind) -> crate::Result<PathBuf> {
let mut path = base_path.to_path_buf();
impl From<S3Error> for crate::Error {
fn from(err: S3Error) -> Self {
Self::InternalError(format!("S3 error: {}", err))
}
}
impl From<CredentialsError> for crate::Error {
fn from(err: CredentialsError) -> Self {
Self::InternalError(format!("S3 Credentials error: {}", err))
}
}
fn get_local_path(base_path: &BlobPaths, kind: &BlobKind) -> PathBuf {
match kind {
BlobKind::LinkedMaildir {
account_id,
document_id,
} => {
let mut path = base_path.path_email.to_path_buf();
path.push(format!("{:x}", account_id));
path.push("Maildir");
path.push("cur");
path.push(format!("{:x}", document_id));
path
}
BlobKind::Linked {
account_id,
collection,
document_id,
} => {
let mut path = base_path.path_other.to_path_buf();
path.push(format!("{:x}", account_id));
path.push(format!("{:x}", collection));
path.push(format!("{:x}", document_id));
}
BlobKind::LinkedMaildir {
account_id,
document_id,
} => {
path.push(format!("{:x}", account_id));
path.push("Maildir");
path.push("cur");
path.push(format!("{:x}", document_id));
path
}
BlobKind::Temporary {
account_id,
@@ -77,22 +151,27 @@ fn get_path(base_path: &Path, kind: &BlobKind) -> crate::Result<PathBuf> {
creation_day,
seq,
} => {
path.push("tmp");
let mut path = base_path.path_temporary.to_path_buf();
path.push(creation_year.to_string());
path.push(creation_month.to_string());
path.push(creation_day.to_string());
path.push(format!("{:x}_{:x}", account_id, seq));
path
}
}
Ok(path)
}
fn get_root_path(base_path: &Path, kind: &BlobKind) -> crate::Result<PathBuf> {
let mut path = base_path.to_path_buf();
fn get_local_root_path(base_path: &BlobPaths, kind: &BlobKind) -> PathBuf {
match kind {
BlobKind::Linked { account_id, .. } | BlobKind::LinkedMaildir { account_id, .. } => {
BlobKind::LinkedMaildir { account_id, .. } => {
let mut path = base_path.path_email.to_path_buf();
path.push(format!("{:x}", account_id));
path
}
BlobKind::Linked { account_id, .. } => {
let mut path = base_path.path_other.to_path_buf();
path.push(format!("{:x}", account_id));
path
}
BlobKind::Temporary {
creation_year,
@@ -100,12 +179,55 @@ fn get_root_path(base_path: &Path, kind: &BlobKind) -> crate::Result<PathBuf> {
creation_day,
..
} => {
path.push("tmp");
let mut path = base_path.path_temporary.to_path_buf();
path.push(creation_year.to_string());
path.push(creation_month.to_string());
path.push(creation_day.to_string());
path
}
}
Ok(path)
}
fn get_s3_path(kind: &BlobKind) -> String {
match kind {
BlobKind::LinkedMaildir {
account_id,
document_id,
} => format!("/{:x}/{:x}", account_id, document_id),
BlobKind::Linked {
account_id,
collection,
document_id,
} => format!("/{:x}/{:x}/{:x}", account_id, collection, document_id),
BlobKind::Temporary {
account_id,
creation_year,
creation_month,
creation_day,
seq,
} => format!(
"/tmp/{}/{}/{}/{:x}_{:x}",
creation_year, creation_month, creation_day, account_id, seq
),
}
}
fn get_s3_root_path(kind: &BlobKind) -> String {
match kind {
BlobKind::LinkedMaildir { account_id, .. } => {
format!("/{:x}/", account_id)
}
BlobKind::Linked { account_id, .. } => {
format!("/{:x}/", account_id)
}
BlobKind::Temporary {
creation_year,
creation_month,
creation_day,
..
} => format!(
"/tmp/{}/{}/{}/",
creation_year, creation_month, creation_day
),
}
}

31
crates/store/src/blob/read.rs
View File

@@ -30,7 +30,7 @@ use tokio::{
use crate::{BlobKind, Store};
use super::{get_path, BlobStore};
use super::{get_local_path, get_s3_path, BlobStore};
impl Store {
pub async fn get_blob(
@@ -40,7 +40,7 @@ impl Store {
) -> crate::Result<Option<Vec<u8>>> {
match &self.blob {
BlobStore::Local(base_path) => {
let blob_path = get_path(base_path, kind)?;
let blob_path = get_local_path(base_path, kind);
let blob_size = match fs::metadata(&blob_path).await {
Ok(m) => m.len(),
Err(_) => return Ok(None),
@@ -70,7 +70,32 @@ impl Store {
buf
}))
}
BlobStore::Remote(_) => todo!(),
BlobStore::Remote(bucket) => {
let path = get_s3_path(kind);
let response = if range.start != 0 || range.end != u32::MAX {
bucket
.get_object_range(
path,
range.start as u64,
Some(range.end.saturating_sub(1) as u64),
)
.await
} else {
bucket.get_object(path).await
};
match response {
Ok(response) if (200..300).contains(&response.status_code()) => {
Ok(Some(response.to_vec()))
}
Ok(response) if response.status_code() == 404 => Ok(None),
Ok(response) => Err(crate::Error::InternalError(format!(
"S3 error code {}: {}",
response.status_code(),
String::from_utf8_lossy(response.as_slice())
))),
Err(err) => Err(err.into()),
}
}
}
}
}

101
crates/store/src/blob/write.rs
View File

@@ -30,22 +30,33 @@ use tokio::{
use crate::{BlobKind, Store};
use super::{get_path, get_root_path, BlobStore};
use super::{get_local_path, get_local_root_path, get_s3_path, get_s3_root_path, BlobStore};
impl Store {
pub async fn put_blob(&self, kind: &BlobKind, data: &[u8]) -> crate::Result<bool> {
pub async fn put_blob(&self, kind: &BlobKind, data: &[u8]) -> crate::Result<()> {
match &self.blob {
BlobStore::Local(base_path) => {
let blob_path = get_path(base_path, kind)?;
let blob_path = get_local_path(base_path, kind);
fs::create_dir_all(blob_path.parent().unwrap()).await?;
let mut blob_file = File::create(&blob_path).await?;
blob_file.write_all(data).await?;
blob_file.flush().await?;
Ok(true)
Ok(())
}
BlobStore::Remote(bucket) => {
let path = get_s3_path(kind);
match bucket.put_object(path, data).await {
Ok(response) if (200..300).contains(&response.status_code()) => Ok(()),
Ok(response) => Err(crate::Error::InternalError(format!(
"S3 error code {}: {}",
response.status_code(),
String::from_utf8_lossy(response.as_slice())
))),
Err(e) => Err(e.into()),
}
}
BlobStore::Remote(_) => todo!(),
}
}
@@ -55,20 +66,19 @@ impl Store {
dest: &BlobKind,
range: Option<Range<u32>>,
) -> crate::Result<bool> {
match &self.blob {
BlobStore::Local(base_path) => {
let dest_path = get_path(base_path, dest)?;
if let Some(range) = range {
if let Some(bytes) = self.get_blob(src, range).await? {
self.put_blob(dest, &bytes).await?;
Ok(true)
} else {
Ok(false)
}
} else {
match &self.blob {
BlobStore::Local(base_path) => {
let dest_path = get_local_path(base_path, dest);
let src_path = get_local_path(base_path, src);
if let Some(range) = range {
if let Some(bytes) = self.get_blob(src, range).await? {
fs::create_dir_all(dest_path.parent().unwrap()).await?;
fs::write(dest_path, bytes).await?;
Ok(true)
} else {
Ok(false)
}
} else {
let src_path = get_path(base_path, src)?;
if fs::metadata(&src_path).await.is_ok() {
fs::create_dir_all(dest_path.parent().unwrap()).await?;
fs::copy(src_path, dest_path).await?;
@@ -77,15 +87,24 @@ impl Store {
Ok(false)
}
}
BlobStore::Remote(bucket) => {
let src_path = get_s3_path(src);
let dest_path = get_s3_path(dest);
bucket
.copy_object_internal(src_path, dest_path)
.await
.map(|code| (200..300).contains(&code))
.map_err(|e| e.into())
}
}
BlobStore::Remote(_) => todo!(),
}
}
pub async fn delete_blob(&self, kind: &BlobKind) -> crate::Result<bool> {
match &self.blob {
BlobStore::Local(base_path) => {
let blob_path = get_path(base_path, kind)?;
let blob_path = get_local_path(base_path, kind);
if blob_path.exists() {
fs::remove_file(&blob_path).await?;
@@ -94,16 +113,52 @@ impl Store {
Ok(false)
}
}
BlobStore::Remote(_) => todo!(),
BlobStore::Remote(bucket) => {
let path = get_s3_path(kind);
bucket
.delete_object(path)
.await
.map(|response| (200..300).contains(&response.status_code()))
.map_err(|e| e.into())
}
}
}
pub async fn bulk_delete_blob(&self, kind: &BlobKind) -> crate::Result<()> {
match &self.blob {
BlobStore::Local(base_path) => fs::remove_dir_all(get_root_path(base_path, kind)?)
BlobStore::Local(base_path) => fs::remove_dir_all(get_local_root_path(base_path, kind))
.await
.map_err(Into::into),
BlobStore::Remote(_) => todo!(),
BlobStore::Remote(bucket) => {
let prefix = get_s3_root_path(kind);
let prefix_base = prefix.strip_prefix('/').unwrap();
let mut is_truncated = true;
while is_truncated {
for item in bucket.list(prefix.clone(), None).await? {
is_truncated = item.is_truncated && !item.contents.is_empty();
for object in item.contents {
if object.key.starts_with(&prefix)
|| object.key.starts_with(prefix_base)
{
let result = bucket.delete_object(object.key).await?;
if !(200..300).contains(&result.status_code()) {
return Err(crate::Error::InternalError(format!(
"Failed to delete bucket item, code {}: {}",
result.status_code(),
String::from_utf8_lossy(result.as_slice())
)));
}
} else {
tracing::debug!(
"Unexpected S3 object while deleting: {}",
item.name
);
}
}
}
}
Ok(())
}
}
}
}