conmonrs/
server.rs

1#![deny(missing_docs)]
2
3#[cfg(feature = "tracing")]
4use crate::telemetry::Telemetry;
5use crate::{
6    child_reaper::ChildReaper,
7    config::{Commands, Config, LogDriver, LogLevel, Verbosity},
8    container_io::{ContainerIO, ContainerIOType},
9    fd_socket::FdSocket,
10    init::{DefaultInit, Init},
11    journal::Journal,
12    listener::{DefaultListener, Listener},
13    pause::Pause,
14    streaming_server::StreamingServer,
15    version::Version,
16};
17use anyhow::{Context, Result, format_err};
18use capnp::text_list::Reader;
19use capnp_rpc::{RpcSystem, rpc_twoparty_capnp::Side, twoparty};
20#[cfg(feature = "tracing")]
21use clap::crate_name;
22use conmon_common::conmon_capnp::conmon::{self, CgroupManager};
23use futures::{AsyncReadExt, FutureExt};
24use libc::_exit;
25use nix::{
26    errno::Errno,
27    sys::signal::Signal,
28    unistd::{ForkResult, fork},
29};
30#[cfg(feature = "tracing")]
31use opentelemetry::trace::{FutureExt as OpenTelemetryFutureExt, TracerProvider};
32#[cfg(feature = "tracing")]
33use opentelemetry_sdk::trace::SdkTracerProvider;
34use std::{fs::File, io::Write, path::Path, process, str::FromStr, sync::Arc};
35use tokio::{
36    fs,
37    runtime::{Builder, Handle},
38    signal::unix::{SignalKind, signal},
39    sync::{RwLock, oneshot},
40    task::{self, LocalSet},
41};
42use tokio_util::compat::TokioAsyncReadCompatExt;
43use tracing::{Instrument, debug, debug_span, info};
44#[cfg(feature = "tracing")]
45use tracing_opentelemetry::OpenTelemetrySpanExt;
46use tracing_subscriber::{filter::LevelFilter, layer::SubscriberExt, prelude::*};
47use twoparty::VatNetwork;
48
49#[derive(Debug)]
50/// The main server structure.
51pub struct Server {
52    /// Server configuration.
53    config: Arc<Config>,
54
55    /// Child reaper instance.
56    reaper: Arc<ChildReaper>,
57
58    /// Fd socket instance.
59    fd_socket: Arc<FdSocket>,
60
61    /// OpenTelemetry tracer instance.
62    #[cfg(feature = "tracing")]
63    tracer: Option<SdkTracerProvider>,
64
65    /// Streaming server instance.
66    streaming_server: Arc<RwLock<StreamingServer>>,
67}
68
69impl Server {
70    /// Server configuration.
71    pub(crate) fn config(&self) -> &Arc<Config> {
72        &self.config
73    }
74
75    /// Child reaper instance.
76    pub(crate) fn reaper(&self) -> &Arc<ChildReaper> {
77        &self.reaper
78    }
79
80    /// Fd socket instance.
81    pub(crate) fn fd_socket(&self) -> &Arc<FdSocket> {
82        &self.fd_socket
83    }
84
85    /// OpenTelemetry tracer instance.
86    #[cfg(feature = "tracing")]
87    pub(crate) fn tracer(&self) -> &Option<SdkTracerProvider> {
88        &self.tracer
89    }
90
91    /// Streaming server instance.
92    pub(crate) fn streaming_server(&self) -> &Arc<RwLock<StreamingServer>> {
93        &self.streaming_server
94    }
95
96    /// Create a new `Server` instance.
97    pub fn new() -> Result<Self> {
98        let server = Self {
99            config: Arc::new(Config::default()),
100            reaper: Default::default(),
101            fd_socket: Default::default(),
102            #[cfg(feature = "tracing")]
103            tracer: Default::default(),
104            streaming_server: Default::default(),
105        };
106
107        if let Some(v) = server.config().version() {
108            Version::new(v == Verbosity::Full).print();
109            process::exit(0);
110        }
111
112        if let Some(v) = server.config().version_json() {
113            Version::new(v == Verbosity::Full).print_json()?;
114            process::exit(0);
115        }
116
117        if let Some(Commands::Pause {
118            base_path,
119            pod_id,
120            ipc,
121            pid,
122            net,
123            user,
124            uts,
125            uid_mappings,
126            gid_mappings,
127        }) = server.config().command()
128        {
129            Pause::run(
130                base_path,
131                pod_id,
132                *ipc,
133                *pid,
134                *net,
135                *user,
136                *uts,
137                uid_mappings,
138                gid_mappings,
139            )
140            .context("run pause")?;
141            process::exit(0);
142        }
143
144        server.config().validate().context("validate config")?;
145
146        Self::init().context("init self")?;
147        Ok(server)
148    }
149
150    /// Start the `Server` instance and consume it.
151    pub fn start(self) -> Result<()> {
152        // We need to fork as early as possible, especially before setting up tokio.
153        // If we don't, the child will have a strange thread space and we're at risk of deadlocking.
154        // We also have to treat the parent as the child (as described in [1]) to ensure we don't
155        // interrupt the child's execution.
156        // 1: https://docs.rs/nix/0.23.0/nix/unistd/fn.fork.html#safety
157        if !self.config().skip_fork() {
158            match unsafe { fork()? } {
159                ForkResult::Parent { child, .. } => {
160                    write!(File::create(self.config().conmon_pidfile())?, "{child}")?;
161                    unsafe { _exit(0) };
162                }
163                ForkResult::Child => (),
164            }
165        }
166
167        // now that we've forked, set self to childreaper
168        let ret = unsafe { libc::prctl(libc::PR_SET_CHILD_SUBREAPER, 1, 0, 0, 0) };
169        if ret != 0 {
170            return Err(Errno::last()).context("set child subreaper");
171        }
172
173        #[cfg(feature = "tracing")]
174        let tracer = self.tracer().clone();
175
176        debug!("Configuring Tokio runtime with current_thread");
177        let rt = Builder::new_current_thread()
178            .enable_io()
179            .enable_time()
180            .build()?;
181        rt.block_on(self.spawn_tasks())?;
182
183        #[cfg(feature = "tracing")]
184        if let Some(tracer) = tracer {
185            tracer.shutdown().context("shutdown tracer")?;
186        }
187
188        rt.shutdown_timeout(std::time::Duration::from_secs(15));
189        Ok(())
190    }
191
192    fn init() -> Result<()> {
193        let init = Init::<DefaultInit>::default();
194        init.unset_locale()?;
195        init.set_default_umask();
196        // While we could configure this, standard practice has it as -1000,
197        // so it may be YAGNI to add configuration.
198        init.set_oom_score("-1000")
199    }
200
201    fn init_logging(&mut self) -> Result<()> {
202        let level = LevelFilter::from_str(self.config().log_level().as_ref())
203            .context("convert log level filter")?;
204
205        #[cfg(feature = "tracing")]
206        let telemetry_layer = if self.config().enable_tracing() {
207            let tracer = Telemetry::layer(self.config().tracing_endpoint())
208                .context("build telemetry layer")?;
209
210            self.tracer = Some(tracer.clone());
211
212            tracing_opentelemetry::layer()
213                .with_tracer(tracer.tracer(crate_name!()))
214                .into()
215        } else {
216            None
217        };
218
219        let registry = tracing_subscriber::registry();
220        #[cfg(feature = "tracing")]
221        let registry = registry.with(telemetry_layer);
222
223        match self.config().log_driver() {
224            LogDriver::None => {}
225            LogDriver::Stdout => {
226                let layer = tracing_subscriber::fmt::layer()
227                    .with_target(true)
228                    .with_line_number(true)
229                    .with_filter(level);
230                registry
231                    .with(layer)
232                    .try_init()
233                    .context("init stdout fmt layer")?;
234                info!("Using stdout logger");
235            }
236            LogDriver::Systemd => {
237                let layer = tracing_subscriber::fmt::layer()
238                    .with_target(true)
239                    .with_line_number(true)
240                    .without_time()
241                    .with_writer(Journal)
242                    .with_filter(level);
243                registry
244                    .with(layer)
245                    .try_init()
246                    .context("init journald fmt layer")?;
247                info!("Using systemd/journald logger");
248            }
249        }
250        info!("Set log level to: {}", self.config().log_level());
251        Ok(())
252    }
253
254    /// Spawns all required tokio tasks.
255    async fn spawn_tasks(mut self) -> Result<()> {
256        self.init_logging().context("init logging")?;
257
258        let (shutdown_tx, shutdown_rx) = oneshot::channel();
259        let socket = self.config().socket();
260        let fd_socket = self.config().fd_socket();
261        let reaper = self.reaper.clone();
262
263        let signal_handler_span = debug_span!("signal_handler");
264        let backend_span = debug_span!("backend");
265
266        // Run both signal handler and backend inside spawn_blocking with LocalSet
267        // This allows concurrent execution within the LocalSet while preventing
268        // blocking of the main current_thread runtime
269        #[cfg(feature = "tracing")]
270        let result = task::spawn_blocking(move || {
271            Handle::current().block_on(async move {
272                let local = LocalSet::new();
273
274                // Spawn signal handler as a local task
275                local.spawn_local(
276                    Self::start_signal_handler(reaper, socket, fd_socket, shutdown_tx)
277                        .with_context(signal_handler_span.context())
278                        .instrument(signal_handler_span),
279                );
280
281                // Run backend on the LocalSet
282                local
283                    .run_until(
284                        self.start_backend(shutdown_rx)
285                            .with_context(backend_span.context())
286                            .instrument(backend_span),
287                    )
288                    .await
289            })
290        })
291        .await?;
292        #[cfg(not(feature = "tracing"))]
293        let result = task::spawn_blocking(move || {
294            Handle::current().block_on(async move {
295                let local = LocalSet::new();
296
297                // Spawn signal handler as a local task
298                local.spawn_local(
299                    Self::start_signal_handler(reaper, socket, fd_socket, shutdown_tx)
300                        .instrument(signal_handler_span),
301                );
302
303                // Run backend on the LocalSet
304                local
305                    .run_until(self.start_backend(shutdown_rx).instrument(backend_span))
306                    .await
307            })
308        })
309        .await?;
310        result
311    }
312
313    async fn start_signal_handler<T: AsRef<Path>>(
314        reaper: Arc<ChildReaper>,
315        socket: T,
316        fd_socket: T,
317        shutdown_tx: oneshot::Sender<()>,
318    ) -> Result<()> {
319        let mut sigterm = signal(SignalKind::terminate())?;
320        let mut sigint = signal(SignalKind::interrupt())?;
321
322        tokio::select! {
323            _ = sigterm.recv() => {
324                info!("Received SIGTERM");
325            }
326            _ = sigint.recv() => {
327                info!("Received SIGINT");
328            }
329        }
330
331        if let Some(pause) = Pause::maybe_shared() {
332            pause.stop();
333        }
334
335        debug!("Starting grandchildren cleanup task");
336        // Always use SIGKILL to ensure immediate termination of container processes
337        reaper
338            .kill_grandchildren(Signal::SIGKILL)
339            .await
340            .context("unable to kill grandchildren")?;
341
342        debug!("Sending shutdown message");
343        shutdown_tx
344            .send(())
345            .map_err(|_| format_err!("unable to send shutdown message"))?;
346
347        debug!("Removing socket file {}", socket.as_ref().display());
348        fs::remove_file(socket)
349            .await
350            .context("remove existing socket file")?;
351
352        debug!("Removing fd socket file {}", fd_socket.as_ref().display());
353        fs::remove_file(fd_socket)
354            .await
355            .or_else(|err| {
356                if err.kind() == std::io::ErrorKind::NotFound {
357                    Ok(())
358                } else {
359                    Err(err)
360                }
361            })
362            .context("remove existing fd socket file")
363    }
364
365    async fn start_backend(self, mut shutdown_rx: oneshot::Receiver<()>) -> Result<()> {
366        let listener =
367            Listener::<DefaultListener>::default().bind_long_path(self.config().socket())?;
368        let client: conmon::Client = capnp_rpc::new_client(self);
369
370        loop {
371            let stream = tokio::select! {
372                _ = &mut shutdown_rx => {
373                    debug!("Received shutdown message");
374                    return Ok(())
375                }
376                stream = listener.accept() => {
377                    stream?.0
378                },
379            };
380            let (reader, writer) = TokioAsyncReadCompatExt::compat(stream).split();
381            let network = Box::new(VatNetwork::new(
382                reader,
383                writer,
384                Side::Server,
385                Default::default(),
386            ));
387            let rpc_system = RpcSystem::new(network, Some(client.clone().client));
388            task::spawn_local(Box::pin(rpc_system.map(|_| ())));
389        }
390    }
391}
392
393pub(crate) struct GenerateRuntimeArgs<'a> {
394    pub(crate) config: &'a Config,
395    pub(crate) id: &'a str,
396    pub(crate) container_io: &'a ContainerIO,
397    pub(crate) pidfile: &'a Path,
398    pub(crate) cgroup_manager: CgroupManager,
399}
400
401impl GenerateRuntimeArgs<'_> {
402    const SYSTEMD_CGROUP_ARG: &'static str = "--systemd-cgroup";
403    const RUNTIME_CRUN: &'static str = "crun";
404    const LOG_LEVEL_FLAG_CRUN: &'static str = "--log-level";
405
406    /// Generate the OCI runtime CLI arguments from the provided parameters.
407    pub fn create_args(
408        self,
409        bundle_path: &Path,
410        global_args: Reader,
411        command_args: Reader,
412    ) -> Result<Vec<String>> {
413        // Pre-allocate capacity for typical arg count to reduce reallocations
414        let mut args = Vec::with_capacity(16);
415        args.extend(self.default_args().context("build default runtime args")?);
416
417        if let Some(rr) = self.config.runtime_root() {
418            args.push(format!("--root={}", rr.display()));
419        }
420
421        if self.cgroup_manager == CgroupManager::Systemd {
422            args.push(Self::SYSTEMD_CGROUP_ARG.into());
423        }
424
425        for arg in global_args {
426            args.push(arg?.to_string()?);
427        }
428
429        // Use static strings where possible to avoid allocations
430        args.push("create".into());
431        args.push("--bundle".into());
432        args.push(bundle_path.display().to_string());
433        args.push("--pid-file".into());
434        args.push(self.pidfile.display().to_string());
435
436        for arg in command_args {
437            args.push(arg?.to_string()?);
438        }
439
440        if let ContainerIOType::Terminal(terminal) = self.container_io.typ() {
441            args.push(format!("--console-socket={}", terminal.path().display()));
442        }
443
444        args.push(self.id.into());
445
446        debug!("Runtime args {:?}", args.join(" "));
447        Ok(args)
448    }
449
450    /// Generate the OCI runtime CLI arguments from the provided parameters.
451    pub(crate) fn exec_sync_args(&self, command: Reader) -> Result<Vec<String>> {
452        let mut args = self
453            .exec_sync_args_without_command()
454            .context("exec sync args without command")?;
455
456        for arg in command {
457            args.push(arg?.to_string()?);
458        }
459
460        debug!("Exec args {:?}", args.join(" "));
461        Ok(args)
462    }
463
464    pub(crate) fn exec_sync_args_without_command(&self) -> Result<Vec<String>> {
465        // Pre-allocate capacity for typical arg count
466        let mut args = Vec::with_capacity(12);
467        args.extend(self.default_args().context("build default runtime args")?);
468
469        if let Some(rr) = self.config.runtime_root() {
470            args.push(format!("--root={}", rr.display()));
471        }
472
473        if self.cgroup_manager == CgroupManager::Systemd {
474            args.push(Self::SYSTEMD_CGROUP_ARG.into());
475        }
476
477        // Use static strings to avoid allocations
478        args.push("exec".into());
479        args.push("-d".into());
480
481        if let ContainerIOType::Terminal(terminal) = self.container_io.typ() {
482            args.push(format!("--console-socket={}", terminal.path().display()));
483            args.push("--tty".into());
484        }
485
486        args.push(format!("--pid-file={}", self.pidfile.display()));
487        args.push(self.id.into());
488
489        Ok(args)
490    }
491
492    /// Build the default arguments for any provided runtime.
493    fn default_args(&self) -> Result<Vec<String>> {
494        let mut args = vec![];
495
496        if self
497            .config
498            .runtime()
499            .file_name()
500            .context("no filename in path")?
501            == Self::RUNTIME_CRUN
502        {
503            debug!("Found crun used as runtime");
504            args.push(format!("--log=journald:{}", self.id));
505
506            match self.config.log_level() {
507                &LogLevel::Debug | &LogLevel::Error => args.push(format!(
508                    "{}={}",
509                    Self::LOG_LEVEL_FLAG_CRUN,
510                    self.config.log_level()
511                )),
512                &LogLevel::Warn => args.push(format!("{}=warning", Self::LOG_LEVEL_FLAG_CRUN)),
513                _ => {}
514            }
515        }
516
517        if let Some(rr) = self.config.runtime_root() {
518            args.push(format!("--root={}", rr.display()));
519        }
520
521        if self.cgroup_manager == CgroupManager::Systemd {
522            args.push(Self::SYSTEMD_CGROUP_ARG.into());
523        }
524
525        Ok(args)
526    }
527}
conmonrs/server.rs

conmonrs/
server.rs
