conmonrs/container_log/journald.rs
1//! The logger which writes the container output to journald.
2
3use crate::{
4 container_io::Pipe,
5 container_log::Line,
6 journal::{self, Entry, Priority, Wait},
7 mutex::lock,
8};
9use anyhow::{Context, Result};
10use std::{
11 io::{self, ErrorKind},
12 path::Path,
13 sync::{
14 Arc, Mutex,
15 atomic::{AtomicBool, Ordering},
16 },
17 time::{Duration, Instant},
18};
19use tokio::task;
20use tracing::{Span, debug, info, warn};
21
22/// Writes the container output to the journal with the container ID fields
23/// and priorities conmon uses, so that `journalctl CONTAINER_ID=<id>` finds
24/// it and podman can put partial lines back together.
25#[derive(Debug)]
26pub(super) struct JournaldLogger {
27 /// Shared with the blocking tasks which send the entries.
28 shared: Arc<Shared>,
29}
30
31#[derive(Debug)]
32struct Shared {
33 /// The fields of every entry, encoded once.
34 fields: Entry,
35 /// Whether the last send failed, so that a journal which is unavailable
36 /// gets reported once and not for every chunk of output.
37 failing: AtomicBool,
38 /// When the lines which last waited for the journal did not go out in
39 /// time. The following chunks drop what the journal does not take right
40 /// away instead of waiting as well, for at least another
41 /// `CHUNK_TIMEOUT` and after that as long as it takes nothing, like a
42 /// stalled one. Waiting for every chunk would slow down the reading of
43 /// the output to about one per `CHUNK_TIMEOUT`, which the other log
44 /// drivers and the container, blocked on a full pipe, would wait for.
45 /// This way a journal which does not keep up holds it back by about half
46 /// at most, and one which stalled not at all.
47 timed_out: Mutex<Option<Instant>>,
48 /// A temporary socket instead of the one of journald.
49 #[cfg(test)]
50 socket: Option<std::path::PathBuf>,
51}
52
53impl Shared {
54 /// Whether the lines which the journal did not take right away get
55 /// dropped instead of waiting for it, after `sent` lines went out
56 /// without waiting, see [`Shared::timed_out`]. Chunks which went out as
57 /// a whole ask as well: they end the back-off like a partly taken one,
58 /// a journal which recovered would get no bounded wait the next time it
59 /// is busy otherwise.
60 fn skip_wait(&self, sent: usize) -> bool {
61 let mut timed_out = lock(&self.timed_out);
62 match *timed_out {
63 Some(at) if sent == 0 || at.elapsed() < CHUNK_TIMEOUT => true,
64 _ => {
65 *timed_out = None;
66 false
67 }
68 }
69 }
70
71 fn socket(&self) -> &Path {
72 #[cfg(test)]
73 if let Some(socket) = &self.socket {
74 return socket;
75 }
76 Path::new(journal::SOCKET_PATH)
77 }
78}
79
80impl JournaldLogger {
81 /// The length of the short container ID field.
82 const SHORT_ID_LEN: usize = 12;
83
84 /// The field marking a line which had no ending yet.
85 const PARTIAL_FIELD: (&'static str, &'static str) = ("CONTAINER_PARTIAL_MESSAGE", "true");
86
87 /// A logger with the fields conmon writes: the container ID, the name
88 /// if given and the syslog identifier, which is the name, or the short
89 /// container ID without one, like for conmon without a tag.
90 pub(super) fn new(id: &str, name: &str) -> Self {
91 let short_id: String = id.chars().take(Self::SHORT_ID_LEN).collect();
92 let mut fields = Entry::default();
93 fields
94 .field("CONTAINER_ID_FULL", id)
95 .field("CONTAINER_ID", &short_id);
96 if !name.is_empty() {
97 fields.field("CONTAINER_NAME", name);
98 }
99 let identifier = [name, &short_id]
100 .into_iter()
101 .find(|s| !s.is_empty())
102 .unwrap_or(journal::IDENTIFIER);
103 fields.field("SYSLOG_IDENTIFIER", identifier);
104 Self {
105 shared: Arc::new(Shared {
106 fields,
107 failing: AtomicBool::new(false),
108 timed_out: Mutex::new(None),
109 #[cfg(test)]
110 socket: None,
111 }),
112 }
113 }
114
115 /// A logger which sends to the `socket` instead of the one of journald.
116 #[cfg(test)]
117 pub(super) fn with_socket(id: &str, socket: &Path) -> Self {
118 let mut logger = Self::new(id, "");
119 Arc::get_mut(&mut logger.shared)
120 .expect("unshared logger")
121 .socket = Some(socket.to_owned());
122 logger
123 }
124
125 /// The journal priority of the output of a pipe.
126 fn priority(pipe: Pipe) -> Priority {
127 match pipe {
128 Pipe::StdOut => Priority::Info,
129 Pipe::StdErr => Priority::Error,
130 }
131 }
132
133 /// Send the lines to the journal and return how many of them it took
134 /// or got dropped. A journal which is unavailable, for example while it
135 /// restarts, must not end the output of the container, so failures are
136 /// only logged, once until a send succeeds again. Without a journal, or
137 /// with one which did not take a line in time, the rest of the chunk is
138 /// dropped right away instead of failing the same way line by line.
139 /// Without waiting, the sending stops at the first line which the
140 /// journal does not take right away. Waiting, the lines which did not
141 /// go out by the deadline of the wait get dropped, a journal which takes
142 /// each line just in time would block the log of the container for
143 /// minutes per chunk otherwise.
144 fn send(shared: &Shared, pipe: Pipe, lines: &[Line<'_>], wait: Wait) -> usize {
145 let priority = Self::priority(pipe);
146 let mut entry = Entry::default();
147 let mut handled = lines.len();
148 let mut failed = 0;
149 let mut last_error = None;
150 for (i, line) in lines.iter().enumerate() {
151 if let Wait::Bounded(deadline) = wait
152 && Instant::now() >= deadline
153 {
154 failed += lines.len() - i;
155 *lock(&shared.timed_out) = Some(Instant::now());
156 last_error = Some(io::Error::new(
157 ErrorKind::TimedOut,
158 format!("journal did not take the chunk within {CHUNK_TIMEOUT:?}"),
159 ));
160 break;
161 }
162 entry.reset(priority, &*line.text).extend(&shared.fields);
163 if line.partial {
164 entry.field(Self::PARTIAL_FIELD.0, Self::PARTIAL_FIELD.1);
165 }
166 let Err(e) = journal::send_at(wait, shared.socket(), &entry) else {
167 continue;
168 };
169 if wait == Wait::Never && e.kind() == ErrorKind::WouldBlock {
170 handled = i;
171 break;
172 }
173 let timed_out = e.kind() == ErrorKind::TimedOut;
174 let rest = timed_out || journal::unavailable(&e);
175 last_error = Some(e);
176 if timed_out {
177 *lock(&shared.timed_out) = Some(Instant::now());
178 }
179 if rest {
180 failed += lines.len() - i;
181 break;
182 }
183 failed += 1;
184 }
185 match last_error {
186 Some(e) if !shared.failing.swap(true, Ordering::Relaxed) => warn!(
187 "Unable to write {failed} lines to the journal, dropping lines \
188 until it takes them again: {e}"
189 ),
190 Some(e) => debug!("Unable to write {failed} lines to the journal: {e}"),
191 // Only once the whole chunk went out, not some lines before the
192 // journal stalled again.
193 None if handled == lines.len() && shared.failing.swap(false, Ordering::Relaxed) => {
194 info!("The journal takes lines again");
195 }
196 None => {}
197 }
198 handled
199 }
200}
201
202/// How long the lines of a chunk which the journal did not take right away
203/// may wait for it in total. A line which started before it may take one
204/// more `journal::SEND_TIMEOUT`, sends which a signal interrupted get
205/// retried only until it.
206const CHUNK_TIMEOUT: Duration = Duration::from_secs(1);
207
208impl JournaldLogger {
209 /// Write whole lines of output of the pipe, one entry each. A journal
210 /// which keeps up takes the lines right away. The ones it does not take
211 /// are sent from a single blocking task, which waits for it a bounded
212 /// time, instead of one per line, unless the last wait timed out, see
213 /// [`Shared::timed_out`]. The lines which borrow the chunk get copied for
214 /// it, the others share their buffer.
215 pub(super) async fn write_lines(&self, pipe: Pipe, lines: &[Line<'_>]) -> Result<()> {
216 let sent = Self::send(&self.shared, pipe, lines, Wait::Never);
217 let skip_wait = self.shared.skip_wait(sent);
218 let Some(rest) = lines.get(sent..).filter(|rest| !rest.is_empty()) else {
219 return Ok(());
220 };
221 if skip_wait {
222 // The timeout got logged already, see `Shared::failing`.
223 debug!(
224 "Unable to write {} lines to the journal: it did not take the last ones in time",
225 rest.len()
226 );
227 return Ok(());
228 }
229 let shared = self.shared.clone();
230 let rest: Vec<_> = rest.iter().cloned().map(Line::into_owned).collect();
231 // The thread has no span, the one of the read loop names the
232 // container in what the send logs.
233 let span = Span::current();
234 // From when a thread of the blocking pool took it, a busy pool must
235 // not drop the chunk without trying.
236 task::spawn_blocking(move || {
237 let _entered = span.entered();
238 let deadline = Instant::now() + CHUNK_TIMEOUT;
239 Self::send(&shared, pipe, &rest, Wait::Bounded(deadline))
240 })
241 .await
242 .map(drop)
243 .context("journal spawn_blocking")
244 }
245}
246
247#[cfg(test)]
248mod tests;