conmonrs/

server.rs

#![deny(missing_docs)]

#[cfg(feature = "tracing")]
use crate::telemetry::Telemetry;
use crate::{
    child_reaper::ChildReaper,
    config::{Commands, Config, LogDriver, LogLevel, Verbosity},
    container_io::{ContainerIO, ContainerIOType},
    fd_socket::FdSocket,
    init::{DefaultInit, Init},
    journal::Journal,
    listener::{DefaultListener, Listener},
    pause::Pause,
    streaming_server::StreamingServer,
    version::Version,
};
use anyhow::{Context, Result, format_err};
use capnp::text_list::Reader;
use capnp_rpc::{RpcSystem, rpc_twoparty_capnp::Side, twoparty};
#[cfg(feature = "tracing")]
use clap::crate_name;
use conmon_common::conmon_capnp::conmon::{self, CgroupManager};
use futures::{AsyncReadExt, FutureExt};
use libc::_exit;
use nix::{
    errno::Errno,
    sys::signal::Signal,
    unistd::{ForkResult, fork},
};
#[cfg(feature = "tracing")]
use opentelemetry::trace::{FutureExt as OpenTelemetryFutureExt, TracerProvider};
#[cfg(feature = "tracing")]
use opentelemetry_sdk::trace::SdkTracerProvider;
use std::{fs::File, io::Write, os::fd::RawFd, path::Path, process, str::FromStr, sync::Arc};
use tokio::{
    fs,
    runtime::{Builder, Handle},
    signal::unix::{SignalKind, signal},
    sync::{RwLock, oneshot},
    task::{self, LocalSet},
};
use tokio_util::compat::TokioAsyncReadCompatExt;
use tracing::{Instrument, debug, debug_span, info};
#[cfg(feature = "tracing")]
use tracing_opentelemetry::OpenTelemetrySpanExt;
use tracing_subscriber::{filter::LevelFilter, layer::SubscriberExt, prelude::*};
use twoparty::VatNetwork;

#[derive(Debug)]
/// The main server structure.
pub struct Server {
    /// Server configuration.
    config: Arc<Config>,

    /// Child reaper instance.
    reaper: Arc<ChildReaper>,

    /// Fd socket instance.
    fd_socket: Arc<FdSocket>,

    /// OpenTelemetry tracer instance.
    #[cfg(feature = "tracing")]
    tracer: Option<SdkTracerProvider>,

    /// Streaming server instance.
    streaming_server: Arc<RwLock<StreamingServer>>,
}

impl Server {
    /// Server configuration.
    pub(crate) fn config(&self) -> &Arc<Config> {
        &self.config
    }

    /// Child reaper instance.
    pub(crate) fn reaper(&self) -> &Arc<ChildReaper> {
        &self.reaper
    }

    /// Fd socket instance.
    pub(crate) fn fd_socket(&self) -> &Arc<FdSocket> {
        &self.fd_socket
    }

    /// OpenTelemetry tracer instance.
    #[cfg(feature = "tracing")]
    pub(crate) fn tracer(&self) -> &Option<SdkTracerProvider> {
        &self.tracer
    }

    /// Streaming server instance.
    pub(crate) fn streaming_server(&self) -> &Arc<RwLock<StreamingServer>> {
        &self.streaming_server
    }

    /// Create a new `Server` instance.
    pub fn new() -> Result<Self> {
        let server = Self {
            config: Arc::new(Config::default()),
            reaper: Default::default(),
            fd_socket: Default::default(),
            #[cfg(feature = "tracing")]
            tracer: Default::default(),
            streaming_server: Default::default(),
        };

        if let Some(v) = server.config().version() {
            Version::new(v == Verbosity::Full).print();
            process::exit(0);
        }

        if let Some(v) = server.config().version_json() {
            Version::new(v == Verbosity::Full).print_json()?;
            process::exit(0);
        }

        if let Some(Commands::Pause {
            base_path,
            pod_id,
            ipc,
            pid,
            net,
            user,
            uts,
            uid_mappings,
            gid_mappings,
        }) = server.config().command()
        {
            Pause::run(
                base_path,
                pod_id,
                *ipc,
                *pid,
                *net,
                *user,
                *uts,
                uid_mappings,
                gid_mappings,
            )
            .context("run pause")?;
            process::exit(0);
        }

        server.config().validate().context("validate config")?;

        Self::init().context("init self")?;
        Ok(server)
    }

