449 lines
16 KiB
Rust
449 lines
16 KiB
Rust
use std::sync::Arc;
|
|
|
|
use fs4::tokio::AsyncFileExt;
|
|
use thiserror::Error;
|
|
use tokio::{io::AsyncWriteExt, process::Command};
|
|
use tracing::warn;
|
|
|
|
use crate::{config::settings::NginxSettings, service::master_handler::MasterHandlerError};
|
|
|
|
#[cfg(test)]
|
|
use mockall::predicate::*;
|
|
// TODO: custom error type
|
|
|
|
#[derive(Debug, Error)]
|
|
pub enum FsHandlerError {
|
|
#[error("Invalid output path: {0}")]
|
|
InvalidOutputPath(String),
|
|
#[error("IO error: {0}")]
|
|
IoError(#[from] std::io::Error),
|
|
}
|
|
|
|
impl From<FsHandlerError> for MasterHandlerError {
|
|
fn from(err: FsHandlerError) -> Self {
|
|
MasterHandlerError::MessageHandlingError(format!("File system handling error: {}", err))
|
|
}
|
|
}
|
|
|
|
pub type FsResult<T> = std::result::Result<T, FsHandlerError>;
|
|
type Result<T> = FsResult<T>;
|
|
|
|
#[async_trait::async_trait]
|
|
#[cfg_attr(test, mockall::automock)]
|
|
pub trait FsHandler: Send + Sync + 'static {
|
|
fn get_deployment_id(config_id: &str, version: &str) -> String
|
|
where
|
|
Self: Sized,
|
|
{
|
|
format!("{}-{}", config_id, version)
|
|
}
|
|
// Write a new config file for nginx.
|
|
// The output_path is a relative path to the nginx config directory of the deployment folder. The actual path to the config should not be assumed by the caller, as it can be different in different environments, but will be promised to be relative to the deployment folder for each the corresponding deployment_id. Path traversal is not allowed.
|
|
async fn write_config(
|
|
&self,
|
|
deployment_id: &str,
|
|
config_content: &str,
|
|
output_path: &str,
|
|
) -> Result<String>;
|
|
// Append a new config content to an existing config file for nginx. This is useful for some use cases where we want to keep the existing config and just add some new config content to it. The output_path is a relative path to the nginx config directory of the deployment folder, which should be the same as the one used in write_config function. Path traversal is not allowed.
|
|
async fn append_config(
|
|
&self,
|
|
deployment_id: &str,
|
|
config_content: &str,
|
|
output_path: &str,
|
|
) -> Result<String>;
|
|
|
|
// clean up old config files that are applied to nginx
|
|
// keep only latest n deployments.
|
|
async fn cleanup_config(&self, n: usize) -> Result<()>;
|
|
|
|
// Persist the root config path of the last successful deployment.
|
|
// Survives agent restarts so Reload/Test commands work without a new ConfigUpdate.
|
|
async fn save_last_deployment(&self, root_config_path: &str) -> Result<()>;
|
|
|
|
// Load the last persisted root config path, if any.
|
|
// Returns Ok(None) when no state file exists or it is empty/corrupt.
|
|
async fn load_last_deployment(&self) -> Result<Option<String>>;
|
|
}
|
|
|
|
pub struct FsHandlerImpl {
|
|
settings: Arc<NginxSettings>,
|
|
}
|
|
|
|
impl FsHandlerImpl {
|
|
pub fn new(settings: Arc<NginxSettings>) -> Self {
|
|
Self { settings }
|
|
}
|
|
|
|
fn validate_config_path(config_path: &str) -> Result<()> {
|
|
if !std::path::Path::new(config_path).exists() {
|
|
return Err(FsHandlerError::InvalidOutputPath(format!(
|
|
"Config file not found at path: {}",
|
|
config_path
|
|
)));
|
|
}
|
|
if !std::path::Path::new(config_path).is_file() {
|
|
return Err(FsHandlerError::InvalidOutputPath(format!(
|
|
"Config path is not a file: {}",
|
|
config_path
|
|
)));
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
fn get_deployment_dir(&self) -> std::path::PathBuf {
|
|
std::path::Path::new(&self.settings.nginx_config_path).join("deployments")
|
|
}
|
|
|
|
fn get_deployment_dir_path(&self, deployment_id: &str) -> std::path::PathBuf {
|
|
self.get_deployment_dir().join(deployment_id)
|
|
}
|
|
|
|
fn get_state_file_path(&self) -> std::path::PathBuf {
|
|
std::path::Path::new(&self.settings.nginx_config_path).join(".last_deployment")
|
|
}
|
|
|
|
async fn get_deployment_config_path(
|
|
&self,
|
|
deployment_id: &str,
|
|
output_path: &str,
|
|
create_dir_if_not_exists: bool,
|
|
) -> Result<std::path::PathBuf> {
|
|
let output_path_obj = std::path::Path::new(output_path);
|
|
if output_path_obj.is_absolute() {
|
|
return Err(FsHandlerError::InvalidOutputPath(
|
|
"Output path must be a relative path".into(),
|
|
));
|
|
}
|
|
if output_path_obj
|
|
.components()
|
|
.any(|comp| comp == std::path::Component::ParentDir)
|
|
{
|
|
return Err(FsHandlerError::InvalidOutputPath(
|
|
"Output path must not contain parent directory traversal".into(),
|
|
));
|
|
}
|
|
|
|
let deployment_config_dir = self.get_deployment_dir_path(deployment_id);
|
|
let full_path = deployment_config_dir.join(output_path);
|
|
if create_dir_if_not_exists {
|
|
if let Some(parent) = full_path.parent() {
|
|
tokio::fs::create_dir_all(parent).await?;
|
|
} else {
|
|
tokio::fs::create_dir_all(&deployment_config_dir).await?;
|
|
}
|
|
}
|
|
Ok(full_path)
|
|
}
|
|
}
|
|
|
|
#[async_trait::async_trait]
|
|
impl FsHandler for FsHandlerImpl {
|
|
async fn write_config(
|
|
&self,
|
|
deployment_id: &str,
|
|
config_content: &str,
|
|
output_path: &str,
|
|
) -> Result<String> {
|
|
let full_output_path = self
|
|
.get_deployment_config_path(deployment_id, output_path, true)
|
|
.await?;
|
|
let parent_dir = full_output_path.parent().ok_or_else(|| {
|
|
FsHandlerError::InvalidOutputPath(format!(
|
|
"Failed to get parent directory of output path: {:?}",
|
|
full_output_path
|
|
))
|
|
})?;
|
|
// ensure the parent directory exists before creating the file
|
|
tokio::fs::create_dir_all(parent_dir).await?;
|
|
let mut file = tokio::fs::OpenOptions::new()
|
|
.write(true)
|
|
.create(true)
|
|
.truncate(true)
|
|
.open(full_output_path.clone())
|
|
.await?;
|
|
// lock the file for writing to prevent concurrent write issue
|
|
file.lock_exclusive()?;
|
|
file.write_all(config_content.as_bytes()).await?;
|
|
file.unlock()?;
|
|
file.flush().await?;
|
|
|
|
Ok(full_output_path.to_string_lossy().to_string())
|
|
}
|
|
|
|
async fn append_config(
|
|
&self,
|
|
deployment_id: &str,
|
|
config_content: &str,
|
|
output_path: &str,
|
|
) -> Result<String> {
|
|
let full_output_path = self
|
|
.get_deployment_config_path(deployment_id, output_path, true)
|
|
.await?;
|
|
let mut file = tokio::fs::OpenOptions::new()
|
|
.write(true)
|
|
.create(true)
|
|
.append(true)
|
|
.open(full_output_path.clone())
|
|
.await?;
|
|
// lock the file for writing to prevent concurrent write issue
|
|
file.lock_exclusive()?;
|
|
file.write_all(config_content.as_bytes()).await?;
|
|
file.unlock()?;
|
|
file.flush().await?;
|
|
|
|
Ok(full_output_path.to_string_lossy().to_string())
|
|
}
|
|
|
|
async fn save_last_deployment(&self, root_config_path: &str) -> Result<()> {
|
|
let state_path = self.get_state_file_path();
|
|
let tmp_path = state_path.with_extension("tmp");
|
|
tokio::fs::write(&tmp_path, format!("{}\n", root_config_path)).await?;
|
|
tokio::fs::rename(&tmp_path, &state_path).await?;
|
|
Ok(())
|
|
}
|
|
|
|
async fn load_last_deployment(&self) -> Result<Option<String>> {
|
|
// primary: try state file
|
|
let state_path = self.get_state_file_path();
|
|
if state_path.exists() {
|
|
let content = tokio::fs::read_to_string(&state_path).await?;
|
|
let path = content.trim().to_string();
|
|
if !path.is_empty() {
|
|
return Ok(Some(path));
|
|
}
|
|
}
|
|
|
|
// fallback: scan deployments directory for the newest deployment
|
|
let deployment_dir = self.get_deployment_dir();
|
|
if !deployment_dir.exists() {
|
|
return Ok(None);
|
|
}
|
|
let mut entries = tokio::fs::read_dir(&deployment_dir).await?;
|
|
let mut candidates: Vec<(std::path::PathBuf, std::time::SystemTime)> = Vec::new();
|
|
while let Some(entry) = entries.next_entry().await? {
|
|
if entry.file_type().await.map_or(false, |t| t.is_dir()) {
|
|
if let Ok(mtime) = entry.metadata().await.and_then(|m| m.modified()) {
|
|
candidates.push((entry.path(), mtime));
|
|
}
|
|
}
|
|
}
|
|
// sort descending by mtime (newest first)
|
|
candidates.sort_by(|a, b| b.1.cmp(&a.1));
|
|
|
|
for (dir, _) in &candidates {
|
|
let mut dir_entries = tokio::fs::read_dir(dir).await?;
|
|
while let Some(file) = dir_entries.next_entry().await? {
|
|
if file.file_type().await.map_or(false, |t| t.is_file()) {
|
|
let name = file.file_name().to_string_lossy().to_string();
|
|
if name == "nginx.conf" || name.ends_with(".conf") {
|
|
let path = file.path().to_string_lossy().to_string();
|
|
return Ok(Some(path));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
Ok(None)
|
|
}
|
|
|
|
async fn cleanup_config(&self, n: usize) -> Result<()> {
|
|
let deployment_dir = self.get_deployment_dir();
|
|
// loop through all files in the deployment dir and delete them
|
|
let mut entries = tokio::fs::read_dir(&deployment_dir).await?;
|
|
let mut deployment_ids = Vec::new();
|
|
while let Some(entry) = entries.next_entry().await? {
|
|
let file_type = entry.file_type().await?;
|
|
if file_type.is_dir()
|
|
&& let Some(deployment_id) = entry.file_name().to_str()
|
|
{
|
|
deployment_ids.push(deployment_id.to_string());
|
|
}
|
|
}
|
|
// sort the deployment ids by modified time in descending order and keep the latest n deployments, delete the rest
|
|
deployment_ids.sort_by_key(|id| {
|
|
let path = self.get_deployment_dir_path(id);
|
|
std::fs::metadata(path)
|
|
.and_then(|meta| meta.modified())
|
|
.unwrap_or(std::time::SystemTime::UNIX_EPOCH)
|
|
});
|
|
for deployment_id in deployment_ids.into_iter().skip(n) {
|
|
let path = self.get_deployment_dir_path(&deployment_id);
|
|
// ensure path is within the deplyment and nginx directory to prevent accidental deletion of other files
|
|
if !path.starts_with(&deployment_dir)
|
|
|| !path.starts_with(&self.settings.nginx_config_path)
|
|
{
|
|
warn!(
|
|
"Skipping deletion of path outside of deployment or nginx config directory: {:?}",
|
|
path
|
|
);
|
|
continue;
|
|
}
|
|
tokio::fs::remove_dir_all(path).await?;
|
|
}
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
#[allow(clippy::expect_used)]
|
|
mod tests {
|
|
use super::*;
|
|
use anyhow::Result;
|
|
use std::sync::Arc;
|
|
use tempfile::TempDir;
|
|
|
|
#[tokio::test]
|
|
async fn write_and_append_config_roundtrip() -> Result<()> {
|
|
let temp = TempDir::new()?;
|
|
let settings = NginxSettings {
|
|
nginx_config_path: temp.path().to_string_lossy().to_string(),
|
|
nginx_binary_path: None,
|
|
override_nginx_reload_command: vec![],
|
|
override_nginx_test_command: vec![],
|
|
nginx_reload_timeout_seconds: 1,
|
|
nginx_test_timeout_seconds: 1,
|
|
};
|
|
|
|
let handler = FsHandlerImpl::new(Arc::new(settings));
|
|
|
|
handler
|
|
.write_config("deployment1", "hello", "conf/nginx.conf")
|
|
.await?;
|
|
|
|
let full_path = temp
|
|
.path()
|
|
.join("deployments")
|
|
.join("deployment1")
|
|
.join("conf/nginx.conf");
|
|
|
|
let content = tokio::fs::read_to_string(&full_path).await?;
|
|
assert_eq!(content, "hello");
|
|
|
|
handler
|
|
.append_config("deployment1", " world", "conf/nginx.conf")
|
|
.await?;
|
|
|
|
let content = tokio::fs::read_to_string(&full_path).await?;
|
|
assert_eq!(content, "hello world");
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn write_config_rejects_absolute_and_traversal_paths() -> Result<()> {
|
|
let temp = TempDir::new()?;
|
|
let settings = NginxSettings {
|
|
nginx_config_path: temp.path().to_string_lossy().to_string(),
|
|
nginx_binary_path: None,
|
|
override_nginx_reload_command: vec![],
|
|
override_nginx_test_command: vec![],
|
|
nginx_reload_timeout_seconds: 1,
|
|
nginx_test_timeout_seconds: 1,
|
|
};
|
|
|
|
let handler = FsHandlerImpl::new(Arc::new(settings));
|
|
|
|
let err = handler
|
|
.write_config("d", "x", "/absolute/path.conf")
|
|
.await
|
|
.err();
|
|
assert!(err.is_some());
|
|
|
|
let err = handler.write_config("d", "x", "../escape.conf").await.err();
|
|
assert!(err.is_some());
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn validate_config_path_checks_file_exists_and_is_file() {
|
|
// missing file
|
|
let res = FsHandlerImpl::validate_config_path("/this/path/does/not/exist.conf");
|
|
assert!(res.is_err());
|
|
|
|
// create a temp dir and ensure a directory is rejected
|
|
let temp = TempDir::new().expect("Failed to create temp dir");
|
|
let dir_path = temp.path();
|
|
let res = FsHandlerImpl::validate_config_path(dir_path.to_string_lossy().as_ref());
|
|
assert!(res.is_err());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn get_deployment_config_path_create_flag_behaviour() -> Result<()> {
|
|
let temp = TempDir::new()?;
|
|
let settings = NginxSettings {
|
|
nginx_config_path: temp.path().to_string_lossy().to_string(),
|
|
nginx_binary_path: None,
|
|
override_nginx_reload_command: vec![],
|
|
override_nginx_test_command: vec![],
|
|
nginx_reload_timeout_seconds: 1,
|
|
nginx_test_timeout_seconds: 1,
|
|
};
|
|
|
|
let handler = FsHandlerImpl::new(Arc::new(settings));
|
|
|
|
// when create_dir_if_not_exists = false, directory shouldn't be created
|
|
let path = handler
|
|
.get_deployment_config_path("did", "conf/nginx.conf", false)
|
|
.await?;
|
|
assert!(
|
|
!path
|
|
.parent()
|
|
.expect("Failed to get parent directory of deployment config path")
|
|
.exists()
|
|
);
|
|
|
|
// when create_dir_if_not_exists = true, directory should be created
|
|
let path = handler
|
|
.get_deployment_config_path("did", "conf/nginx.conf", true)
|
|
.await?;
|
|
assert!(
|
|
path.parent()
|
|
.expect("Failed to get parent directory of deployment config path")
|
|
.exists()
|
|
);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn cleanup_config_deletes_expected_deployments() -> Result<()> {
|
|
let temp = TempDir::new()?;
|
|
let settings = NginxSettings {
|
|
nginx_config_path: temp.path().to_string_lossy().to_string(),
|
|
nginx_binary_path: None,
|
|
override_nginx_reload_command: vec![],
|
|
override_nginx_test_command: vec![],
|
|
nginx_reload_timeout_seconds: 1,
|
|
nginx_test_timeout_seconds: 1,
|
|
};
|
|
|
|
let handler = FsHandlerImpl::new(Arc::new(settings));
|
|
let base = temp.path().join("deployments");
|
|
|
|
// create three deployments sequentially so mtimes differ
|
|
for id in &["d1", "d2", "d3"] {
|
|
let p = base.join(id);
|
|
std::fs::create_dir_all(&p)?;
|
|
std::fs::write(p.join("file"), b"x")?;
|
|
std::thread::sleep(std::time::Duration::from_millis(500));
|
|
}
|
|
|
|
// call cleanup keeping 1; current implementation keeps the oldest n, so expect only d1 remains
|
|
handler.cleanup_config(1).await?;
|
|
|
|
let mut exists = vec![];
|
|
for id in &["d1", "d2", "d3"] {
|
|
exists.push((id.to_string(), base.join(id).exists()));
|
|
}
|
|
|
|
// d1 should remain, others removed (matches current implementation behavior)
|
|
assert!(exists.iter().find(|(id, e)| id == "d1" && *e).is_some());
|
|
assert!(exists.iter().find(|(id, e)| id == "d2" && !*e).is_some());
|
|
assert!(exists.iter().find(|(id, e)| id == "d3" && !*e).is_some());
|
|
|
|
Ok(())
|
|
}
|
|
}
|