use async_trait::async_trait;
#[cfg(target_os = "linux")]
use caps::CapSet;
#[cfg(target_os = "linux")]
use libc::RLIM_INFINITY;
use nix::sys::{
mman::{MlockAllFlags, mlockall},
resource::{Resource, setrlimit},
};
use nix::unistd::{Group, Uid, User, setgid, setuid};
use tokio::sync::mpsc::Sender;
use tokio::task::JoinHandle;
use tracing::info;
use crate::dirs::Dirs;
use crate::errors::SetupFailure;
use crate::events::{ControlEvent, Event};
use crate::{WFS, Wora};
#[derive(Clone, Copy, Debug, Eq, PartialEq, Hash)]
pub enum RuntimeSignal {
Hangup,
Interrupt,
Quit,
Terminate,
User1,
User2,
}
pub type SignalMapper<AppEv> = std::sync::Arc<dyn Fn(RuntimeSignal) -> Option<Event<AppEv>> + Send + Sync + 'static>;
pub fn default_signal_mapper<AppEv>() -> SignalMapper<AppEv> {
std::sync::Arc::new(|signal| match signal {
RuntimeSignal::Hangup => Some(Event::Control(ControlEvent::ReloadConfiguration)),
RuntimeSignal::Interrupt | RuntimeSignal::Quit | RuntimeSignal::Terminate => {
Some(Event::Control(ControlEvent::Shutdown(Some(chrono::Utc::now().naive_utc()))))
}
RuntimeSignal::User1 => Some(Event::Control(ControlEvent::LogRotation)),
RuntimeSignal::User2 => None,
})
}
#[async_trait]
pub trait AsyncExecutor<AppEv: Send + 'static, AppMetric>: Send + Sync + Clone {
fn id(&self) -> &'static str;
fn dirs(&self) -> &Dirs;
async fn setup(&mut self, wora: &Wora<AppEv, AppMetric>, fs: impl WFS) -> Result<(), SetupFailure>;
async fn spawn_runtime_event_sources(&self, _sender: Sender<Event<AppEv>>) -> Result<Vec<JoinHandle<()>>, SetupFailure> {
Ok(vec![])
}
async fn spawn_runtime_event_sources_with_signal_mapper(
&self,
sender: Sender<Event<AppEv>>,
_signal_mapper: SignalMapper<AppEv>,
) -> Result<Vec<JoinHandle<()>>, SetupFailure> {
self.spawn_runtime_event_sources(sender).await
}
async fn on_runtime_ready(&self, _app_name: &str, _dirs: &Dirs, _fs: impl WFS) -> Result<(), SetupFailure> {
Ok(())
}
async fn on_runtime_draining(&self, _app_name: &str, _dirs: &Dirs, _fs: impl WFS) -> Result<(), SetupFailure> {
Ok(())
}
async fn is_ready(&self, wora: &Wora<AppEv, AppMetric>, fs: impl WFS) -> bool;
async fn on_runtime_stopping(&self, _app_name: &str, _dirs: &Dirs, _fs: impl WFS) -> Result<(), SetupFailure> {
Ok(())
}
async fn end(&self, wora: &Wora<AppEv, AppMetric>, fs: impl WFS);
fn disable_memory_limits(&self) -> Result<(), SetupFailure> {
info!("disabling memory limits");
#[cfg(target_os = "linux")]
setrlimit(Resource::RLIMIT_MEMLOCK, RLIM_INFINITY, RLIM_INFINITY)?;
Ok(())
}
fn disable_paging_mem_to_swap(&self) -> Result<(), SetupFailure> {
info!("disabling paging memory to swap");
mlockall(MlockAllFlags::all())?;
Ok(())
}
fn disable_core_dumps(&self) -> Result<(), SetupFailure> {
info!("disabling core dumps");
setrlimit(Resource::RLIMIT_CORE, 0, 0)?;
Ok(())
}
fn run_as_user_and_group(&self, user_name: &str, group_name: &str) -> Result<(), SetupFailure> {
let new_group = Group::from_name(group_name)?.ok_or_else(|| SetupFailure::UnknownSystemUser(group_name.to_string()))?;
let new_user = User::from_name(user_name)?.ok_or_else(|| SetupFailure::UnknownSystemUser(user_name.to_string()))?;
setgid(new_group.gid)?;
info!("process now runs as group: {} (id: {})", group_name, new_group.gid);
setuid(new_user.uid)?;
info!("process now runs as user: {} (id: {})", user_name, new_user.uid);
Ok(())
}
#[cfg(target_os = "linux")]
fn has_no_caps(&self) -> Result<bool, SetupFailure> {
let effective = caps::read(None, CapSet::Effective)?;
Ok(effective.is_empty())
}
fn is_running_as_root(&self) -> bool {
Uid::effective().is_root()
}
}