conmonrs/
container_log.rs

1//! The log drivers of the containers, which write the container output to
2//! the CRI log files or to journald.
3
4mod cri;
5mod journald;
6mod line_buffer;
7mod log_file;
8#[cfg(test)]
9pub(crate) mod testing;
10
11use crate::conmon_capnp::conmon::log_driver::{self, Owned, Type};
12use crate::{
13    container_io::Pipe,
14    container_log::{
15        cri::CriLogger, journald::JournaldLogger, line_buffer::LineBuffer, log_file::LogFile,
16    },
17    fs_util::check_absolute,
18    server::Server,
19};
20use anyhow::{Context, Result, bail};
21use bytes::Bytes;
22use capnp::struct_list::Reader;
23use std::{io, ops::Deref, path::Path, sync::Arc};
24use tokio::{
25    sync::{Mutex, Semaphore},
26    task,
27};
28
29/// A container log shared by its users, like the read loops of the stdio
30/// and the log reopening.
31pub(crate) type SharedContainerLog = Arc<Mutex<ContainerLog>>;
32
33/// A configured log driver.
34///
35/// A driver either takes the raw chunks as they are read, like the CRI log
36/// which writes partial lines right away, or whole lines which
37/// [`ContainerLog`] collects across chunks once for all such drivers.
38#[derive(Debug)]
39enum Driver {
40    Cri(CriLogger),
41    Journald(JournaldLogger),
42}
43
44impl Driver {
45    /// Whether the driver takes whole lines instead of raw chunks.
46    fn buffers_lines(&self) -> bool {
47        match self {
48            Driver::Cri(_) => false,
49            Driver::Journald(_) => true,
50        }
51    }
52
53    /// The log file of the driver, none for one which writes elsewhere.
54    fn file(&mut self) -> Option<&mut LogFile> {
55        match self {
56            Driver::Cri(logger) => Some(logger.file()),
57            Driver::Journald(_) => None,
58        }
59    }
60
61    /// Prepare the driver, open its file.
62    async fn init(&mut self) -> Result<()> {
63        match self.file() {
64            Some(file) => file.open().await,
65            None => Ok(()),
66        }
67    }
68
69    /// Start over with a new file for log rotation.
70    async fn reopen(&mut self) -> Result<()> {
71        match self.file() {
72            Some(file) => file.reopen().await,
73            None => Ok(()),
74        }
75    }
76
77    /// Write a raw chunk of output of the given pipe, if the driver takes
78    /// chunks.
79    async fn write_chunk(&mut self, pipe: Pipe, bytes: &[u8]) -> Result<()> {
80        match self {
81            Driver::Cri(logger) => logger.write_chunk(pipe, bytes).await,
82            Driver::Journald(_) => Ok(()),
83        }
84    }
85
86    /// Write whole lines of output of the given pipe, if the driver takes
87    /// lines.
88    async fn write_lines(&mut self, pipe: Pipe, lines: &[Line<'_>]) -> Result<()> {
89        match self {
90            Driver::Cri(_) => Ok(()),
91            Driver::Journald(logger) => logger.write_lines(pipe, lines).await,
92        }
93    }
94}
95
96/// The log drivers of a container.
97#[derive(Debug, Default)]
98pub(crate) struct ContainerLog {
99    drivers: Vec<Driver>,
100
101    /// Collects whole lines for the journald drivers, the only ones which
102    /// take them.
103    lines: LineBuffer,
104
105    /// Whether any driver takes whole lines.
106    buffers_lines: bool,
107}
108
109impl ContainerLog {
110    /// Create a new default SharedContainerLog.
111    pub(crate) fn new() -> SharedContainerLog {
112        Arc::new(Mutex::new(Self::default()))
113    }
114
115    /// Check if this logger has any drivers configured.
116    pub(crate) fn has_drivers(&self) -> bool {
117        !self.drivers.is_empty()
118    }
119
120    /// Create a new SharedContainerLog for the container `id` from the
121    /// requested drivers.
122    pub(crate) fn from_request(id: &str, reader: Reader<Owned>) -> Result<SharedContainerLog> {
123        let drivers = reader
124            .iter()
125            .map(|driver| -> Result<Driver> {
126                let max_size = Self::max_size(&driver);
127                Ok(match driver.get_type()? {
128                    Type::ContainerRuntimeInterface => {
129                        Driver::Cri(CriLogger::new(Self::path(&driver)?, max_size))
130                    }
131                    // Retired, `createContainer` rejects it with the other
132                    // retired fields before, see `reject_retired`.
133                    Type::JsonOld => {
134                        debug_assert!(false, "the retired log driver type got through");
135                        bail!("unsupported log driver type")
136                    }
137                    Type::Journald => Driver::Journald(JournaldLogger::new(
138                        id,
139                        driver.get_container_name()?.to_str()?,
140                    )),
141                })
142            })
143            .collect::<Result<Vec<_>>>()?;
144        let buffers_lines = drivers.iter().any(Driver::buffers_lines);
145        Ok(Arc::new(Mutex::new(Self {
146            drivers,
147            lines: LineBuffer::default(),
148            buffers_lines,
149        })))
150    }
151
152    /// The path of a requested driver, which has to be absolute.
153    fn path<'a>(driver: &log_driver::Reader<'a>) -> Result<&'a Path> {
154        let path = Path::new(driver.get_path()?.to_str()?);
155        check_absolute("log path", path)?;
156        Ok(path)
157    }
158
159    /// The maximum log size of a requested driver, zero means unlimited.
160    fn max_size(driver: &log_driver::Reader<'_>) -> Option<usize> {
161        match driver.get_max_size() {
162            0 => None,
163            // More than the address space is no limit either.
164            max => usize::try_from(max).ok(),
165        }
166    }
167
168    /// Asynchronously initialize all loggers.
169    pub(crate) async fn init(&mut self) -> Result<()> {
170        for driver in &mut self.drivers {
171            driver.init().await?;
172        }
173        Ok(())
174    }
175
176    /// Reopen the container logs. All drivers get reopened even if one
177    /// fails, the first error is returned.
178    pub(crate) async fn reopen(&mut self) -> Result<()> {
179        let mut result = Ok(());
180        for driver in &mut self.drivers {
181            result = result.and(driver.reopen().await);
182        }
183        result
184    }
185
186    /// Write the contents of the provided bytes into all loggers. All of
187    /// them get the chunk even if one fails, the first error is returned.
188    pub(crate) async fn write(&mut self, pipe: Pipe, bytes: &[u8]) -> Result<()> {
189        let mut result = Ok(());
190        for driver in &mut self.drivers {
191            result = result.and(driver.write_chunk(pipe, bytes).await);
192        }
193        if self.buffers_lines {
194            let lines = self.lines.push(pipe, bytes);
195            let res = self.write_lines(pipe, &lines).await;
196            result = result.and(res);
197        }
198        result
199    }
200
201    /// Whether a line of the pipe waits for its ending.
202    pub(crate) fn has_pending(&self, pipe: Pipe) -> bool {
203        self.buffers_lines && self.lines.has_pending(pipe)
204    }
205
206    /// Write the line of the pipe which waits for its ending as a partial
207    /// one, because the output paused.
208    pub(crate) async fn flush(&mut self, pipe: Pipe) -> Result<()> {
209        let lines = self.lines.flush(pipe);
210        self.write_lines(pipe, &lines).await
211    }
212
213    /// The output of the pipe ended, write the line which still waits for
214    /// its ending.
215    pub(crate) async fn finish(&mut self, pipe: Pipe) -> Result<()> {
216        let lines = self.lines.finish(pipe);
217        self.write_lines(pipe, &lines).await
218    }
219
220    /// Sync the data of the log files to disk, like conmon does once the
221    /// container exited. The log is not locked while the files sync, they
222    /// get synced even if one fails, the first error is returned. A file
223    /// which cannot be synced, like a device, has nothing to sync. Only a
224    /// few logs sync at a time, syncs which hang on a stuck disk keep a
225    /// quarter of the threads of the blocking pool at most, see
226    /// [`Server::MAX_BLOCKING_THREADS`]. Tests, which share the static, do
227    /// not wait for each other.
228    pub(crate) async fn sync(log: &SharedContainerLog) -> Result<()> {
229        const PERMITS: usize = if cfg!(test) {
230            Semaphore::MAX_PERMITS
231        } else {
232            Server::MAX_BLOCKING_THREADS / 4
233        };
234        static SYNC: Semaphore = Semaphore::const_new(PERMITS);
235        let files: Vec<_> = log
236            .lock()
237            .await
238            .drivers
239            .iter_mut()
240            .filter_map(|driver| driver.file()?.file.clone())
241            .collect();
242        if files.is_empty() {
243            return Ok(());
244        }
245        let permit = SYNC.acquire().await.context("wait for log sync")?;
246        task::spawn_blocking(move || {
247            let _permit = permit;
248            files
249                .iter()
250                .map(|file| match file.sync_data() {
251                    Err(e) if e.raw_os_error() == Some(libc::EINVAL) => Ok(()),
252                    res => res,
253                })
254                .fold(Ok(()), io::Result::and)
255        })
256        .await
257        .context("join log sync")?
258        .context("sync log file")
259    }
260
261    /// Hand the lines to every driver which takes them. All of them get the
262    /// lines even if one fails, the first error is returned.
263    async fn write_lines(&mut self, pipe: Pipe, lines: &[Line<'_>]) -> Result<()> {
264        if lines.is_empty() {
265            return Ok(());
266        }
267        let mut result = Ok(());
268        for driver in self.drivers.iter_mut().filter(|d| d.buffers_lines()) {
269            result = result.and(driver.write_lines(pipe, lines).await);
270        }
271        result
272    }
273}
274
275/// Split a chunk into its lines. Every line but the last ends with a newline,
276/// the last one is partial if the chunk does not end with one. An empty chunk
277/// has no lines.
278fn lines(bytes: &[u8]) -> impl Iterator<Item = (&[u8], bool)> {
279    let mut ends = memchr::memchr_iter(b'\n', bytes);
280    let mut start = 0;
281    std::iter::from_fn(move || {
282        let (end, partial) = match ends.next() {
283            Some(end) => (end + 1, false),
284            None if start < bytes.len() => (bytes.len(), true),
285            None => return None,
286        };
287        let line = &bytes[start..end];
288        start = end;
289        Some((line, partial))
290    })
291}
292
293/// Strip the line ending, which a terminal turns into "\r\n".
294fn strip_line_ending(line: &[u8]) -> &[u8] {
295    line.strip_suffix(b"\r\n")
296        .or_else(|| line.strip_suffix(b"\n"))
297        .unwrap_or(line)
298}
299
300/// A whole or partial line without its ending. Partial means that the line
301/// had no ending yet: it continues in the next entry, paused, or the output
302/// ended without one, the same as the `P` marker of the CRI log.
303#[derive(Clone, Debug, PartialEq, Eq)]
304struct Line<'a> {
305    text: Text<'a>,
306    partial: bool,
307}
308
309impl Line<'_> {
310    /// The line with a text of its own, for a task which outlives the chunk.
311    fn into_owned(self) -> Line<'static> {
312        Line {
313            text: self.text.into_owned(),
314            partial: self.partial,
315        }
316    }
317}
318
319/// The text of a [`Line`]: borrowed from the chunk which held all of it, or
320/// collected across chunks in the buffer of its pipe.
321#[derive(Clone, Debug, PartialEq, Eq)]
322enum Text<'a> {
323    Chunk(&'a [u8]),
324    Buffered(Bytes),
325}
326
327impl Text<'_> {
328    fn into_owned(self) -> Text<'static> {
329        match self {
330            Text::Chunk(text) => Text::Buffered(Bytes::copy_from_slice(text)),
331            Text::Buffered(text) => Text::Buffered(text),
332        }
333    }
334}
335
336impl Deref for Text<'_> {
337    type Target = [u8];
338
339    fn deref(&self) -> &[u8] {
340        match self {
341            Text::Chunk(text) => text,
342            Text::Buffered(text) => text,
343        }
344    }
345}
346
347/// The entry points of the fuzz targets, see `crate::fuzzing`.
348#[cfg(any(test, feature = "fuzzing"))]
349pub(crate) mod fuzzing;
350
351#[cfg(test)]
352mod tests;
conmonrs/container_log.rs

conmonrs/
container_log.rs
