mirror of
https://github.com/Portabase/agent.git
synced 2026-10-15 00:52:55 +00:00
fix: pgdump-cluster (#68)
* feat: add as_str/from_str to PostgresDumpFormat * feat: resolve pg_dumpall/psql binary names * feat: add include_globals field to database config * feat: add pg_dumpall/psql globals dump and apply * feat: add postgres backup bundle (manifest + build + resolve) * feat: bundle globals into postgres backup when include_globals is set * feat: replay globals before pg_restore when backup archive is a bundle * refactor: bind FD restore tempdir guard once to clear unused warnings * docs: demonstrate include_globals in sample databases.json * chore: silence test-only re-export warning in non-test builds * revert: remove include_globals feature, restore plain pg_dump/pg_restore * feat: add pg_dumpall/psql binary names and is_superuser check * feat: add postgresql-cluster db type and config parsing * feat: pg_dumpall cluster backup and psql restore * feat: route postgresql-cluster through PostgresClusterDatabase * docs: add postgresql-cluster sample to databases.json * refactor: split cluster mode into cluster/ module (backup, restore, database) * test: mirror cluster tests into src/tests/domain/cluster/
This commit is contained in:
@@ -1,5 +1,6 @@
|
||||
use crate::domain::mongodb::database::MongoDatabase;
|
||||
use crate::domain::mysql::database::MySQLDatabase;
|
||||
use crate::domain::postgres::cluster::database::PostgresClusterDatabase;
|
||||
use crate::domain::postgres::database::PostgresDatabase;
|
||||
use crate::domain::postgres::{detect_format_from_file, detect_format_from_size};
|
||||
use crate::domain::redis::database::RedisDatabase;
|
||||
@@ -31,6 +32,7 @@ impl DatabaseFactory {
|
||||
let format = detect_format_from_size(&cfg).await;
|
||||
Arc::new(PostgresDatabase::new(cfg, format))
|
||||
}
|
||||
DbType::PostgresqlCluster => Arc::new(PostgresClusterDatabase::new(cfg)),
|
||||
DbType::Mysql => Arc::new(MySQLDatabase::new(cfg)),
|
||||
DbType::Mariadb => Arc::new(MariaDBDatabase::new(cfg)),
|
||||
DbType::MongoDB => Arc::new(MongoDatabase::new(cfg)),
|
||||
@@ -48,6 +50,7 @@ impl DatabaseFactory {
|
||||
let format = detect_format_from_file(restore_file);
|
||||
Arc::new(PostgresDatabase::new(cfg, format))
|
||||
}
|
||||
DbType::PostgresqlCluster => Arc::new(PostgresClusterDatabase::new(cfg)),
|
||||
DbType::Mysql => Arc::new(MySQLDatabase::new(cfg)),
|
||||
DbType::Mariadb => Arc::new(MariaDBDatabase::new(cfg)),
|
||||
DbType::MongoDB => Arc::new(MongoDatabase::new(cfg)),
|
||||
|
||||
@@ -0,0 +1,79 @@
|
||||
use anyhow::Result;
|
||||
use std::collections::HashMap;
|
||||
use std::path::PathBuf;
|
||||
use std::process::Command;
|
||||
use std::sync::Arc;
|
||||
use std::time::Instant;
|
||||
|
||||
use super::super::connection::{
|
||||
is_superuser, pg_dumpall_binary_name, select_pg_path, server_version,
|
||||
};
|
||||
use crate::services::backup::logger::JobLogger;
|
||||
use crate::services::config::DatabaseConfig;
|
||||
|
||||
pub async fn run(
|
||||
cfg: DatabaseConfig,
|
||||
backup_dir: PathBuf,
|
||||
env: HashMap<String, String>,
|
||||
logger: Arc<JobLogger>,
|
||||
) -> Result<PathBuf> {
|
||||
tokio::task::spawn_blocking(move || -> Result<PathBuf> {
|
||||
logger.log("info", format!("Starting cluster backup for {}", cfg.name));
|
||||
|
||||
let version = match futures::executor::block_on(server_version(&cfg)) {
|
||||
Ok(v) => v,
|
||||
Err(e) => {
|
||||
logger.log("error", format!("Failed to get server version for {}: {:?}", cfg.name, e));
|
||||
return Err(e.into());
|
||||
}
|
||||
};
|
||||
|
||||
match futures::executor::block_on(is_superuser(&cfg)) {
|
||||
Ok(true) => {}
|
||||
Ok(false) => {
|
||||
logger.log("error", format!("postgresql-cluster backup requires a superuser role for {}", cfg.name));
|
||||
anyhow::bail!("postgresql-cluster backup requires a superuser role for {}", cfg.name);
|
||||
}
|
||||
Err(e) => {
|
||||
logger.log("error", format!("Failed to check superuser status for {}: {:?}", cfg.name, e));
|
||||
return Err(e.into());
|
||||
}
|
||||
}
|
||||
|
||||
let pg_dumpall = select_pg_path(&version).join(pg_dumpall_binary_name());
|
||||
let file_path = backup_dir.join(format!("{}.sql", cfg.generated_id));
|
||||
|
||||
logger.log("info", format!("Running pg_dumpall for cluster {} via {:?}", cfg.name, pg_dumpall));
|
||||
|
||||
let start = Instant::now();
|
||||
let output = Command::new(&pg_dumpall)
|
||||
.arg("--host").arg(&cfg.host)
|
||||
.arg("--port").arg(cfg.port.to_string())
|
||||
.arg("--username").arg(&cfg.username)
|
||||
.arg("-v")
|
||||
.arg("-f").arg(&file_path)
|
||||
.envs(env)
|
||||
.output();
|
||||
let duration_ms = start.elapsed().as_millis() as f64;
|
||||
|
||||
match output {
|
||||
Ok(o) => {
|
||||
let stderr = String::from_utf8_lossy(&o.stderr).to_string();
|
||||
let exit_code = o.status.code().unwrap_or(-1);
|
||||
if o.status.success() {
|
||||
logger.log_command("pg_dumpall", if stderr.is_empty() { None } else { Some(stderr) }, Some(0), Some(duration_ms));
|
||||
logger.log("info", format!("Cluster backup completed for {} at {:?}", cfg.name, file_path));
|
||||
Ok(file_path)
|
||||
} else {
|
||||
logger.log_command("pg_dumpall", Some(stderr), Some(exit_code), Some(duration_ms));
|
||||
anyhow::bail!("Cluster backup (pg_dumpall) failed for {}", cfg.name);
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
logger.log_command("pg_dumpall", Some(e.to_string()), Some(-1), Some(duration_ms));
|
||||
Err(e.into())
|
||||
}
|
||||
}
|
||||
})
|
||||
.await?
|
||||
}
|
||||
@@ -0,0 +1,53 @@
|
||||
use anyhow::Result;
|
||||
use async_trait::async_trait;
|
||||
use std::collections::HashMap;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::Arc;
|
||||
|
||||
use super::super::ping;
|
||||
use super::{backup, restore};
|
||||
use crate::domain::factory::Database;
|
||||
use crate::services::backup::logger::JobLogger;
|
||||
use crate::services::config::DatabaseConfig;
|
||||
use crate::utils::locks::{DbOpLock, FileLock};
|
||||
|
||||
pub struct PostgresClusterDatabase {
|
||||
pub cfg: DatabaseConfig,
|
||||
}
|
||||
|
||||
impl PostgresClusterDatabase {
|
||||
pub fn new(cfg: DatabaseConfig) -> Self {
|
||||
Self { cfg }
|
||||
}
|
||||
|
||||
fn build_env(&self) -> HashMap<String, String> {
|
||||
let mut envs = std::env::vars().collect::<HashMap<_, _>>();
|
||||
envs.insert("PGPASSWORD".to_string(), self.cfg.password.to_string());
|
||||
envs
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl Database for PostgresClusterDatabase {
|
||||
fn file_extension(&self) -> &'static str {
|
||||
".sql"
|
||||
}
|
||||
|
||||
async fn ping(&self) -> Result<bool> {
|
||||
ping::run(self.cfg.clone()).await
|
||||
}
|
||||
|
||||
async fn backup(&self, dir: &Path, logger: Arc<JobLogger>) -> Result<PathBuf> {
|
||||
FileLock::acquire(&self.cfg.generated_id, DbOpLock::Backup.as_str()).await?;
|
||||
let res = backup::run(self.cfg.clone(), dir.to_path_buf(), self.build_env(), logger).await;
|
||||
FileLock::release(&self.cfg.generated_id).await?;
|
||||
res
|
||||
}
|
||||
|
||||
async fn restore(&self, file: &Path, logger: Arc<JobLogger>) -> Result<()> {
|
||||
FileLock::acquire(&self.cfg.generated_id, DbOpLock::Restore.as_str()).await?;
|
||||
let res = restore::run(self.cfg.clone(), file.to_path_buf(), self.build_env(), logger).await;
|
||||
FileLock::release(&self.cfg.generated_id).await?;
|
||||
res
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,3 @@
|
||||
pub mod backup;
|
||||
pub mod database;
|
||||
pub mod restore;
|
||||
@@ -0,0 +1,78 @@
|
||||
use anyhow::Result;
|
||||
use std::collections::HashMap;
|
||||
use std::path::PathBuf;
|
||||
use std::process::Command;
|
||||
use std::sync::Arc;
|
||||
use std::time::Instant;
|
||||
|
||||
use super::super::connection::{is_superuser, psql_binary_name, select_pg_path, server_version};
|
||||
use crate::services::backup::logger::JobLogger;
|
||||
use crate::services::config::DatabaseConfig;
|
||||
|
||||
pub async fn run(
|
||||
cfg: DatabaseConfig,
|
||||
restore_file: PathBuf,
|
||||
env: HashMap<String, String>,
|
||||
logger: Arc<JobLogger>,
|
||||
) -> Result<()> {
|
||||
tokio::task::spawn_blocking(move || -> Result<()> {
|
||||
logger.log("info", format!("Starting cluster restore for {}", cfg.name));
|
||||
|
||||
let version = match futures::executor::block_on(server_version(&cfg)) {
|
||||
Ok(v) => v,
|
||||
Err(e) => {
|
||||
logger.log("error", format!("Failed to get server version for {}: {:?}", cfg.name, e));
|
||||
return Err(e.into());
|
||||
}
|
||||
};
|
||||
|
||||
match futures::executor::block_on(is_superuser(&cfg)) {
|
||||
Ok(true) => {}
|
||||
Ok(false) => {
|
||||
logger.log("error", format!("postgresql-cluster restore requires a superuser role for {}", cfg.name));
|
||||
anyhow::bail!("postgresql-cluster restore requires a superuser role for {}", cfg.name);
|
||||
}
|
||||
Err(e) => {
|
||||
logger.log("error", format!("Failed to check superuser status for {}: {:?}", cfg.name, e));
|
||||
return Err(e.into());
|
||||
}
|
||||
}
|
||||
|
||||
let psql = select_pg_path(&version).join(psql_binary_name());
|
||||
|
||||
logger.log("info", format!("Replaying cluster dump for {} via {:?}", cfg.name, psql));
|
||||
|
||||
let start = Instant::now();
|
||||
let output = Command::new(&psql)
|
||||
.arg("--host").arg(&cfg.host)
|
||||
.arg("--port").arg(cfg.port.to_string())
|
||||
.arg("--username").arg(&cfg.username)
|
||||
.arg("--dbname").arg("postgres")
|
||||
.arg("-f").arg(&restore_file)
|
||||
.envs(env)
|
||||
.output();
|
||||
let duration_ms = start.elapsed().as_millis() as f64;
|
||||
|
||||
match output {
|
||||
Ok(o) => {
|
||||
let stderr = String::from_utf8_lossy(&o.stderr).to_string();
|
||||
let stdout = String::from_utf8_lossy(&o.stdout).to_string();
|
||||
let combined = format!("{}{}", stdout, stderr);
|
||||
let exit_code = o.status.code().unwrap_or(-1);
|
||||
if o.status.success() {
|
||||
logger.log_command("psql", if combined.is_empty() { None } else { Some(combined) }, Some(0), Some(duration_ms));
|
||||
logger.log("info", format!("Cluster restore completed for {}", cfg.name));
|
||||
Ok(())
|
||||
} else {
|
||||
logger.log_command("psql", if combined.is_empty() { None } else { Some(combined) }, Some(exit_code), Some(duration_ms));
|
||||
anyhow::bail!("Cluster restore (psql) failed for {}", cfg.name);
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
logger.log_command("psql", Some(e.to_string()), Some(-1), Some(duration_ms));
|
||||
Err(e.into())
|
||||
}
|
||||
}
|
||||
})
|
||||
.await?
|
||||
}
|
||||
@@ -33,32 +33,21 @@ pub async fn server_version(cfg: &DatabaseConfig) -> Result<String> {
|
||||
Ok(version)
|
||||
}
|
||||
|
||||
/// Resolves the `bin` directory of a PostgreSQL installation for the given
|
||||
/// major version, in a cross-platform way.
|
||||
///
|
||||
/// Resolution order:
|
||||
/// 1. The `PG_BIN_DIR` environment variable, if set, is used as-is. This
|
||||
/// allows users/CI to override detection for non-standard installs
|
||||
/// (e.g. portable PostgreSQL distributions, custom install locations).
|
||||
/// 2. Platform-specific default install locations (Debian/Ubuntu packages,
|
||||
/// the official Windows installer, Homebrew/Postgres.app on macOS, and
|
||||
/// common RPM-based layouts on other Linux distros).
|
||||
/// 3. A `PATH` lookup for `pg_dump` (`pg_dump.exe` on Windows), returning
|
||||
/// its parent directory.
|
||||
/// 4. The historical Debian/Ubuntu path as a last-resort fallback, so the
|
||||
/// function keeps returning a `PathBuf` (never panics) even when nothing
|
||||
/// was found, preserving the previous behavior for callers.
|
||||
///
|
||||
/// The override is sourced from `CONFIG.pg_bin_dir` (the `PG_BIN_DIR`
|
||||
/// environment variable). An empty value means "unset" and falls through to
|
||||
/// detection.
|
||||
pub async fn is_superuser(cfg: &DatabaseConfig) -> Result<bool> {
|
||||
let client = connect(cfg).await?;
|
||||
let is_super: bool = client
|
||||
.query_one("SELECT current_setting('is_superuser') = 'on';", &[])
|
||||
.await?
|
||||
.get(0);
|
||||
|
||||
Ok(is_super)
|
||||
}
|
||||
|
||||
|
||||
pub fn select_pg_path(version: &str) -> std::path::PathBuf {
|
||||
select_pg_path_with(version, &CONFIG.pg_bin_dir)
|
||||
}
|
||||
|
||||
/// Inner resolver behind [`select_pg_path`], parameterized over the
|
||||
/// `PG_BIN_DIR` override. Kept pure (no env / no `CONFIG` access) so it is
|
||||
/// unit-testable without mutating process-global state.
|
||||
pub(crate) fn select_pg_path_with(version: &str, pg_bin_dir: &str) -> std::path::PathBuf {
|
||||
let major = version.split('.').next().unwrap_or("17");
|
||||
|
||||
@@ -109,6 +98,22 @@ pub(crate) fn pg_dump_binary_name() -> &'static str {
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn pg_dumpall_binary_name() -> &'static str {
|
||||
if cfg!(target_os = "windows") {
|
||||
"pg_dumpall.exe"
|
||||
} else {
|
||||
"pg_dumpall"
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn psql_binary_name() -> &'static str {
|
||||
if cfg!(target_os = "windows") {
|
||||
"psql.exe"
|
||||
} else {
|
||||
"psql"
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn pg_dump_exists_in(dir: &std::path::Path) -> bool {
|
||||
dir.join(pg_dump_binary_name()).is_file()
|
||||
}
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
pub mod backup;
|
||||
pub(crate) mod cluster;
|
||||
pub(crate) mod connection;
|
||||
pub mod database;
|
||||
mod format;
|
||||
|
||||
Reference in New Issue
Block a user