pub mod api;
pub mod auth;
pub mod config;
pub mod db;
pub mod docker;
pub mod engine;
pub mod error;
pub mod panel;
pub mod rustic;
pub mod scheduler;
pub mod secret;
pub mod state;
#[cfg(test)]
mod test_support;
use std::future::Future;
use std::sync::Arc;
use crate::db::crypto::Clave;
use crate::engine::{Ajustes, Dependencias, Motor};
use crate::scheduler::{Ajustes as AjustesDelReloj, PLAZO_DE_PARADA, Planificador};
use crate::state::{AppState, EstadoDeBase};
pub async fn serve(
state: AppState,
shutdown: impl Future<Output = ()> + Send + 'static,
) -> std::io::Result<()> {
let listener = tokio::net::TcpListener::bind(state.config.bind_address).await?;
axum::serve(listener, panel::app(state, panel::PANEL_POR_DEFECTO))
.with_graceful_shutdown(shutdown)
.await
}
async fn apagado_ordenado(
motor: Motor,
planificador: Planificador,
senal: impl Future<Output = ()>,
) {
senal.await;
planificador.parar(PLAZO_DE_PARADA).await;
motor.cerrar_el_hub();
}
pub async fn shutdown_signal() {
let ctrl_c = async {
match tokio::signal::ctrl_c().await {
Ok(()) => {}
Err(error) => {
tracing::error!(%error, "no se pudo atender SIGINT");
std::future::pending::<()>().await;
}
}
};
#[cfg(unix)]
let terminate = async {
match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) {
Ok(mut signal) => {
signal.recv().await;
}
Err(error) => {
tracing::error!(%error, "no se pudo atender SIGTERM");
std::future::pending::<()>().await;
}
}
};
#[cfg(not(unix))]
let terminate = std::future::pending::<()>();
tokio::select! {
() = ctrl_c => {}
() = terminate => {}
}
tracing::info!("señal de cierre recibida: se deja de aceptar conexiones");
}
const OBJETIVO_DEL_MOTOR: &str = "bollard";
pub const DIRECTIVA_DEL_MOTOR: &str = "bollard=warn";
const NIVELES_QUE_YA_LO_SILENCIAN: [&str; 3] = ["warn", "error", "off"];
fn componer_el_filtro(lo_que_puso: &str) -> (String, Option<String>) {
(
format!("{lo_que_puso},{DIRECTIVA_DEL_MOTOR}"),
aviso_del_filtro(lo_que_puso),
)
}
fn aviso_del_filtro(lo_que_puso: &str) -> Option<String> {
let mut pidio_el_objetivo = false;
let mut pidio_un_sub_objetivo = false;
for directiva in lo_que_puso.split(',') {
let (destino, nivel) = match directiva.split_once('=') {
Some((destino, nivel)) => (destino, Some(nivel.trim())),
None => (directiva, None),
};
let destino = destino.split('[').next().unwrap_or(destino).trim();
let hablador = nivel.is_none_or(|nivel| !NIVELES_QUE_YA_LO_SILENCIAN.contains(&nivel));
if !hablador {
continue;
}
if destino == OBJETIVO_DEL_MOTOR {
pidio_el_objetivo = true;
} else if destino
.strip_prefix(OBJETIVO_DEL_MOTOR)
.is_some_and(|resto| resto.starts_with("::"))
{
pidio_un_sub_objetivo = true;
}
}
if pidio_un_sub_objetivo {
return Some(format!(
"el filtro de registro pide un sub-objetivo de `{OBJETIVO_DEL_MOTOR}` más hablador \
que `warn`: esa directiva es más específica que `{DIRECTIVA_DEL_MOTOR}` y sigue por \
delante, así que ese objetivo puede escribir en el registro el `env` del `exec` de \
la extracción, donde viaja la contraseña de la base"
));
}
if pidio_el_objetivo {
return Some(format!(
"el filtro de registro pide `{OBJETIVO_DEL_MOTOR}` más hablador que `warn`, y el \
proceso lo pisa con `{DIRECTIVA_DEL_MOTOR}`: ese objetivo registra el `env` del \
`exec` de la extracción, donde viaja la contraseña de la base, así que la directiva \
del proyecto no se toca"
));
}
None
}
fn init_tracing(filtro: &str) {
let (aplicado, aviso) = componer_el_filtro(filtro);
let filter = match tracing_subscriber::EnvFilter::try_new(&aplicado) {
Ok(filter) => filter,
Err(error) => {
eprintln!(
"el filtro de registro {aplicado:?} no es válido: {error}; se usa {}",
crate::config::DEFAULT_LOG_FILTER
);
tracing_subscriber::EnvFilter::new(
componer_el_filtro(crate::config::DEFAULT_LOG_FILTER).0,
)
}
};
if let Err(error) = tracing_subscriber::fmt().with_env_filter(filter).try_init() {
eprintln!("no se pudo inicializar el registro: {error}");
}
tracing::info!(filtro = %aplicado, "filtro de registro efectivo");
if let Some(aviso) = aviso {
tracing::warn!("{aviso}");
}
}
#[tokio::main]
async fn main() -> std::process::ExitCode {
let config = match config::Config::from_env() {
Ok(config) => config,
Err(error) => {
eprintln!("kesto no puede arrancar: {error}");
return std::process::ExitCode::FAILURE;
}
};
init_tracing(&config.log_filter);
tracing::info!(config = %config.describe(), "configuración efectiva al arrancar");
for aviso in config.warnings() {
tracing::warn!("{aviso}");
}
let pool = match db::preparar(&config).await {
Ok(pool) => pool,
Err(error) => {
eprintln!("kesto no puede arrancar: {error}");
return std::process::ExitCode::FAILURE;
}
};
let clave = Arc::new(Clave::desde_maestra(&config.master_key));
let db = EstadoDeBase::nuevo(pool, clave);
let ajustes = Ajustes::nuevos(config.docker_host.clone());
let espera_de_parada = ajustes.espera_de_parada;
let motor = match Motor::arrancar(ajustes, dependencias_del_motor(&db)) {
Ok(motor) => motor,
Err(error) => {
eprintln!("kesto no puede arrancar: {error}");
return std::process::ExitCode::FAILURE;
}
};
if let Err(error) = motor.comprobar().await {
tracing::warn!(
%error,
"el motor de copias no puede hablar con el motor de contenedores: las tareas que \
volquen un contenedor fallarán hasta que su socket responda; el resto de Kesto \
funciona"
);
}
let planificador =
match Planificador::arrancar(motor.clone(), db.pool.clone(), AjustesDelReloj::nuevos())
.await
{
Ok(planificador) => planificador,
Err(error) => {
eprintln!("kesto no puede arrancar: {error}");
return std::process::ExitCode::FAILURE;
}
};
planificador.vigilar();
let address = config.bind_address;
let state = match AppState::new(config, db, motor.clone()) {
Ok(state) => state,
Err(error) => {
eprintln!("kesto no puede arrancar: {error}");
return std::process::ExitCode::FAILURE;
}
};
if let Some(auth) = state.auth.as_ref()
&& let Some(aviso) = auth::descubrimiento_al_arrancar(auth).await
{
tracing::warn!("{aviso}");
}
tracing::info!(%address, "escuchando");
let resultado = serve(
state,
apagado_ordenado(motor.clone(), planificador, shutdown_signal()),
)
.await;
let parada = tokio::task::spawn_blocking(move || motor.parar(espera_de_parada));
if let Err(error) = parada.await {
tracing::error!(%error, "la parada del motor no ha llegado a término");
}
if let Err(error) = resultado {
eprintln!("el servidor se detuvo con error: {error}");
return std::process::ExitCode::FAILURE;
}
std::process::ExitCode::SUCCESS
}
fn dependencias_del_motor(db: &EstadoDeBase) -> Dependencias {
Dependencias {
repositorios: db.repositorios.clone(),
targets: db.targets.clone(),
tareas: db.tareas.clone(),
ejecuciones: db.ejecuciones.clone(),
}
}
#[cfg(test)]
mod tests {
use tokio::sync::broadcast::error::TryRecvError;
use super::{DIRECTIVA_DEL_MOTOR, apagado_ordenado, aviso_del_filtro, componer_el_filtro};
use crate::db;
use crate::scheduler::{Ajustes as AjustesDelReloj, Planificador};
use crate::test_support::{
config_con_dsn, directorio_de_prueba, estado_desarrollo, migrador_roto,
};
#[test]
fn el_filtro_que_se_aplica_lleva_la_directiva_del_motor_detras() {
for (puesto, esperado) in [
("info", "info,bollard=warn"),
("debug", "debug,bollard=warn"),
("debug,bollard=debug", "debug,bollard=debug,bollard=warn"),
] {
let (aplicado, _) = componer_el_filtro(puesto);
assert_eq!(aplicado, esperado, "el filtro compuesto no es el esperado");
assert!(
aplicado.contains(DIRECTIVA_DEL_MOTOR),
"el filtro que se aplica no lleva la directiva del motor: {aplicado}"
);
assert!(
aplicado.ends_with(DIRECTIVA_DEL_MOTOR),
"la directiva del proyecto tiene que ir la última para ganar: {aplicado}"
);
assert!(
tracing_subscriber::EnvFilter::try_new(&aplicado).is_ok(),
"el filtro compuesto no lo acepta EnvFilter: {aplicado}"
);
}
let (aplicado, _) = componer_el_filtro("debug,bollard=debug");
let interpretado = tracing_subscriber::EnvFilter::try_new(&aplicado)
.expect("el filtro compuesto es válido")
.to_string();
assert!(
interpretado.contains(DIRECTIVA_DEL_MOTOR),
"el filtro interpretado no lleva la directiva del motor: {interpretado}"
);
assert!(
!interpretado.contains("bollard=debug"),
"la directiva del operador sobrevivió a la del proyecto, que va detrás y tiene que \
sustituirla: {interpretado}"
);
}
#[test]
fn el_filtro_de_reserva_lleva_la_directiva() {
let (reserva, _) = componer_el_filtro(crate::config::DEFAULT_LOG_FILTER);
assert!(
reserva.contains(DIRECTIVA_DEL_MOTOR),
"el filtro de reserva no lleva la directiva del motor: {reserva}"
);
}
#[test]
fn un_filtro_que_pide_el_motor_mas_hablador_deja_constancia() {
for ruidoso in ["debug,bollard=debug", "trace,bollard=trace", "bollard"] {
let aviso = aviso_del_filtro(ruidoso).unwrap_or_else(|| {
panic!("un filtro que pide el motor tiene que avisar: {ruidoso}")
});
assert!(
aviso.contains("bollard") && aviso.contains("contraseña"),
"el aviso no dice de qué va: {aviso}"
);
assert!(
aviso.contains(DIRECTIVA_DEL_MOTOR),
"el aviso no nombra la directiva del proyecto: {aviso}"
);
}
let del_sub_objetivo = aviso_del_filtro("debug,bollard::docker=debug")
.expect("un sub-objetivo del motor más hablador tiene que avisar");
assert!(
del_sub_objetivo.contains("sub-objetivo"),
"el aviso no distingue el sub-objetivo, que es el que queda abierto: {del_sub_objetivo}"
);
for silencioso in [
"debug",
"info",
"debug,bollard=warn",
"info,bollard=error",
"info,bollard=off",
] {
assert!(
aviso_del_filtro(silencioso).is_none(),
"un filtro que no pide nada hablador del motor no puede avisar: {silencioso}"
);
}
}
#[tokio::test]
async fn el_apagado_cierra_el_hub_antes_de_resolver() {
let estado = estado_desarrollo().await;
let motor = estado.motor.clone();
let planificador = Planificador::arrancar(
motor.clone(),
estado.db.pool.clone(),
AjustesDelReloj::nuevos(),
)
.await
.expect("el planificador tiene que poder montarse en la prueba");
let mut cliente = motor
.hub()
.suscribir()
.expect("el hub no está cerrado antes de apagarse");
assert!(
matches!(cliente.try_recv(), Err(TryRecvError::Empty)),
"el canal del hub ya estaba cerrado antes del apagado: la aserción de abajo no diría \
nada sobre este futuro"
);
apagado_ordenado(motor, planificador.clone(), std::future::ready(())).await;
assert!(
matches!(cliente.try_recv(), Err(TryRecvError::Closed)),
"el apagado no cerró el hub, así que el cuerpo del stream de un cliente conectado no \
terminaría y el cierre ordenado se quedaría esperándolo hasta el `SIGKILL` del período \
de gracia"
);
assert!(
planificador.esta_parado(),
"el apagado no paró el planificador: se esperaba que el reloj quedara parado dentro del \
futuro que `serve` recibe, desde el instante en que la señal llega"
);
}
#[tokio::test]
async fn el_arranque_prepara_la_base_antes_de_escuchar() {
let directorio = directorio_de_prueba();
let pool = db::preparar(&config_con_dsn(&directorio.dsn()))
.await
.expect("el arranque debe poder preparar la base");
let tablas = sqlx::query_scalar::<_, String>(
"SELECT name FROM sqlite_master WHERE type = 'table' ORDER BY name",
)
.fetch_all(&pool)
.await
.expect("la consulta de tablas no puede fallar");
for esperada in [
"repositories",
"backup_targets",
"backup_jobs",
"job_executions",
] {
assert!(
tablas.iter().any(|tabla| tabla == esperada),
"el arranque no dejó la tabla {esperada}: {tablas:?}"
);
}
pool.close().await;
}
#[tokio::test]
async fn un_fallo_de_migracion_impide_arrancar() {
let directorio = directorio_de_prueba();
let error = db::preparar_con(&config_con_dsn(&directorio.dsn()), &migrador_roto())
.await
.expect_err("una migración que falla no puede dar un arranque en pie");
assert!(
matches!(error, db::DbError::Migracion { .. }),
"el fallo debe llegar como error de migración: {error}"
);
let pool = db::abrir(&directorio.dsn())
.await
.expect("la base debe poder abrirse para comprobar que no quedó a medias");
let tablas = sqlx::query_scalar::<_, String>(
"SELECT name FROM sqlite_master WHERE type = 'table' ORDER BY name",
)
.fetch_all(&pool)
.await
.expect("la consulta de tablas no puede fallar");
pool.close().await;
assert!(
!tablas.iter().any(|tabla| tabla == "repositories"),
"el arranque falló pero dejó el esquema a medias: {tablas:?}"
);
}
}