    /// Close file descriptors inherited from whatever started us.
    ///
    /// conmon-rs daemonizes and outlives the process that spawned it, so any
    /// descriptor it inherits stays open for the lifetime of the server. If
    /// the spawning process was itself started with extra descriptors open,
    /// holding them can keep that process tree from ever finishing - a test
    /// harness waiting for its own pipe to close, for example.
    ///
    /// conmon does the same thing, see `close_all_fds_ge_than(3)` there. Any
    /// descriptor conmon-rs actually needs is either created after this point
    /// or received later over the fd socket, so nothing above stderr is worth
    /// keeping.
    fn close_inherited_fds() {
        for fd in Self::inherited_fds() {
            // The descriptor used to enumerate them is already closed by now,
            // so check before closing to avoid closing an unrelated descriptor
            // that reused the number. This runs before tokio starts, so nothing
            // else can be opening descriptors concurrently.
            //
            // These are raw libc calls rather than the `nix` wrappers on
            // purpose: `BorrowedFd`/`OwnedFd` require the descriptor to be open
            // for the duration of the borrow, and whether it is still open is
            // exactly what is unknown here. `fcntl(2)` and `close(2)` take a
            // plain integer with no such precondition and report `EBADF`.
            //
            // SAFETY: both calls are defined for any integer; a stale or
            // invalid descriptor number fails with `EBADF` rather than
            // affecting anything else.
            if unsafe { libc::fcntl(fd, libc::F_GETFD) } != -1 {
                unsafe { libc::close(fd) };
            }
        }
    }

    /// Every open descriptor above stderr. Split out from `close_inherited_fds`
    /// so it can be tested without closing the test runner's own descriptors.
    ///
    /// This is Linux only: it enumerates `/proc/self/fd` and would report
    /// nothing at all where that is unavailable, silently leaking the inherited
    /// descriptors. conmon-rs already requires Linux elsewhere (`prctl`, epoll,
    /// journald), so the assertion below just makes a port fail here loudly
    /// rather than lose the behaviour quietly. A log line would not work,
    /// `close_inherited_fds` runs well before `init_logging`.
    fn inherited_fds() -> Vec<RawFd> {
        const {
            assert!(
                cfg!(target_os = "linux"),
                "inherited fds are enumerated through /proc/self/fd, which is Linux only"
            )
        };

        let mut fds = Vec::new();

        if let Ok(entries) = std::fs::read_dir("/proc/self/fd") {
            for entry in entries.flatten() {
                if let Some(fd) = entry
                    .file_name()
                    .to_str()
                    .and_then(|name| name.parse::<RawFd>().ok())
                    && fd > 2
                {
                    fds.push(fd);
                }
            }
        }

        fds
    }

