archivist_lib/logs/
serial.rs

1// Copyright 2023 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5use crate::logs::repository::LogsRepository;
6use anyhow::Error;
7use diagnostics_data::{Data, Logs};
8use fidl_fuchsia_diagnostics::{Selector, StreamMode};
9use fuchsia_trace as ftrace;
10use futures::StreamExt;
11use log::warn;
12use selectors::FastError;
13use std::borrow::Cow;
14use std::collections::HashSet;
15use std::fmt::Display;
16use std::io::{self, Write};
17use std::sync::Arc;
18
19const MAX_SERIAL_WRITE_SIZE: usize = 256;
20
21#[derive(Default)]
22pub struct SerialConfig {
23    selectors: Vec<Selector>,
24    denied_tags: HashSet<String>,
25}
26
27impl SerialConfig {
28    /// Creates a new serial configuration from the given structured config values.
29    pub fn new<C, T>(allowed_components: Vec<C>, denied_tags: Vec<T>) -> Self
30    where
31        C: AsRef<str> + Display,
32        T: Into<String>,
33    {
34        let selectors = allowed_components
35            .into_iter()
36            .filter_map(|selector| {
37                match selectors::parse_component_selector::<FastError>(selector.as_ref()) {
38                    Ok(s) => Some(Selector {
39                        component_selector: Some(s),
40                        tree_selector: None,
41                        ..Selector::default()
42                    }),
43                    Err(err) => {
44                        warn!(selector:%, err:?; "Failed to parse component selector");
45                        None
46                    }
47                }
48            })
49            .collect();
50        Self { selectors, denied_tags: HashSet::from_iter(denied_tags.into_iter().map(Into::into)) }
51    }
52
53    /// Returns a future that resolves when there's no more logs to write to serial. This can only
54    /// happen when all log sink connections have been closed for the components that were
55    /// configured to emit logs.
56    pub async fn write_logs<S: Write>(self, repo: Arc<LogsRepository>, mut sink: S) {
57        let Self { denied_tags, selectors } = self;
58        let mut log_stream = repo.logs_cursor(
59            StreamMode::SnapshotThenSubscribe,
60            Some(selectors),
61            ftrace::Id::random(),
62        );
63        while let Some(log) = log_stream.next().await {
64            SerialWriter::log(log.as_ref(), &denied_tags, &mut sink).ok();
65        }
66    }
67}
68
69/// A sink to write to serial. This Write implementation must be used together with SerialWriter.
70#[derive(Default)]
71pub struct SerialSink;
72
73impl Write for SerialSink {
74    fn write(&mut self, buffer: &[u8]) -> io::Result<usize> {
75        if cfg!(debug_assertions) {
76            debug_assert!(buffer.len() <= MAX_SERIAL_WRITE_SIZE);
77        } else {
78            use std::sync::atomic::{AtomicBool, Ordering};
79            static ALREADY_LOGGED: AtomicBool = AtomicBool::new(false);
80            if buffer.len() > MAX_SERIAL_WRITE_SIZE && !ALREADY_LOGGED.swap(true, Ordering::Relaxed)
81            {
82                let size = buffer.len();
83                log::error!(
84                    size;
85                    "Skipping write to serial due to internal error. Exceeded max buffer size."
86                );
87                return Ok(buffer.len());
88            }
89        }
90        // SAFETY: calling a syscall. We pass a pointer to the buffer and its exact size.
91        unsafe {
92            zx::sys::zx_debug_write(buffer.as_ptr(), buffer.len());
93        }
94        Ok(buffer.len())
95    }
96
97    fn flush(&mut self) -> io::Result<()> {
98        Ok(())
99    }
100}
101
102struct SerialWriter<'a, S> {
103    buffer: Vec<u8>,
104    denied_tags: &'a HashSet<String>,
105    sink: &'a mut S,
106}
107
108impl<S: Write> Write for SerialWriter<'_, S> {
109    fn write(&mut self, data: &[u8]) -> io::Result<usize> {
110        // -1 since we always write a `\n` when flushing.
111        let count = (self.buffer.capacity() - self.buffer.len() - 1).min(data.len());
112        let actual_count = self.buffer.write(&data[..count])?;
113        debug_assert_eq!(actual_count, count);
114        if self.buffer.len() == self.buffer.capacity() - 1 {
115            self.flush()?;
116        }
117        Ok(actual_count)
118    }
119
120    fn flush(&mut self) -> io::Result<()> {
121        debug_assert!(self.buffer.len() < MAX_SERIAL_WRITE_SIZE);
122        let wrote = self.buffer.write(b"\n")?;
123        debug_assert_eq!(wrote, 1);
124        self.sink.write_all(self.buffer.as_slice())?;
125        self.buffer.clear();
126        Ok(())
127    }
128}
129
130impl<'a, S: Write> SerialWriter<'a, S> {
131    fn log(
132        log: &Data<Logs>,
133        denied_tags: &'a HashSet<String>,
134        sink: &'a mut S,
135    ) -> Result<(), Error> {
136        let mut this =
137            Self { buffer: Vec::with_capacity(MAX_SERIAL_WRITE_SIZE), sink, denied_tags };
138        write!(
139            &mut this,
140            "[{:05}.{:03}] {:05}:{:05}> [",
141            zx::MonotonicDuration::from_nanos(log.metadata.timestamp.into_nanos()).into_seconds(),
142            zx::MonotonicDuration::from_nanos(log.metadata.timestamp.into_nanos()).into_millis()
143                % 1000,
144            log.pid().unwrap_or(0),
145            log.tid().unwrap_or(0)
146        )?;
147
148        let empty_tags = log.tags().map(|tags| tags.is_empty()).unwrap_or(true);
149        if empty_tags {
150            write!(&mut this, "{}", log.component_name())?;
151        } else {
152            // Unwrap is safe, if we are here it means that we actually have tags.
153            let tags = log.tags().unwrap();
154            for (i, tag) in tags.iter().enumerate() {
155                if this.denied_tags.contains(tag) {
156                    return Ok(());
157                }
158                write!(&mut this, "{tag}")?;
159                if i < tags.len() - 1 {
160                    write!(&mut this, ", ")?;
161                }
162            }
163        }
164
165        write!(&mut this, "] {}: ", log.severity())?;
166        let mut pending_message_parts = [Cow::Borrowed(log.msg().unwrap_or(""))]
167            .into_iter()
168            .chain(log.payload_keys_strings().map(|s| Cow::Owned(format!(" {s}"))));
169        let mut pending_str = None;
170
171        loop {
172            let (data, offset) = match pending_str.take() {
173                Some((s, offset)) => (s, offset),
174                None => match pending_message_parts.next() {
175                    Some(s) => (s, 0),
176                    None => break,
177                },
178            };
179            let count = this.write(&data.as_bytes()[offset..])?;
180            if offset + count < data.len() {
181                pending_str = Some((data, offset + count));
182            }
183        }
184        if !this.buffer.is_empty() {
185            this.flush()?;
186        }
187        Ok(())
188    }
189}
190
191#[cfg(test)]
192mod tests {
193    use std::task::Poll;
194
195    use super::*;
196    use crate::identity::ComponentIdentity;
197    use crate::logs::testing::make_message;
198    use diagnostics_data::{BuilderArgs, LogsDataBuilder, LogsField, LogsProperty, Severity};
199    use fuchsia_async as fasync;
200    use futures::channel::mpsc;
201    use futures::future::poll_fn;
202    use futures::FutureExt;
203    use moniker::ExtendedMoniker;
204    use std::pin::pin;
205    use zx::BootInstant;
206
207    struct TestSink {
208        snd: mpsc::UnboundedSender<String>,
209    }
210
211    impl TestSink {
212        fn new() -> (Self, mpsc::UnboundedReceiver<String>) {
213            let (snd, rcv) = mpsc::unbounded();
214            (Self { snd }, rcv)
215        }
216    }
217
218    impl Write for TestSink {
219        fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
220            let string = String::from_utf8(buf.to_vec()).expect("wrote valid utf8");
221            self.snd.unbounded_send(string).expect("sent item");
222            Ok(buf.len())
223        }
224
225        fn flush(&mut self) -> io::Result<()> {
226            Ok(())
227        }
228    }
229
230    #[fuchsia::test]
231    fn write_to_serial_handles_denied_tags() {
232        let log = LogsDataBuilder::new(BuilderArgs {
233            timestamp: BootInstant::from_nanos(1),
234            component_url: Some("url".into()),
235            moniker: "core/foo".try_into().unwrap(),
236            severity: Severity::Info,
237        })
238        .add_tag("denied-tag")
239        .build();
240        let denied_tags = HashSet::from_iter(["denied-tag".to_string()]);
241        let mut sink = Vec::new();
242        SerialWriter::log(&log, &denied_tags, &mut sink).expect("write succeeded");
243        assert!(sink.is_empty());
244    }
245
246    #[fuchsia::test]
247    fn write_to_serial_splits_lines() {
248        let message = concat!(
249            "Lorem ipsum dolor sit amet, consectetur adipiscing elit. Aliquam accumsan eu neque ",
250            "quis molestie. Nam rhoncus sapien non eleifend tristique. Duis quis turpis volutpat ",
251            "neque bibendum molestie. Etiam ac sapien justo. Nullam aliquet ipsum nec tincidunt."
252        );
253        let log = LogsDataBuilder::new(BuilderArgs {
254            timestamp: BootInstant::from_nanos(123456789),
255            component_url: Some("url".into()),
256            moniker: "core/foo".try_into().unwrap(),
257            severity: Severity::Info,
258        })
259        .add_tag("bar")
260        .set_message(message)
261        .add_key(LogsProperty::String(LogsField::Other("key".to_string()), "value".to_string()))
262        .add_key(LogsProperty::Int(LogsField::Other("other_key".to_string()), 3))
263        .set_pid(1234)
264        .set_tid(5678)
265        .build();
266        let mut sink = Vec::new();
267        SerialWriter::log(&log, &HashSet::new(), &mut sink).expect("write succeeded");
268        assert_eq!(
269            String::from_utf8(sink).unwrap(),
270            format!(
271                "[00000.123] 01234:05678> [bar] INFO: {}\n{} key=value other_key=3\n",
272                &message[..218],
273                &message[218..]
274            )
275        );
276    }
277
278    #[fuchsia::test]
279    fn when_no_tags_are_present_the_component_name_is_used() {
280        let log = LogsDataBuilder::new(BuilderArgs {
281            timestamp: BootInstant::from_nanos(123456789),
282            component_url: Some("url".into()),
283            moniker: "core/foo".try_into().unwrap(),
284            severity: Severity::Info,
285        })
286        .set_message("my msg")
287        .set_pid(1234)
288        .set_tid(5678)
289        .build();
290        let mut sink = Vec::new();
291        SerialWriter::log(&log, &HashSet::new(), &mut sink).expect("write succeeded");
292        assert_eq!(
293            String::from_utf8(sink).unwrap(),
294            "[00000.123] 01234:05678> [foo] INFO: my msg\n"
295        );
296    }
297
298    #[fuchsia::test]
299    async fn writes_ingested_logs() {
300        let serial_config = SerialConfig::new(vec!["bootstrap/**", "/core/foo"], vec!["foo"]);
301        let repo = LogsRepository::for_test(fasync::Scope::new());
302
303        let bootstrap_foo_container = repo.get_log_container(Arc::new(ComponentIdentity::new(
304            ExtendedMoniker::parse_str("./bootstrap/foo").unwrap(),
305            "fuchsia-pkg://bootstrap-foo",
306        )));
307        let bootstrap_bar_container = repo.get_log_container(Arc::new(ComponentIdentity::new(
308            ExtendedMoniker::parse_str("./bootstrap/bar").unwrap(),
309            "fuchsia-pkg://bootstrap-bar",
310        )));
311
312        let core_foo_container = repo.get_log_container(Arc::new(ComponentIdentity::new(
313            ExtendedMoniker::parse_str("./core/foo").unwrap(),
314            "fuchsia-pkg://core-foo",
315        )));
316        let core_baz_container = repo.get_log_container(Arc::new(ComponentIdentity::new(
317            ExtendedMoniker::parse_str("./core/baz").unwrap(),
318            "fuchsia-pkg://core-baz",
319        )));
320
321        bootstrap_foo_container.ingest_message(make_message(
322            "a",
323            None,
324            zx::BootInstant::from_nanos(1),
325        ));
326        core_baz_container.ingest_message(make_message("c", None, zx::BootInstant::from_nanos(2)));
327        let (sink, rcv) = TestSink::new();
328        let mut serial_task = pin!(serial_config.write_logs(Arc::clone(&repo), sink));
329        bootstrap_bar_container.ingest_message(make_message(
330            "b",
331            Some("foo"),
332            zx::BootInstant::from_nanos(3),
333        ));
334        core_foo_container.ingest_message(make_message("c", None, zx::BootInstant::from_nanos(4)));
335        poll_fn(|context| loop {
336            if Poll::Pending == serial_task.poll_unpin(context) {
337                return Poll::Ready(());
338            }
339        })
340        .await;
341        let received = rcv.take(2).collect::<Vec<_>>().now_or_never().unwrap();
342
343        // We must see the logs emitted before we installed the serial listener and after. We must
344        // not see the log from /core/baz and we must not see the log from bootstrap/bar with tag
345        // "foo".
346        assert_eq!(
347            received,
348            vec![
349                "[00000.000] 00001:00002> [foo] DEBUG: a\n",
350                "[00000.000] 00001:00002> [foo] DEBUG: c\n"
351            ]
352        );
353    }
354}
archivist_lib/logs/serial.rs

archivist_lib/logs/
serial.rs
