mirror of
https://github.com/Portabase/agent.git
synced 2026-10-14 00:30:05 +00:00
feat(storage): add RcloneProvider upload path
This commit is contained in:
@@ -9,6 +9,7 @@ use providers::azure_blob;
|
||||
use providers::google_cloud_storage;
|
||||
use providers::google_drive;
|
||||
use providers::local;
|
||||
use providers::rclone;
|
||||
use providers::s3;
|
||||
use std::sync::Arc;
|
||||
use tracing::{error, info};
|
||||
@@ -39,6 +40,7 @@ pub fn get_provider(storage: &DatabaseStorage) -> Option<Box<dyn StorageProvider
|
||||
"google-cloud-storage" => Some(Box::new(
|
||||
google_cloud_storage::GoogleCloudStorageProvider {},
|
||||
)),
|
||||
"rclone" => Some(Box::new(rclone::RcloneProvider {})),
|
||||
_ => {
|
||||
error!("Unknown storage provider: {}", storage.provider);
|
||||
None
|
||||
|
||||
@@ -1,2 +1,114 @@
|
||||
pub mod helpers;
|
||||
pub mod models;
|
||||
|
||||
use crate::core::context::Context;
|
||||
use crate::services::api::models::agent::status::DatabaseStorage;
|
||||
use crate::services::backup::models::{BackupResult, UploadResult};
|
||||
use crate::services::storage::StorageProvider;
|
||||
use crate::services::storage::providers::rclone::helpers::{
|
||||
rcat, remote_target, validate_config, write_config,
|
||||
};
|
||||
use crate::services::storage::providers::rclone::models::RcloneProviderConfig;
|
||||
use crate::utils::common::BackupMethod;
|
||||
use crate::utils::file::{full_file_name, full_file_path};
|
||||
use crate::utils::stream::build_stream;
|
||||
use async_trait::async_trait;
|
||||
use std::sync::Arc;
|
||||
use tokio::fs;
|
||||
use tracing::{error, info};
|
||||
|
||||
pub struct RcloneProvider {}
|
||||
|
||||
/// Failure shorthand — every early return reports the same shape.
|
||||
fn failed(storage_id: &str, error: impl ToString, total_size: Option<u64>) -> UploadResult {
|
||||
UploadResult {
|
||||
storage_id: storage_id.to_string(),
|
||||
success: false,
|
||||
error: Some(error.to_string()),
|
||||
remote_file_path: None,
|
||||
total_size,
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl StorageProvider for RcloneProvider {
|
||||
async fn upload(
|
||||
&self,
|
||||
ctx: Arc<Context>,
|
||||
result: BackupResult,
|
||||
_method: BackupMethod,
|
||||
storage: &DatabaseStorage,
|
||||
encrypt: Option<bool>,
|
||||
_backup_storage_id: &str,
|
||||
) -> UploadResult {
|
||||
let storage_id = storage.id.clone();
|
||||
|
||||
let Some(file_path) = result.backup_file else {
|
||||
return failed(&storage_id, "Missing backup file path", None);
|
||||
};
|
||||
|
||||
let total_size = match fs::metadata(&file_path).await {
|
||||
Ok(meta) => meta.len(),
|
||||
Err(e) => {
|
||||
error!("Failed to get file size: {}", e);
|
||||
return failed(&storage_id, e, None);
|
||||
}
|
||||
};
|
||||
|
||||
let config: RcloneProviderConfig = match storage.clone().config.try_into() {
|
||||
Ok(c) => c,
|
||||
Err(e) => {
|
||||
error!("rclone config deserialization failed: {}", e);
|
||||
return failed(&storage_id, e, Some(total_size));
|
||||
}
|
||||
};
|
||||
|
||||
if let Err(e) = validate_config(&config.config_text, &config.remote_name) {
|
||||
error!("rclone config rejected: {}", e);
|
||||
return failed(&storage_id, e, Some(total_size));
|
||||
}
|
||||
|
||||
let encrypt = encrypt.unwrap_or(false);
|
||||
|
||||
let upload = match build_stream(&file_path, encrypt, &ctx.edge_key.master_key_b64).await {
|
||||
Ok(u) => u,
|
||||
Err(e) => {
|
||||
error!("Stream build failed: {}", e);
|
||||
return failed(&storage_id, e, Some(total_size));
|
||||
}
|
||||
};
|
||||
|
||||
let file_name = full_file_name(encrypt);
|
||||
let remote_file_path = full_file_path(&file_name, storage.folder_name.as_deref());
|
||||
|
||||
// Held for the whole transfer; the temp file is removed when it drops.
|
||||
let config_file = match write_config(&config.config_text) {
|
||||
Ok(f) => f,
|
||||
Err(e) => {
|
||||
error!("rclone config write failed: {}", e);
|
||||
return failed(&storage_id, e, Some(total_size));
|
||||
}
|
||||
};
|
||||
|
||||
let target = remote_target(&config.remote_name, &config.remote_path, &remote_file_path);
|
||||
|
||||
info!("Starting rclone upload to {}", target);
|
||||
|
||||
match rcat(config_file.path(), &target, upload.stream).await {
|
||||
Ok(()) => {
|
||||
info!("rclone upload successful: {}", remote_file_path);
|
||||
UploadResult {
|
||||
storage_id,
|
||||
success: true,
|
||||
error: None,
|
||||
remote_file_path: Some(remote_file_path),
|
||||
total_size: Some(total_size),
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
error!("rclone upload failed: {:?}", e);
|
||||
failed(&storage_id, e, Some(total_size))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -259,3 +259,131 @@ async fn rcat_reports_rclone_stderr_when_the_remote_is_unreachable() {
|
||||
"rclone stderr must be included: {msg}"
|
||||
);
|
||||
}
|
||||
|
||||
use crate::core::context::Context;
|
||||
use crate::services::api::ApiClient;
|
||||
use crate::services::backup::models::BackupResult;
|
||||
use crate::services::config::DbType;
|
||||
use crate::services::storage::providers::rclone::RcloneProvider;
|
||||
use crate::services::storage::{StorageProvider, get_provider};
|
||||
use crate::utils::common::BackupMethod;
|
||||
use crate::utils::edge_key::EdgeKey;
|
||||
|
||||
use std::io::Write as _;
|
||||
use std::sync::Arc;
|
||||
use tempfile::NamedTempFile;
|
||||
|
||||
fn test_context() -> Arc<Context> {
|
||||
Arc::new(Context {
|
||||
edge_key: EdgeKey {
|
||||
server_url: String::new(),
|
||||
agent_id: "agent-1".to_string(),
|
||||
master_key_b64: String::new(),
|
||||
},
|
||||
api: ApiClient::new(String::new()),
|
||||
})
|
||||
}
|
||||
|
||||
fn storage_for(config_text: &str, remote_path: &str) -> DatabaseStorage {
|
||||
serde_json::from_value(serde_json::json!({
|
||||
"id": "storage-1",
|
||||
"provider": "rclone",
|
||||
"folderName": "backups",
|
||||
"config": {
|
||||
"configText": config_text,
|
||||
"remoteName": "minio",
|
||||
"remotePath": remote_path,
|
||||
}
|
||||
}))
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn factory_resolves_the_rclone_provider_key() {
|
||||
let storage = storage_for(OVH_CONFIG, "bucket");
|
||||
assert!(
|
||||
get_provider(&storage).is_some(),
|
||||
"get_provider must recognise the \"rclone\" key"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn provider_uploads_an_unencrypted_backup_to_minio() {
|
||||
init_tracing_for_test();
|
||||
|
||||
let (_container, endpoint) = start_minio().await;
|
||||
let config_text = minio_config(&endpoint);
|
||||
|
||||
let bootstrap = write_config(&config_text).unwrap();
|
||||
rclone_ok(bootstrap.path(), &["mkdir", &format!("minio:{BUCKET}")]);
|
||||
|
||||
let payload = vec![42u8; 64 * 1024];
|
||||
let mut backup_file = NamedTempFile::new().unwrap();
|
||||
backup_file.write_all(&payload).unwrap();
|
||||
backup_file.flush().unwrap();
|
||||
|
||||
let storage = storage_for(&config_text, BUCKET);
|
||||
|
||||
let result = RcloneProvider {}
|
||||
.upload(
|
||||
test_context(),
|
||||
BackupResult {
|
||||
generated_id: "db-1".to_string(),
|
||||
db_type: DbType::Postgresql,
|
||||
status: "success".to_string(),
|
||||
backup_file: Some(backup_file.path().to_path_buf()),
|
||||
code: None,
|
||||
},
|
||||
BackupMethod::Automatic,
|
||||
&storage,
|
||||
Some(false),
|
||||
"backup-storage-1",
|
||||
)
|
||||
.await;
|
||||
|
||||
assert!(result.success, "upload failed: {:?}", result.error);
|
||||
assert_eq!(result.total_size, Some(payload.len() as u64));
|
||||
|
||||
let remote_file_path = result.remote_file_path.expect("remote path must be reported");
|
||||
assert!(
|
||||
remote_file_path.starts_with("backups/"),
|
||||
"folder_name must prefix the path: {remote_file_path}"
|
||||
);
|
||||
|
||||
let target = remote_target("minio", BUCKET, &remote_file_path);
|
||||
assert_eq!(rclone_ok(bootstrap.path(), &["cat", &target]), payload);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn provider_refuses_a_blocked_backend_without_spawning_rclone() {
|
||||
init_tracing_for_test();
|
||||
|
||||
let mut backup_file = NamedTempFile::new().unwrap();
|
||||
backup_file.write_all(b"payload").unwrap();
|
||||
backup_file.flush().unwrap();
|
||||
|
||||
let storage = storage_for("[minio]\ntype = local\n", "bucket");
|
||||
|
||||
let result = RcloneProvider {}
|
||||
.upload(
|
||||
test_context(),
|
||||
BackupResult {
|
||||
generated_id: "db-1".to_string(),
|
||||
db_type: DbType::Postgresql,
|
||||
status: "success".to_string(),
|
||||
backup_file: Some(backup_file.path().to_path_buf()),
|
||||
code: None,
|
||||
},
|
||||
BackupMethod::Automatic,
|
||||
&storage,
|
||||
Some(false),
|
||||
"backup-storage-1",
|
||||
)
|
||||
.await;
|
||||
|
||||
assert!(!result.success);
|
||||
assert!(
|
||||
result.error.unwrap_or_default().contains("local"),
|
||||
"the error must name the rejected backend type"
|
||||
);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user