    /// Start the `Server` instance and consume it.
    pub fn start(self) -> Result<()> {
        // We need to fork as early as possible, especially before setting up tokio.
        // If we don't, the child will have a strange thread space and we're at risk of deadlocking.
        // We also have to treat the parent as the child (as described in [1]) to ensure we don't
        // interrupt the child's execution.
        // 1: https://docs.rs/nix/0.23.0/nix/unistd/fn.fork.html#safety
        if !self.config().skip_fork() {
            match unsafe { fork()? } {
                ForkResult::Parent { child, .. } => {
                    write!(File::create(self.config().conmon_pidfile())?, "{child}")?;
                    unsafe { _exit(0) };
                }
                ForkResult::Child => (),
            }
        }

        // Deliberately outside the `skip_fork` check above: the server outlives
        // its parent either way, so the descriptors have to go either way.
        Self::close_inherited_fds();

        // now that we've forked, set self to childreaper
        let ret = unsafe { libc::prctl(libc::PR_SET_CHILD_SUBREAPER, 1, 0, 0, 0) };
        if ret != 0 {
            return Err(Errno::last()).context("set child subreaper");
        }

        #[cfg(feature = "tracing")]
        let tracer = self.tracer().clone();

        debug!("Configuring Tokio runtime with current_thread");
        let rt = Builder::new_current_thread()
            .enable_io()
            .enable_time()
            .build()?;
        rt.block_on(self.spawn_tasks())?;

        #[cfg(feature = "tracing")]
        if let Some(tracer) = tracer {
            tracer.shutdown().context("shutdown tracer")?;
        }

        rt.shutdown_timeout(std::time::Duration::from_secs(15));
        Ok(())
    }

    fn init() -> Result<()> {
        let init = Init::<DefaultInit>::default();
        init.unset_locale()?;
        init.set_default_umask();
        // While we could configure this, standard practice has it as -1000,
        // so it may be YAGNI to add configuration.
        init.set_oom_score("-1000")
    }

    fn init_logging(&mut self) -> Result<()> {
        let level = LevelFilter::from_str(self.config().log_level().as_ref())
            .context("convert log level filter")?;

        #[cfg(feature = "tracing")]
        let telemetry_layer = if self.config().enable_tracing() {
            let tracer = Telemetry::layer(self.config().tracing_endpoint())
                .context("build telemetry layer")?;

            self.tracer = Some(tracer.clone());

            tracing_opentelemetry::layer()
                .with_tracer(tracer.tracer(crate_name!()))
                .into()
        } else {
            None
        };

        let registry = tracing_subscriber::registry();
        #[cfg(feature = "tracing")]
        let registry = registry.with(telemetry_layer);

        match self.config().log_driver() {
            LogDriver::None => {}
            LogDriver::Stdout => {
                let layer = tracing_subscriber::fmt::layer()
                    .with_target(true)
                    .with_line_number(true)
                    .with_filter(level);
                registry
                    .with(layer)
                    .try_init()
                    .context("init stdout fmt layer")?;
                info!("Using stdout logger");
            }
            LogDriver::Systemd => {
                let layer = tracing_subscriber::fmt::layer()
                    .with_target(true)
                    .with_line_number(true)
                    .without_time()
                    .with_writer(Journal)
                    .with_filter(level);
                registry
                    .with(layer)
                    .try_init()
                    .context("init journald fmt layer")?;
                info!("Using systemd/journald logger");
            }
        }
        info!("Set log level to: {}", self.config().log_level());
        Ok(())
    }

    /// Spawns all required tokio tasks.
    async fn spawn_tasks(mut self) -> Result<()> {
        self.init_logging().context("init logging")?;

        let (shutdown_tx, shutdown_rx) = oneshot::channel();
        let socket = self.config().socket();
        let fd_socket = self.config().fd_socket();
        let reaper = self.reaper.clone();

        let signal_handler_span = debug_span!("signal_handler");
        let backend_span = debug_span!("backend");

        // Run both signal handler and backend inside spawn_blocking with LocalSet
        // This allows concurrent execution within the LocalSet while preventing
        // blocking of the main current_thread runtime
        #[cfg(feature = "tracing")]
        let result = task::spawn_blocking(move || {
            Handle::current().block_on(async move {
                let local = LocalSet::new();

                // Spawn signal handler as a local task
                local.spawn_local(
                    Self::start_signal_handler(reaper, socket, fd_socket, shutdown_tx)
                        .with_context(signal_handler_span.context())
                        .instrument(signal_handler_span),
                );

                // Run backend on the LocalSet
                local
                    .run_until(
                        self.start_backend(shutdown_rx)
                            .with_context(backend_span.context())
                            .instrument(backend_span),
                    )
                    .await
            })
        })
        .await?;
        #[cfg(not(feature = "tracing"))]
        let result = task::spawn_blocking(move || {
            Handle::current().block_on(async move {
                let local = LocalSet::new();

                // Spawn signal handler as a local task
                local.spawn_local(
                    Self::start_signal_handler(reaper, socket, fd_socket, shutdown_tx)
                        .instrument(signal_handler_span),
                );

                // Run backend on the LocalSet
                local
                    .run_until(self.start_backend(shutdown_rx).instrument(backend_span))
                    .await
            })
        })
        .await?;
        result
    }

    async fn start_signal_handler<T: AsRef<Path>>(
        reaper: Arc<ChildReaper>,
        socket: T,
        fd_socket: T,
        shutdown_tx: oneshot::Sender<()>,
    ) -> Result<()> {
        let mut sigterm = signal(SignalKind::terminate())?;
        let mut sigint = signal(SignalKind::interrupt())?;

        tokio::select! {
            _ = sigterm.recv() => {
                info!("Received SIGTERM");
            }
            _ = sigint.recv() => {
                info!("Received SIGINT");
            }
        }

        if let Some(pause) = Pause::maybe_shared() {
            pause.stop();
        }

        debug!("Starting grandchildren cleanup task");
        // Always use SIGKILL to ensure immediate termination of container processes
        reaper
            .kill_grandchildren(Signal::SIGKILL)
            .await
            .context("unable to kill grandchildren")?;

        debug!("Sending shutdown message");
        shutdown_tx
            .send(())
            .map_err(|_| format_err!("unable to send shutdown message"))?;

        debug!("Removing socket file {}", socket.as_ref().display());
        fs::remove_file(socket)
            .await
            .context("remove existing socket file")?;

        debug!("Removing fd socket file {}", fd_socket.as_ref().display());
        fs::remove_file(fd_socket)
            .await
            .or_else(|err| {
                if err.kind() == std::io::ErrorKind::NotFound {
                    Ok(())
                } else {
                    Err(err)
                }
            })
            .context("remove existing fd socket file")
    }

    async fn start_backend(self, mut shutdown_rx: oneshot::Receiver<()>) -> Result<()> {
        let listener =
            Listener::<DefaultListener>::default().bind_long_path(self.config().socket())?;
        let client: conmon::Client = capnp_rpc::new_client(self);

        loop {
            let stream = tokio::select! {
                _ = &mut shutdown_rx => {
                    debug!("Received shutdown message");
                    return Ok(())
                }
                stream = listener.accept() => {
                    stream?.0
                },
            };
            let (reader, writer) = TokioAsyncReadCompatExt::compat(stream).split();
            let network = Box::new(VatNetwork::new(
                reader,
                writer,
                Side::Server,
                Default::default(),
            ));
            let rpc_system = RpcSystem::new(network, Some(client.clone().client));
            task::spawn_local(Box::pin(rpc_system.map(|_| ())));
        }
    }
}

pub(crate) struct GenerateRuntimeArgs<'a> {
    pub(crate) config: &'a Config,
    pub(crate) id: &'a str,
    pub(crate) container_io: &'a ContainerIO,
    pub(crate) pidfile: &'a Path,
    pub(crate) cgroup_manager: CgroupManager,
}

impl GenerateRuntimeArgs<'_> {
    const SYSTEMD_CGROUP_ARG: &'static str = "--systemd-cgroup";
    const RUNTIME_CRUN: &'static str = "crun";
    const LOG_LEVEL_FLAG_CRUN: &'static str = "--log-level";

    /// Generate the OCI runtime CLI arguments from the provided parameters.
    pub fn create_args(
        self,
        bundle_path: &Path,
        global_args: Reader,
        command_args: Reader,
    ) -> Result<Vec<String>> {
        // Pre-allocate capacity for typical arg count to reduce reallocations
        let mut args = Vec::with_capacity(16);
        args.extend(self.default_args().context("build default runtime args")?);

        if let Some(rr) = self.config.runtime_root() {
            args.push(format!("--root={}", rr.display()));
        }

        if self.cgroup_manager == CgroupManager::Systemd {
            args.push(Self::SYSTEMD_CGROUP_ARG.into());
        }

        for arg in global_args {
            args.push(arg?.to_string()?);
        }

        // Use static strings where possible to avoid allocations
        args.push("create".into());
        args.push("--bundle".into());
        args.push(bundle_path.display().to_string());
        args.push("--pid-file".into());
        args.push(self.pidfile.display().to_string());

        for arg in command_args {
            args.push(arg?.to_string()?);
        }

        if let ContainerIOType::Terminal(terminal) = self.container_io.typ() {
            args.push(format!("--console-socket={}", terminal.path().display()));
        }

        args.push(self.id.into());

        debug!("Runtime args {:?}", args.join(" "));
        Ok(args)
    }

    /// Generate the OCI runtime CLI arguments from the provided parameters.
    pub(crate) fn exec_sync_args(&self, command: Reader) -> Result<Vec<String>> {
        let mut args = self
            .exec_sync_args_without_command()
            .context("exec sync args without command")?;

        for arg in command {
            args.push(arg?.to_string()?);
        }

        debug!("Exec args {:?}", args.join(" "));
        Ok(args)
    }

    pub(crate) fn exec_sync_args_without_command(&self) -> Result<Vec<String>> {
        // Pre-allocate capacity for typical arg count
        let mut args = Vec::with_capacity(12);
        args.extend(self.default_args().context("build default runtime args")?);

        if let Some(rr) = self.config.runtime_root() {
            args.push(format!("--root={}", rr.display()));
        }

        if self.cgroup_manager == CgroupManager::Systemd {
            args.push(Self::SYSTEMD_CGROUP_ARG.into());
        }

        // Use static strings to avoid allocations
        args.push("exec".into());
        args.push("-d".into());

        if let ContainerIOType::Terminal(terminal) = self.container_io.typ() {
            args.push(format!("--console-socket={}", terminal.path().display()));
            args.push("--tty".into());
        }

        args.push(format!("--pid-file={}", self.pidfile.display()));
        args.push(self.id.into());

        Ok(args)
    }

    /// Build the default arguments for any provided runtime.
    fn default_args(&self) -> Result<Vec<String>> {
        let mut args = vec![];

        if self
            .config
            .runtime()
            .file_name()
            .context("no filename in path")?
            == Self::RUNTIME_CRUN
        {
            debug!("Found crun used as runtime");
            args.push(format!("--log=journald:{}", self.id));

            match self.config.log_level() {
                &LogLevel::Debug | &LogLevel::Error => args.push(format!(
                    "{}={}",
                    Self::LOG_LEVEL_FLAG_CRUN,
                    self.config.log_level()
                )),
                &LogLevel::Warn => args.push(format!("{}=warning", Self::LOG_LEVEL_FLAG_CRUN)),
                _ => {}
            }
        }

        if let Some(rr) = self.config.runtime_root() {
            args.push(format!("--root={}", rr.display()));
        }

        if self.cgroup_manager == CgroupManager::Systemd {
            args.push(Self::SYSTEMD_CGROUP_ARG.into());
        }

        Ok(args)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::os::fd::AsRawFd;

    #[test]
    fn inherited_fds_reports_descriptors_above_stderr() -> Result<()> {
        let file = tempfile::tempfile()?;
        let fd = file.as_raw_fd();

        let fds = Server::inherited_fds();

        assert!(
            fds.contains(&fd),
            "expected an open descriptor {fd} to be reported, got {fds:?}"
        );
        // stdin, stdout and stderr are never reported, they are not ours to close.
        assert!(!fds.contains(&0));
        assert!(!fds.contains(&1));
        assert!(!fds.contains(&2));

        Ok(())
    }
}