log_command/
log_formatter.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::filter::LogFilterCriteria;
6use crate::log_socket_stream::{JsonDeserializeError, LogsDataStream};
7use crate::{
8    DetailedDateTime, InstanceGetter, LogCommand, LogError, LogProcessingResult, LogSubCommand,
9    TimeFormat,
10};
11use anyhow::Result;
12use async_trait::async_trait;
13use diagnostics_data::{
14    Data, LogTextColor, LogTextDisplayOptions, LogTextPresenter, LogTimeDisplayFormat, Logs,
15    LogsData, LogsDataBuilder, LogsField, LogsProperty, Severity, Timezone,
16};
17use futures_util::future::Either;
18use futures_util::stream::FuturesUnordered;
19use futures_util::{StreamExt, select};
20use serde::{Deserialize, Serialize};
21use std::fmt::Display;
22use std::io::Write;
23use std::time::Duration;
24use thiserror::Error;
25use writer::ToolIO;
26
27pub use diagnostics_data::Timestamp;
28
29pub const TIMESTAMP_FORMAT: &str = "%Y-%m-%d %H:%M:%S.%3f";
30
31/// Type of data in a log entry
32#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
33pub enum LogData {
34    /// A log entry from the target
35    TargetLog(LogsData),
36}
37
38impl LogData {
39    /// Gets the LogData as a target log.
40    pub fn as_target_log(&self) -> Option<&LogsData> {
41        match self {
42            LogData::TargetLog(log) => Some(log),
43        }
44    }
45
46    pub fn as_target_log_mut(&mut self) -> Option<&mut LogsData> {
47        match self {
48            LogData::TargetLog(log) => Some(log),
49        }
50    }
51}
52
53impl From<LogsData> for LogData {
54    fn from(data: LogsData) -> Self {
55        Self::TargetLog(data)
56    }
57}
58
59/// A log entry from either the host, target, or
60/// a symbolized log.
61#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
62pub struct LogEntry {
63    /// The log
64    pub data: LogData,
65}
66
67impl LogEntry {
68    fn utc_timestamp(&self, boot_ts: Option<Timestamp>) -> Timestamp {
69        Timestamp::from_nanos(match &self.data {
70            LogData::TargetLog(data) => {
71                data.metadata.timestamp.into_nanos()
72                    + boot_ts.map(|value| value.into_nanos()).unwrap_or(0)
73            }
74        })
75    }
76}
77
78// Required if we want to use ffx's built-in I/O, but
79// this isn't really applicable to us because we have
80// custom formatting rules.
81impl Display for LogEntry {
82    fn fmt(&self, _f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
83        unreachable!("UNSUPPORTED -- This type cannot be formatted with std format.");
84    }
85}
86
87/// A trait for symbolizing log entries
88#[async_trait(?Send)]
89pub trait Symbolize {
90    /// Symbolizes a LogEntry and optionally produces a result.
91    /// The symbolizer may choose to discard the result.
92    /// This method may be called multiple times concurrently.
93    async fn symbolize(&self, entry: LogEntry) -> Option<LogEntry>;
94}
95
96async fn handle_value<S>(one: Data<Logs>, symbolizer: &S) -> Option<LogEntry>
97where
98    S: Symbolize + ?Sized,
99{
100    let entry = LogEntry { data: one.into() };
101    symbolizer.symbolize(entry).await
102}
103
104fn generate_timestamp_message(boot_timestamp: Timestamp) -> LogEntry {
105    LogEntry {
106        data: LogData::TargetLog(
107            LogsDataBuilder::new(diagnostics_data::BuilderArgs {
108                moniker: "ffx".try_into().unwrap(),
109                timestamp: Timestamp::from_nanos(0),
110                component_url: Some("ffx".into()),
111                severity: Severity::Info,
112            })
113            .set_message("Logging started")
114            .add_key(LogsProperty::String(
115                LogsField::Other("utc_time_now".into()),
116                chrono::Utc::now().to_rfc3339(),
117            ))
118            .add_key(LogsProperty::Int(
119                LogsField::Other("current_boot_timestamp".to_string()),
120                boot_timestamp.into_nanos(),
121            ))
122            .build(),
123        ),
124    }
125}
126
127/// Reads logs from a socket and formats them using the given formatter and symbolizer.
128pub async fn dump_logs_from_socket<F, S>(
129    socket: flex_client::AsyncSocket,
130    formatter: &mut F,
131    symbolizer: &S,
132    include_timestamp: bool,
133) -> Result<LogProcessingResult, JsonDeserializeError>
134where
135    F: LogFormatter + BootTimeAccessor,
136    S: Symbolize + ?Sized,
137{
138    let mut decoder = Box::pin(LogsDataStream::new(socket).fuse());
139    let mut symbolize_pending = FuturesUnordered::new();
140    if include_timestamp && !formatter.is_utc_time_format() {
141        formatter.push_log(generate_timestamp_message(formatter.get_boot_timestamp())).await?;
142    }
143    while let Some(value) = select! {
144        res = decoder.next() => Some(Either::Left(res)),
145        res = symbolize_pending.next() => Some(Either::Right(res)),
146        complete => None,
147    } {
148        match value {
149            Either::Left(Some(result)) => match result {
150                Ok(log) => symbolize_pending.push(handle_value(log, symbolizer)),
151                Err(e) => return Err(e),
152            },
153            Either::Right(Some(Some(symbolized))) => match formatter.push_log(symbolized).await? {
154                LogProcessingResult::Exit => {
155                    formatter.flush().await?;
156                    return Ok(LogProcessingResult::Exit);
157                }
158                LogProcessingResult::Continue => {}
159            },
160            _ => {}
161        }
162    }
163    formatter.flush().await?;
164    Ok(LogProcessingResult::Continue)
165}
166
167/// Reads FXT logs from a socket and formats them using the given formatter and symbolizer.
168pub async fn dump_fxt_logs_from_socket<F, S>(
169    socket: flex_client::AsyncSocket,
170    formatter: &mut F,
171    symbolizer: &S,
172    include_timestamp: bool,
173) -> Result<LogProcessingResult, LogError>
174where
175    F: LogFormatter + BootTimeAccessor,
176    S: Symbolize + ?Sized,
177{
178    let streamer = crate::fxt_streamer::FxtStreamer::new(socket);
179    let mut decoder = std::pin::pin!(streamer.stream());
180    let mut symbolize_pending = FuturesUnordered::new();
181    if include_timestamp && !formatter.is_utc_time_format() {
182        formatter.push_log(generate_timestamp_message(formatter.get_boot_timestamp())).await?;
183    }
184    while let Some(value) = select! {
185        res = decoder.next() => Some(Either::Left(res)),
186        res = symbolize_pending.next() => Some(Either::Right(res)),
187        complete => None,
188    } {
189        match value {
190            Either::Left(Some(result)) => match result {
191                Ok(log) => symbolize_pending.push(handle_value(log, symbolizer)),
192                Err(e) => return Err(e),
193            },
194            Either::Right(Some(Some(symbolized))) => match formatter.push_log(symbolized).await? {
195                LogProcessingResult::Exit => {
196                    formatter.flush().await?;
197                    return Ok(LogProcessingResult::Exit);
198                }
199                LogProcessingResult::Continue => {}
200            },
201            _ => {}
202        }
203    }
204    formatter.flush().await?;
205    Ok(LogProcessingResult::Continue)
206}
207
208pub trait BootTimeAccessor {
209    /// Sets the boot timestamp in nanoseconds since the Unix epoch.
210    fn set_boot_timestamp(&mut self, _boot_ts_nanos: Timestamp);
211
212    /// Returns the boot timestamp in nanoseconds since the Unix epoch.
213    fn get_boot_timestamp(&self) -> Timestamp;
214}
215
216/// Timestamp filter which is either either boot-based or UTC-based.
217#[derive(Clone, Debug)]
218pub struct DeviceOrLocalTimestamp {
219    /// Timestamp in boot time
220    pub timestamp: Timestamp,
221    /// True if this filter should be applied to boot time,
222    /// false if UTC time.
223    pub is_boot: bool,
224}
225
226impl DeviceOrLocalTimestamp {
227    /// Creates a DeviceOrLocalTimestamp from a real-time date/time or
228    /// a boot date/time. Returns None if both rtc and boot are None.
229    /// Returns None if the timestamp is "now".
230    pub fn new(
231        rtc: Option<&DetailedDateTime>,
232        boot: Option<&Duration>,
233    ) -> Option<DeviceOrLocalTimestamp> {
234        rtc.as_ref()
235            .filter(|value| !value.is_now)
236            .map(|value| DeviceOrLocalTimestamp {
237                timestamp: Timestamp::from_nanos(
238                    value.naive_utc().and_utc().timestamp_nanos_opt().unwrap(),
239                ),
240                is_boot: false,
241            })
242            .or_else(|| {
243                boot.map(|value| DeviceOrLocalTimestamp {
244                    timestamp: Timestamp::from_nanos(value.as_nanos() as i64),
245                    is_boot: true,
246                })
247            })
248    }
249}
250
251/// Log formatter options
252#[derive(Clone, Debug)]
253pub struct LogFormatterOptions {
254    /// Text display options
255    pub display: Option<LogTextDisplayOptions>,
256    /// Only display logs since the specified time.
257    pub since: Option<DeviceOrLocalTimestamp>,
258    /// Only display logs until the specified time.
259    pub until: Option<DeviceOrLocalTimestamp>,
260    /// Only display the last N log lines.
261    pub tail: Option<usize>,
262    /// Sort logs by timestamp when flushing (used in snapshot/dump mode).
263    pub sort: bool,
264}
265
266impl Default for LogFormatterOptions {
267    fn default() -> Self {
268        LogFormatterOptions {
269            display: Some(Default::default()),
270            since: None,
271            until: None,
272            tail: None,
273            sort: false,
274        }
275    }
276}
277
278/// Log formatter error
279#[derive(Error, Debug)]
280pub enum FormatterError {
281    /// An unknown error occurred
282    #[error(transparent)]
283    Other(#[from] anyhow::Error),
284    /// An IO error occurred
285    #[error(transparent)]
286    IO(#[from] std::io::Error),
287}
288
289impl FormatterError {
290    pub fn is_broken_pipe(&self) -> bool {
291        match self {
292            FormatterError::IO(error) => error.kind() == std::io::ErrorKind::BrokenPipe,
293            FormatterError::Other(_) => false,
294        }
295    }
296}
297
298/// Maximum number of log entries to retain in the reorder window during `dump` (`sort: true`).
299const REORDER_WINDOW_SIZE: usize = 64;
300
301/// Maximum timestamp difference (1 second) considered plausible for out-of-order arrival due to
302/// thread preemption between timestamp capture and IOBuffer write. Messages with larger skew are
303/// treated as having untrusted/bogus timestamps and are not reordered across each other.
304const MAX_REORDER_SKEW_NANOS: i64 = 1_000_000_000;
305
306/// Default formatter implementation
307pub struct DefaultLogFormatter<W>
308where
309    W: Write + ToolIO<OutputItem = LogEntry>,
310{
311    writer: W,
312    filters: LogFilterCriteria,
313    options: LogFormatterOptions,
314    boot_ts_nanos: Option<Timestamp>,
315    /// Timestamp of the most recently emitted log entry in the current boot, used to highlight
316    /// timestamps that precede the immediately previous line.
317    last_timestamp: Option<Timestamp>,
318    tail_queue: std::collections::VecDeque<LogEntry>,
319    tail_limit: Option<usize>,
320}
321
322/// Converts from UTC time to boot time.
323fn utc_to_boot(boot_ts: Timestamp, utc: Timestamp) -> Timestamp {
324    Timestamp::from_nanos(utc.into_nanos() - boot_ts.into_nanos())
325}
326
327#[async_trait(?Send)]
328impl<W> LogFormatter for DefaultLogFormatter<W>
329where
330    W: Write + ToolIO<OutputItem = LogEntry>,
331{
332    async fn push_log(&mut self, log_entry: LogEntry) -> Result<LogProcessingResult, LogError> {
333        self.push_log_internal(log_entry, true).await.or_else(|err| {
334            if err.is_broken_pipe() { Ok(LogProcessingResult::Exit) } else { Err(err) }
335        })
336    }
337
338    fn is_utc_time_format(&self) -> bool {
339        self.options.display.iter().any(|options| match options.time_format {
340            LogTimeDisplayFormat::Original => false,
341            LogTimeDisplayFormat::WallTime { tz, offset: _ } => tz == Timezone::Utc,
342        })
343    }
344
345    async fn flush(&mut self) -> Result<(), LogError> {
346        self.flush_tail().await
347    }
348}
349
350impl<W> BootTimeAccessor for DefaultLogFormatter<W>
351where
352    W: Write + ToolIO<OutputItem = LogEntry>,
353{
354    fn set_boot_timestamp(&mut self, boot_ts_nanos: Timestamp) {
355        if let Some(LogTextDisplayOptions {
356            time_format: LogTimeDisplayFormat::WallTime { offset, .. },
357            ..
358        }) = &mut self.options.display
359        {
360            *offset = boot_ts_nanos.into_nanos();
361        }
362        self.boot_ts_nanos = Some(boot_ts_nanos);
363        // Reset timestamp tracking across target reboots so timestamps from the new boot (which
364        // restart near zero) are not treated as out-of-order relative to the previous boot.
365        self.last_timestamp = None;
366    }
367    fn get_boot_timestamp(&self) -> Timestamp {
368        debug_assert!(self.boot_ts_nanos.is_some());
369        self.boot_ts_nanos.unwrap_or_else(|| Timestamp::from_nanos(0))
370    }
371}
372
373/// Object which contains a Writer that can be borrowed
374pub trait WriterContainer<W>
375where
376    W: Write + ToolIO<OutputItem = LogEntry>,
377{
378    fn writer(&mut self) -> &mut W;
379}
380
381impl<W> WriterContainer<W> for DefaultLogFormatter<W>
382where
383    W: Write + ToolIO<OutputItem = LogEntry>,
384{
385    fn writer(&mut self) -> &mut W {
386        &mut self.writer
387    }
388}
389
390impl<W> DefaultLogFormatter<W>
391where
392    W: Write + ToolIO<OutputItem = LogEntry>,
393{
394    /// Creates a new DefaultLogFormatter with the given writer and options.
395    pub fn new(filters: LogFilterCriteria, writer: W, options: LogFormatterOptions) -> Self {
396        let tail_limit = options.tail;
397        Self {
398            filters,
399            writer,
400            options,
401            boot_ts_nanos: None,
402            last_timestamp: None,
403            tail_queue: std::collections::VecDeque::new(),
404            tail_limit,
405        }
406    }
407
408    pub async fn expand_monikers(&mut self, getter: &impl InstanceGetter) -> Result<(), LogError> {
409        let warnings = self.filters.expand_monikers(getter).await?;
410        for warning in warnings {
411            writeln!(
412                self.writer.stderr(),
413                "WARN: Provided moniker '{}' was not an exact match. Using fuzzy match '{}' instead. Please check your component topology variations.",
414                warning.query,
415                warning.resolved
416            )?;
417        }
418        Ok(())
419    }
420
421    pub async fn push_unfiltered_log(
422        &mut self,
423        log_entry: LogEntry,
424    ) -> Result<LogProcessingResult, LogError> {
425        self.push_log_internal(log_entry, false).await
426    }
427
428    /// Inserts `log_entry` into `self.tail_queue`, bubbling it backwards past any preceding
429    /// entries whose timestamps are slightly newer (within `MAX_REORDER_SKEW_NANOS`).
430    ///
431    /// If `prev_ts - new_ts > MAX_REORDER_SKEW_NANOS`, either `prev_ts` is a bogus far-future
432    /// timestamp or `new_ts` is a bogus far-past timestamp; in both cases we stop and do not
433    /// reorder across the gap so that messages with untrusted timestamps remain in their original
434    /// arrival position.
435    fn insert_into_reorder_window(&mut self, log_entry: LogEntry) {
436        let LogData::TargetLog(new_data) = &log_entry.data;
437        let new_ts = new_data.metadata.timestamp.into_nanos();
438        let mut insert_pos = self.tail_queue.len();
439        while insert_pos > 0 {
440            let LogData::TargetLog(prev_data) = &self.tail_queue[insert_pos - 1].data;
441            let prev_ts = prev_data.metadata.timestamp.into_nanos();
442            let diff = prev_ts.saturating_sub(new_ts);
443            if diff > 0 && diff <= MAX_REORDER_SKEW_NANOS {
444                insert_pos -= 1;
445            } else {
446                break;
447            }
448        }
449        self.tail_queue.insert(insert_pos, log_entry);
450    }
451
452    async fn flush_tail(&mut self) -> Result<(), LogError> {
453        // When both `sort` and `tail_limit` are active, `push_log_internal` retains up to
454        // `limit.max(REORDER_WINDOW_SIZE)` entries so that logs near the tail boundary can be
455        // reordered before we trim to the final `limit` entries here.
456        if let Some(limit) = self.tail_limit
457            && self.tail_queue.len() > limit
458        {
459            let drain_count = self.tail_queue.len() - limit;
460            self.tail_queue.drain(..drain_count);
461        }
462        while let Some(log_entry) = self.tail_queue.pop_front() {
463            if let Err(err) = self.write_log_entry(log_entry, true) {
464                if err.is_broken_pipe() {
465                    return Ok(());
466                }
467                return Err(err);
468            }
469        }
470        Ok(())
471    }
472
473    fn write_log_entry(
474        &mut self,
475        log_entry: LogEntry,
476        enable_filters: bool,
477    ) -> Result<(), LogError> {
478        let is_out_of_order = if enable_filters {
479            let LogData::TargetLog(data) = &log_entry.data;
480            let ts = data.metadata.timestamp;
481            // Highlight whenever a message's timestamp precedes the immediately previous message.
482            // Tracking only the previous timestamp keeps the check simple and self-correcting if
483            // one or more messages carry bogus timestamps.
484            let out_of_order = self.last_timestamp.is_some_and(|last_ts| ts < last_ts);
485            self.last_timestamp = Some(ts);
486            out_of_order
487        } else {
488            false
489        };
490        match self.options.display {
491            Some(mut text_options) => {
492                text_options.highlight_timestamp = is_out_of_order;
493                let mut options_for_this_line_only = self.options.clone();
494                options_for_this_line_only.display = Some(text_options);
495                // For host logs, don't apply the boot time offset
496                // as this is with reference to the UTC timeline
497                if !enable_filters
498                    && let LogTimeDisplayFormat::WallTime { ref mut offset, .. } =
499                        options_for_this_line_only.display.as_mut().unwrap().time_format
500                {
501                    // 1 nanosecond so that LogTimeDisplayFormat in diagnostics_data
502                    // knows that we have a valid UTC offset. It normally falls back if
503                    // the UTC offset is 0. It prints at millisecond precision so being
504                    // off by +1 nanosecond isn't an issue.
505                    *offset = 1;
506                }
507                self.format_text_log(options_for_this_line_only, log_entry)
508                    .map_err(LogError::FormatterError)?;
509            }
510            None => {
511                self.writer.item(&log_entry).map_err(|err| LogError::UnknownError(err.into()))?;
512            }
513        };
514        Ok(())
515    }
516
517    async fn push_log_internal(
518        &mut self,
519        log_entry: LogEntry,
520        enable_filters: bool,
521    ) -> Result<LogProcessingResult, LogError> {
522        if enable_filters {
523            if self.filter_by_timestamp(&log_entry, self.options.since.as_ref(), |a, b| a <= b) {
524                return Ok(LogProcessingResult::Continue);
525            }
526
527            if self.filter_by_timestamp(&log_entry, self.options.until.as_ref(), |a, b| a >= b) {
528                if self.options.sort {
529                    // When sorting in dump (snapshot) mode, logs may arrive slightly out of
530                    // timestamp order, or a single log may carry a bogus far-future timestamp.
531                    // Skip the record instead of exiting early so we don't miss subsequent
532                    // records in the snapshot whose timestamps are earlier than `until`.
533                    return Ok(LogProcessingResult::Continue);
534                }
535                return Ok(LogProcessingResult::Exit);
536            }
537
538            if !self.filters.matches(&log_entry) {
539                return Ok(LogProcessingResult::Continue);
540            }
541        }
542
543        if self.options.sort {
544            self.insert_into_reorder_window(log_entry);
545            if let Some(limit) = self.tail_limit {
546                // Keep at least `REORDER_WINDOW_SIZE` entries so out-of-order arrivals near the
547                // tail cutoff can be reordered before `flush_tail` trims to `limit`.
548                let window = limit.max(REORDER_WINDOW_SIZE);
549                while self.tail_queue.len() > window {
550                    self.tail_queue.pop_front();
551                }
552            } else {
553                // Emit entries once they fall behind the reorder window so dump output streams
554                // progressively with bounded memory.
555                while self.tail_queue.len() > REORDER_WINDOW_SIZE {
556                    let oldest = self.tail_queue.pop_front().unwrap();
557                    self.write_log_entry(oldest, enable_filters)?;
558                }
559            }
560        } else if let Some(limit) = self.tail_limit {
561            self.tail_queue.push_back(log_entry);
562            if self.tail_queue.len() > limit {
563                self.tail_queue.pop_front();
564            }
565        } else {
566            self.write_log_entry(log_entry, enable_filters)?;
567        }
568        Ok(LogProcessingResult::Continue)
569    }
570
571    /// Creates a new DefaultLogFormatter from command-line arguments.
572    pub fn new_from_args(cmd: &LogCommand, writer: W) -> Result<Self, LogError> {
573        let is_json = writer.is_machine();
574
575        Ok(DefaultLogFormatter::new(
576            LogFilterCriteria::try_from(cmd.clone())?,
577            writer,
578            LogFormatterOptions {
579                display: if is_json {
580                    None
581                } else {
582                    Some(LogTextDisplayOptions {
583                        show_tags: !cmd.hide_tags(),
584                        color: if cmd.no_color() {
585                            LogTextColor::None
586                        } else {
587                            LogTextColor::BySeverity
588                        },
589                        show_metadata: cmd.show_metadata(),
590                        time_format: match cmd.clock() {
591                            TimeFormat::Boot => LogTimeDisplayFormat::Original,
592                            TimeFormat::Local => LogTimeDisplayFormat::WallTime {
593                                tz: Timezone::Local,
594                                // This will receive a correct value when logging actually starts,
595                                // see `set_boot_timestamp()` method on the log formatter.
596                                offset: 0,
597                            },
598                            TimeFormat::Utc => LogTimeDisplayFormat::WallTime {
599                                tz: Timezone::Utc,
600                                // This will receive a correct value when logging actually starts,
601                                // see `set_boot_timestamp()` method on the log formatter.
602                                offset: 0,
603                            },
604                        },
605                        show_file: !cmd.hide_file(),
606                        show_moniker: !cmd.hide_moniker(),
607                        show_full_moniker: cmd.show_full_moniker(),
608                        prefer_url_component_name: cmd.prefer_url_component_name(),
609                        highlight_timestamp: false,
610                    })
611                },
612                since: DeviceOrLocalTimestamp::new(cmd.since(), cmd.since_boot().as_ref()),
613                until: DeviceOrLocalTimestamp::new(cmd.until(), cmd.until_boot().as_ref()),
614                tail: match &cmd.sub_command {
615                    Some(LogSubCommand::Dump(dump)) => dump.tail,
616                    _ => None,
617                },
618                sort: matches!(cmd.sub_command, Some(LogSubCommand::Dump(..))),
619            },
620        ))
621    }
622
623    fn filter_by_timestamp(
624        &self,
625        log_entry: &LogEntry,
626        timestamp: Option<&DeviceOrLocalTimestamp>,
627        callback: impl Fn(&Timestamp, &Timestamp) -> bool,
628    ) -> bool {
629        let Some(timestamp) = timestamp else {
630            return false;
631        };
632        if timestamp.is_boot {
633            callback(
634                &utc_to_boot(
635                    self.get_boot_timestamp(),
636                    log_entry.utc_timestamp(self.boot_ts_nanos),
637                ),
638                &timestamp.timestamp,
639            )
640        } else {
641            callback(&log_entry.utc_timestamp(self.boot_ts_nanos), &timestamp.timestamp)
642        }
643    }
644
645    // This function's arguments are copied to make lifetimes in push_log easier since borrowing
646    // &self would complicate spam highlighting.
647    fn format_text_log(
648        &mut self,
649        options: LogFormatterOptions,
650        log_entry: LogEntry,
651    ) -> Result<(), FormatterError> {
652        let text_options = match options.display {
653            Some(o) => o,
654            None => {
655                unreachable!("If we are here, we can only be formatting text");
656            }
657        };
658        match log_entry {
659            LogEntry { data: LogData::TargetLog(data), .. } => {
660                // TODO(https://fxbug.dev/42072442): Add support for log spam redaction and other
661                // features listed in the design doc.
662                writeln!(self.writer, "{}", LogTextPresenter::new(&data, text_options))?;
663            }
664        }
665        Ok(())
666    }
667}
668
669#[allow(dead_code)] // TODO(https://fxbug.dev/421409178)
670/// Symbolizer that does nothing.
671pub struct NoOpSymbolizer;
672
673#[async_trait(?Send)]
674impl Symbolize for NoOpSymbolizer {
675    async fn symbolize(&self, entry: LogEntry) -> Option<LogEntry> {
676        Some(entry)
677    }
678}
679
680/// Trait for formatting logs one at a time.
681#[async_trait(?Send)]
682pub trait LogFormatter {
683    /// Formats a log entry and writes it to the output.
684    async fn push_log(&mut self, log_entry: LogEntry) -> Result<LogProcessingResult, LogError>;
685
686    /// Returns true if the formatter is configured to output in UTC time format.
687    fn is_utc_time_format(&self) -> bool;
688
689    /// Flushes any buffered logs.
690    async fn flush(&mut self) -> Result<(), LogError> {
691        Ok(())
692    }
693}
694
695#[cfg(test)]
696mod test {
697    use crate::parse_time;
698    use assert_matches::assert_matches;
699    use diagnostics_data::{LogsDataBuilder, Severity};
700    use std::cell::Cell;
701    use writer::{Format, JsonWriter, TestBuffers};
702
703    use super::*;
704
705    const DEFAULT_TS_NANOS: u64 = 1615535969000000000;
706
707    struct FakeFormatter {
708        logs: Vec<LogEntry>,
709        boot_timestamp: Timestamp,
710        is_utc_time_format: bool,
711    }
712
713    impl FakeFormatter {
714        fn new() -> Self {
715            Self {
716                logs: Vec::new(),
717                boot_timestamp: Timestamp::from_nanos(0),
718                is_utc_time_format: false,
719            }
720        }
721    }
722
723    impl BootTimeAccessor for FakeFormatter {
724        fn set_boot_timestamp(&mut self, boot_ts_nanos: Timestamp) {
725            self.boot_timestamp = boot_ts_nanos;
726        }
727
728        fn get_boot_timestamp(&self) -> Timestamp {
729            self.boot_timestamp
730        }
731    }
732
733    #[async_trait(?Send)]
734    impl LogFormatter for FakeFormatter {
735        async fn push_log(&mut self, log_entry: LogEntry) -> Result<LogProcessingResult, LogError> {
736            self.logs.push(log_entry);
737            Ok(LogProcessingResult::Continue)
738        }
739
740        fn is_utc_time_format(&self) -> bool {
741            self.is_utc_time_format
742        }
743    }
744
745    /// Symbolizer that prints "Fuchsia".
746    pub struct FakeFuchsiaSymbolizer;
747
748    fn set_log_msg(entry: &mut LogEntry, msg: impl Into<String>) {
749        *entry.data.as_target_log_mut().unwrap().msg_mut().unwrap() = msg.into();
750    }
751
752    #[async_trait(?Send)]
753    impl Symbolize for FakeFuchsiaSymbolizer {
754        async fn symbolize(&self, mut entry: LogEntry) -> Option<LogEntry> {
755            set_log_msg(&mut entry, "Fuchsia");
756            Some(entry)
757        }
758    }
759
760    struct FakeSymbolizerCallback {
761        should_discard: Cell<bool>,
762    }
763
764    impl FakeSymbolizerCallback {
765        fn new() -> Self {
766            Self { should_discard: Cell::new(true) }
767        }
768    }
769
770    async fn dump_logs_from_socket<F, S>(
771        socket: fuchsia_async::Socket,
772        formatter: &mut F,
773        symbolizer: &S,
774    ) -> Result<LogProcessingResult, JsonDeserializeError>
775    where
776        F: LogFormatter + BootTimeAccessor,
777        S: Symbolize + ?Sized,
778    {
779        super::dump_logs_from_socket(socket, formatter, symbolizer, false).await
780    }
781
782    #[async_trait(?Send)]
783    impl Symbolize for FakeSymbolizerCallback {
784        async fn symbolize(&self, mut input: LogEntry) -> Option<LogEntry> {
785            self.should_discard.set(!self.should_discard.get());
786            if self.should_discard.get() {
787                None
788            } else {
789                set_log_msg(&mut input, "symbolized log");
790                Some(input)
791            }
792        }
793    }
794
795    #[fuchsia::test]
796    async fn test_boot_timestamp_setter() {
797        let buffers = TestBuffers::default();
798        let stdout = JsonWriter::<LogEntry>::new_test(None, &buffers);
799        let options = LogFormatterOptions {
800            display: Some(LogTextDisplayOptions {
801                time_format: LogTimeDisplayFormat::WallTime { tz: Timezone::Utc, offset: 0 },
802                ..Default::default()
803            }),
804            ..Default::default()
805        };
806        let mut formatter =
807            DefaultLogFormatter::new(LogFilterCriteria::default(), stdout, options.clone());
808        formatter.set_boot_timestamp(Timestamp::from_nanos(1234));
809        assert_eq!(formatter.get_boot_timestamp(), Timestamp::from_nanos(1234));
810
811        // Boot timestamp is supported when using JSON output (for filtering)
812        let buffers = TestBuffers::default();
813        let output = JsonWriter::<LogEntry>::new_test(None, &buffers);
814        let options = LogFormatterOptions { display: None, ..Default::default() };
815        let mut formatter = DefaultLogFormatter::new(LogFilterCriteria::default(), output, options);
816        formatter.set_boot_timestamp(Timestamp::from_nanos(1234));
817        assert_eq!(formatter.get_boot_timestamp(), Timestamp::from_nanos(1234));
818    }
819
820    #[fuchsia::test]
821    async fn test_format_single_message() {
822        let symbolizer = NoOpSymbolizer {};
823        let mut formatter = FakeFormatter::new();
824        let target_log = LogsDataBuilder::new(diagnostics_data::BuilderArgs {
825            moniker: "ffx".try_into().unwrap(),
826            timestamp: Timestamp::from_nanos(0),
827            component_url: Some("ffx".into()),
828            severity: Severity::Info,
829        })
830        .set_message("Hello world!")
831        .build();
832        let (sender, receiver) = zx::Socket::create_stream();
833        sender
834            .write(serde_json::to_string(&target_log).unwrap().as_bytes())
835            .expect("failed to write target log");
836        drop(sender);
837        dump_logs_from_socket(flex_client::socket_to_async(receiver), &mut formatter, &symbolizer)
838            .await
839            .unwrap();
840        assert_eq!(formatter.logs, vec![LogEntry { data: LogData::TargetLog(target_log) }]);
841    }
842
843    #[fuchsia::test]
844    async fn test_format_utc_timestamp() {
845        let symbolizer = NoOpSymbolizer {};
846        let mut formatter = FakeFormatter::new();
847        formatter.set_boot_timestamp(Timestamp::from_nanos(DEFAULT_TS_NANOS as i64));
848        let (_, receiver) = zx::Socket::create_stream();
849        super::dump_logs_from_socket(
850            flex_client::socket_to_async(receiver),
851            &mut formatter,
852            &symbolizer,
853            true,
854        )
855        .await
856        .unwrap();
857        let target_log = formatter.logs[0].data.as_target_log().unwrap();
858        let properties = target_log.payload_keys().unwrap();
859        assert_eq!(target_log.msg().unwrap(), "Logging started");
860
861        // Ensure the end has a valid timestamp
862        chrono::DateTime::parse_from_rfc3339(
863            properties.get_property("utc_time_now").unwrap().string().unwrap(),
864        )
865        .unwrap();
866        assert_eq!(
867            properties.get_property("current_boot_timestamp").unwrap().int().unwrap(),
868            DEFAULT_TS_NANOS as i64
869        );
870    }
871
872    #[fuchsia::test]
873    async fn test_format_utc_timestamp_does_not_print_if_utc_time() {
874        let symbolizer = NoOpSymbolizer {};
875        let mut formatter = FakeFormatter::new();
876        formatter.is_utc_time_format = true;
877        formatter.set_boot_timestamp(Timestamp::from_nanos(DEFAULT_TS_NANOS as i64));
878        let (_, receiver) = zx::Socket::create_stream();
879        super::dump_logs_from_socket(
880            flex_client::socket_to_async(receiver),
881            &mut formatter,
882            &symbolizer,
883            true,
884        )
885        .await
886        .unwrap();
887        assert_eq!(formatter.logs.len(), 0);
888    }
889
890    #[fuchsia::test]
891    async fn test_format_multiple_messages() {
892        let symbolizer = NoOpSymbolizer {};
893        let mut formatter = FakeFormatter::new();
894        let (sender, receiver) = zx::Socket::create_stream();
895        let target_log_0 = LogsDataBuilder::new(diagnostics_data::BuilderArgs {
896            moniker: "ffx".try_into().unwrap(),
897            timestamp: Timestamp::from_nanos(0),
898            component_url: Some("ffx".into()),
899            severity: Severity::Info,
900        })
901        .set_message("Hello world!")
902        .set_pid(1)
903        .set_tid(2)
904        .build();
905        let target_log_1 = LogsDataBuilder::new(diagnostics_data::BuilderArgs {
906            moniker: "ffx".try_into().unwrap(),
907            timestamp: Timestamp::from_nanos(1),
908            component_url: Some("ffx".into()),
909            severity: Severity::Info,
910        })
911        .set_message("Hello world 2!")
912        .build();
913        sender
914            .write(serde_json::to_string(&vec![&target_log_0, &target_log_1]).unwrap().as_bytes())
915            .expect("failed to write target log");
916        drop(sender);
917        dump_logs_from_socket(flex_client::socket_to_async(receiver), &mut formatter, &symbolizer)
918            .await
919            .unwrap();
920        assert_eq!(
921            formatter.logs,
922            vec![
923                LogEntry { data: LogData::TargetLog(target_log_0) },
924                LogEntry { data: LogData::TargetLog(target_log_1) }
925            ]
926        );
927    }
928
929    #[fuchsia::test]
930    async fn test_format_timestamp_filter() {
931        // test since and until args for the LogFormatter
932        let symbolizer = NoOpSymbolizer {};
933        let buffers = TestBuffers::default();
934        let stdout = JsonWriter::<LogEntry>::new_test(None, &buffers);
935        let mut formatter = DefaultLogFormatter::new(
936            LogFilterCriteria::default(),
937            stdout,
938            LogFormatterOptions {
939                since: Some(DeviceOrLocalTimestamp {
940                    timestamp: Timestamp::from_nanos(1),
941                    is_boot: true,
942                }),
943                until: Some(DeviceOrLocalTimestamp {
944                    timestamp: Timestamp::from_nanos(3),
945                    is_boot: true,
946                }),
947                ..Default::default()
948            },
949        );
950        formatter.set_boot_timestamp(Timestamp::from_nanos(0));
951
952        let (sender, receiver) = zx::Socket::create_stream();
953        let target_log_0 = LogsDataBuilder::new(diagnostics_data::BuilderArgs {
954            moniker: "ffx".try_into().unwrap(),
955            timestamp: Timestamp::from_nanos(0),
956            component_url: Some("ffx".into()),
957            severity: Severity::Info,
958        })
959        .set_message("Hello world!")
960        .build();
961        let target_log_1 = LogsDataBuilder::new(diagnostics_data::BuilderArgs {
962            moniker: "ffx".try_into().unwrap(),
963            timestamp: Timestamp::from_nanos(1),
964            component_url: Some("ffx".into()),
965            severity: Severity::Info,
966        })
967        .set_message("Hello world 2!")
968        .build();
969        let target_log_2 = LogsDataBuilder::new(diagnostics_data::BuilderArgs {
970            moniker: "ffx".try_into().unwrap(),
971            timestamp: Timestamp::from_nanos(2),
972            component_url: Some("ffx".into()),
973            severity: Severity::Info,
974        })
975        .set_pid(1)
976        .set_tid(2)
977        .set_message("Hello world 3!")
978        .build();
979        let target_log_3 = LogsDataBuilder::new(diagnostics_data::BuilderArgs {
980            moniker: "ffx".try_into().unwrap(),
981            timestamp: Timestamp::from_nanos(3),
982            component_url: Some("ffx".into()),
983            severity: Severity::Info,
984        })
985        .set_message("Hello world 4!")
986        .set_pid(1)
987        .set_tid(2)
988        .build();
989        sender
990            .write(
991                serde_json::to_string(&vec![
992                    &target_log_0,
993                    &target_log_1,
994                    &target_log_2,
995                    &target_log_3,
996                ])
997                .unwrap()
998                .as_bytes(),
999            )
1000            .expect("failed to write target log");
1001        drop(sender);
1002        assert_matches!(
1003            dump_logs_from_socket(
1004                flex_client::socket_to_async(receiver),
1005                &mut formatter,
1006                &symbolizer,
1007            )
1008            .await,
1009            Ok(LogProcessingResult::Exit)
1010        );
1011        assert_eq!(
1012            buffers.stdout.into_string(),
1013            "[00000.000000][1][2][ffx] INFO: Hello world 3!\n"
1014        );
1015    }
1016
1017    fn make_log_with_timestamp(timestamp: i64) -> LogsData {
1018        LogsDataBuilder::new(diagnostics_data::BuilderArgs {
1019            moniker: "ffx".try_into().unwrap(),
1020            timestamp: Timestamp::from_nanos(timestamp),
1021            component_url: Some("ffx".into()),
1022            severity: Severity::Info,
1023        })
1024        .set_message(format!("Hello world {timestamp}!"))
1025        .set_pid(1)
1026        .set_tid(2)
1027        .build()
1028    }
1029
1030    #[fuchsia::test]
1031    async fn test_format_timestamp_filter_utc() {
1032        // test since and until args for the LogFormatter
1033        let symbolizer = NoOpSymbolizer {};
1034        let buffers = TestBuffers::default();
1035        let stdout = JsonWriter::<LogEntry>::new_test(None, &buffers);
1036        let mut formatter = DefaultLogFormatter::new(
1037            LogFilterCriteria::default(),
1038            stdout,
1039            LogFormatterOptions {
1040                since: Some(DeviceOrLocalTimestamp {
1041                    timestamp: Timestamp::from_nanos(1),
1042                    is_boot: false,
1043                }),
1044                until: Some(DeviceOrLocalTimestamp {
1045                    timestamp: Timestamp::from_nanos(3),
1046                    is_boot: false,
1047                }),
1048                display: Some(LogTextDisplayOptions {
1049                    time_format: LogTimeDisplayFormat::WallTime { tz: Timezone::Utc, offset: 1 },
1050                    ..Default::default()
1051                }),
1052                ..Default::default()
1053            },
1054        );
1055        formatter.set_boot_timestamp(Timestamp::from_nanos(1));
1056
1057        let (sender, receiver) = zx::Socket::create_stream();
1058        let logs = (0..4).map(make_log_with_timestamp).collect::<Vec<_>>();
1059        sender
1060            .write(serde_json::to_string(&logs).unwrap().as_bytes())
1061            .expect("failed to write target log");
1062        drop(sender);
1063        assert_matches!(
1064            dump_logs_from_socket(
1065                flex_client::socket_to_async(receiver),
1066                &mut formatter,
1067                &symbolizer,
1068            )
1069            .await,
1070            Ok(LogProcessingResult::Exit)
1071        );
1072        assert_eq!(
1073            buffers.stdout.into_string(),
1074            "[1970-01-01 00:00:00.000][1][2][ffx] INFO: Hello world 1!\n"
1075        );
1076    }
1077
1078    fn logs_data_builder() -> LogsDataBuilder {
1079        diagnostics_data::LogsDataBuilder::new(diagnostics_data::BuilderArgs {
1080            timestamp: Timestamp::from_nanos(default_ts().as_nanos() as i64),
1081            component_url: Some("component_url".into()),
1082            moniker: "some/moniker".try_into().unwrap(),
1083            severity: diagnostics_data::Severity::Warn,
1084        })
1085        .set_pid(1)
1086        .set_tid(2)
1087    }
1088
1089    fn default_ts() -> Duration {
1090        Duration::from_nanos(DEFAULT_TS_NANOS)
1091    }
1092
1093    fn log_entry() -> LogEntry {
1094        LogEntry {
1095            data: LogData::TargetLog(
1096                logs_data_builder().add_tag("tag1").add_tag("tag2").set_message("message").build(),
1097            ),
1098        }
1099    }
1100
1101    #[fuchsia::test]
1102    async fn test_default_formatter() {
1103        let buffers = TestBuffers::default();
1104        let stdout = JsonWriter::<LogEntry>::new_test(None, &buffers);
1105        let options = LogFormatterOptions::default();
1106        let mut formatter =
1107            DefaultLogFormatter::new(LogFilterCriteria::default(), stdout, options.clone());
1108        formatter.push_log(log_entry()).await.unwrap();
1109        drop(formatter);
1110        assert_eq!(
1111            buffers.into_stdout_str(),
1112            "[1615535969.000000][1][2][some/moniker][tag1,tag2] WARN: message\n"
1113        );
1114    }
1115
1116    #[fuchsia::test]
1117    async fn test_default_formatter_with_hidden_metadata() {
1118        let buffers = TestBuffers::default();
1119        let stdout = JsonWriter::<LogEntry>::new_test(None, &buffers);
1120        let options = LogFormatterOptions {
1121            display: Some(LogTextDisplayOptions { show_metadata: false, ..Default::default() }),
1122            ..LogFormatterOptions::default()
1123        };
1124        let mut formatter =
1125            DefaultLogFormatter::new(LogFilterCriteria::default(), stdout, options.clone());
1126        formatter.push_log(log_entry()).await.unwrap();
1127        drop(formatter);
1128        assert_eq!(
1129            buffers.into_stdout_str(),
1130            "[1615535969.000000][some/moniker][tag1,tag2] WARN: message\n"
1131        );
1132    }
1133
1134    #[fuchsia::test]
1135    async fn test_default_formatter_with_json() {
1136        let buffers = TestBuffers::default();
1137        let stdout = JsonWriter::<LogEntry>::new_test(Some(Format::Json), &buffers);
1138        let options = LogFormatterOptions { display: None, ..Default::default() };
1139        {
1140            let mut formatter =
1141                DefaultLogFormatter::new(LogFilterCriteria::default(), stdout, options.clone());
1142            formatter.push_log(log_entry()).await.unwrap();
1143        }
1144        assert_eq!(
1145            serde_json::from_str::<LogEntry>(&buffers.into_stdout_str()).unwrap(),
1146            log_entry()
1147        );
1148    }
1149
1150    fn emit_log(sender: &mut zx::Socket, msg: &str, timestamp: i64) -> Data<Logs> {
1151        let target_log = LogsDataBuilder::new(diagnostics_data::BuilderArgs {
1152            moniker: "ffx".try_into().unwrap(),
1153            timestamp: Timestamp::from_nanos(timestamp),
1154            component_url: Some("ffx".into()),
1155            severity: Severity::Info,
1156        })
1157        .set_message(msg)
1158        .build();
1159
1160        sender
1161            .write(serde_json::to_string(&target_log).unwrap().as_bytes())
1162            .expect("failed to write target log");
1163        target_log
1164    }
1165
1166    #[fuchsia::test]
1167    async fn test_default_formatter_discards_when_told_by_symbolizer() {
1168        let mut formatter = FakeFormatter::new();
1169        let (mut sender, receiver) = zx::Socket::create_stream();
1170        let mut target_log_0 = emit_log(&mut sender, "Hello world!", 0);
1171        emit_log(&mut sender, "Dropped world!", 1);
1172        let mut target_log_2 = emit_log(&mut sender, "Hello world!", 2);
1173        emit_log(&mut sender, "Dropped world!", 3);
1174        let mut target_log_4 = emit_log(&mut sender, "Hello world!", 4);
1175        drop(sender);
1176        // Drop every other log.
1177        let symbolizer = FakeSymbolizerCallback::new();
1178        *target_log_0.msg_mut().unwrap() = "symbolized log".into();
1179        *target_log_2.msg_mut().unwrap() = "symbolized log".into();
1180        *target_log_4.msg_mut().unwrap() = "symbolized log".into();
1181        dump_logs_from_socket(flex_client::socket_to_async(receiver), &mut formatter, &symbolizer)
1182            .await
1183            .unwrap();
1184        assert_eq!(
1185            formatter.logs,
1186            vec![
1187                LogEntry { data: LogData::TargetLog(target_log_0) },
1188                LogEntry { data: LogData::TargetLog(target_log_2) },
1189                LogEntry { data: LogData::TargetLog(target_log_4) }
1190            ],
1191        );
1192    }
1193
1194    #[fuchsia::test]
1195    async fn test_symbolized_output() {
1196        let symbolizer = FakeFuchsiaSymbolizer;
1197        let buffers = TestBuffers::default();
1198        let output = JsonWriter::<LogEntry>::new_test(None, &buffers);
1199        let mut formatter = DefaultLogFormatter::new(
1200            LogFilterCriteria::default(),
1201            output,
1202            LogFormatterOptions { ..Default::default() },
1203        );
1204        formatter.set_boot_timestamp(Timestamp::from_nanos(0));
1205        let target_log = LogsDataBuilder::new(diagnostics_data::BuilderArgs {
1206            moniker: "ffx".try_into().unwrap(),
1207            timestamp: Timestamp::from_nanos(0),
1208            component_url: Some("ffx".into()),
1209            severity: Severity::Info,
1210        })
1211        .set_pid(1)
1212        .set_tid(2)
1213        .set_message("Hello world!")
1214        .build();
1215        let (sender, receiver) = zx::Socket::create_stream();
1216        sender
1217            .write(serde_json::to_string(&target_log).unwrap().as_bytes())
1218            .expect("failed to write target log");
1219        drop(sender);
1220        dump_logs_from_socket(flex_client::socket_to_async(receiver), &mut formatter, &symbolizer)
1221            .await
1222            .unwrap();
1223        assert_eq!(buffers.stdout.into_string(), "[00000.000000][1][2][ffx] INFO: Fuchsia\n");
1224    }
1225
1226    #[test]
1227    fn test_device_or_local_timestamp_returns_none_if_now_is_passed() {
1228        assert_matches!(DeviceOrLocalTimestamp::new(Some(&parse_time("now").unwrap()), None), None);
1229    }
1230
1231    struct BrokenPipeWriter;
1232    impl std::io::Write for BrokenPipeWriter {
1233        fn write(&mut self, _buf: &[u8]) -> std::io::Result<usize> {
1234            Err(std::io::Error::new(std::io::ErrorKind::BrokenPipe, "broken pipe"))
1235        }
1236
1237        fn flush(&mut self) -> std::io::Result<()> {
1238            Err(std::io::Error::new(std::io::ErrorKind::BrokenPipe, "broken pipe"))
1239        }
1240    }
1241
1242    impl ToolIO for BrokenPipeWriter {
1243        type OutputItem = LogEntry;
1244        fn is_machine(&self) -> bool {
1245            false
1246        }
1247
1248        fn stderr(&mut self) -> &mut dyn std::io::Write {
1249            self
1250        }
1251
1252        fn item(&mut self, _value: &Self::OutputItem) -> writer::Result<()> {
1253            Err(writer::Error::Io(std::io::Error::new(
1254                std::io::ErrorKind::BrokenPipe,
1255                "broken pipe",
1256            )))
1257        }
1258    }
1259
1260    #[fuchsia::test]
1261    async fn test_default_formatter_exits_on_broken_pipe() {
1262        let stdout = BrokenPipeWriter;
1263        let options = LogFormatterOptions::default();
1264        let mut formatter =
1265            DefaultLogFormatter::new(LogFilterCriteria::default(), stdout, options.clone());
1266        let result = formatter.push_log(log_entry()).await;
1267        assert_matches!(result, Ok(LogProcessingResult::Exit));
1268    }
1269
1270    #[test]
1271    fn test_formatter_error_is_broken_pipe() {
1272        assert!(
1273            FormatterError::IO(std::io::Error::new(std::io::ErrorKind::BrokenPipe, "broken pipe"))
1274                .is_broken_pipe()
1275        );
1276        assert!(!FormatterError::IO(std::io::Error::other("other")).is_broken_pipe());
1277        assert!(!FormatterError::Other(anyhow::anyhow!("other")).is_broken_pipe());
1278    }
1279
1280    #[cfg(not(feature = "fdomain"))]
1281    #[fuchsia::test]
1282    async fn test_json_and_fxt_output_identical() {
1283        use diagnostics_data::ExtendedMoniker;
1284        use diagnostics_log_encoding::encode::{Encoder, EncoderOpts, ResizableBuffer};
1285        use diagnostics_log_encoding::{Argument, Header, LOG_CONTROL_BIT, MONIKER, Record, URL};
1286        use diagnostics_message::MonikerWithUrl;
1287        use flyweights::FlyStr;
1288        use zerocopy::{FromBytes, IntoBytes};
1289
1290        let symbolizer = NoOpSymbolizer {};
1291        let options = LogFormatterOptions::default();
1292
1293        let fn_encode = |record: Record<'_>, tag: u32| -> Vec<u8> {
1294            let mut encoder = Encoder::new(
1295                std::io::Cursor::new(ResizableBuffer::from(Vec::new())),
1296                EncoderOpts::default(),
1297            );
1298            encoder.write_record(record).unwrap();
1299            let mut bytes = encoder.take().into_inner().into_inner();
1300            let mut header = Header::read_from_bytes(&bytes[0..8]).unwrap();
1301            header.set_tag(tag);
1302            bytes[0..8].copy_from_slice(header.as_bytes());
1303            bytes
1304        };
1305
1306        let manifest_bytes = fn_encode(
1307            Record {
1308                timestamp: zx::BootInstant::from_nanos(0),
1309                severity: 0x30, // INFO
1310                arguments: vec![
1311                    Argument::other(MONIKER, "core/foo"),
1312                    Argument::other(URL, "fuchsia-pkg://foo"),
1313                ],
1314            },
1315            1 | LOG_CONTROL_BIT,
1316        );
1317
1318        let log_bytes1 = fn_encode(
1319            Record {
1320                timestamp: zx::BootInstant::from_nanos(123456),
1321                severity: 0x30, // INFO
1322                arguments: vec![
1323                    Argument::pid(zx::Koid::from_raw(1000)),
1324                    Argument::tid(zx::Koid::from_raw(2000)),
1325                    Argument::tag("my_tag"),
1326                    Argument::message("Hello identical world!"),
1327                ],
1328            },
1329            1,
1330        );
1331
1332        let log_bytes2 = fn_encode(
1333            Record {
1334                timestamp: zx::BootInstant::from_nanos(123457),
1335                severity: 0x40, // WARN
1336                arguments: vec![
1337                    Argument::file("src/main.rs"),
1338                    Argument::line(42),
1339                    Argument::dropped(5),
1340                    Argument::message("Warning with source location and dropped count"),
1341                ],
1342            },
1343            1,
1344        );
1345
1346        let log_bytes3_signed = fn_encode(
1347            Record {
1348                timestamp: zx::BootInstant::from_nanos(123458),
1349                severity: 0x50, // ERROR
1350                arguments: vec![
1351                    Argument::message("Error with custom signed int"),
1352                    Argument::new("signed_val", -12345i64),
1353                ],
1354            },
1355            1,
1356        );
1357
1358        let log_bytes3_unsigned = fn_encode(
1359            Record {
1360                timestamp: zx::BootInstant::from_nanos(123458),
1361                severity: 0x50, // ERROR
1362                arguments: vec![
1363                    Argument::message("Error with custom unsigned int"),
1364                    Argument::new("unsigned_val", 67890u64),
1365                ],
1366            },
1367            1,
1368        );
1369
1370        let log_bytes3_bool = fn_encode(
1371            Record {
1372                timestamp: zx::BootInstant::from_nanos(123458),
1373                severity: 0x50, // ERROR
1374                arguments: vec![
1375                    Argument::message("Error with custom boolean"),
1376                    Argument::new("bool_val", true),
1377                ],
1378            },
1379            1,
1380        );
1381
1382        let log_bytes3_str = fn_encode(
1383            Record {
1384                timestamp: zx::BootInstant::from_nanos(123458),
1385                severity: 0x50, // ERROR
1386                arguments: vec![
1387                    Argument::message("Error with custom string"),
1388                    Argument::new("string_val", "custom string"),
1389                ],
1390            },
1391            1,
1392        );
1393
1394        let log_bytes4_pi = fn_encode(
1395            Record {
1396                timestamp: zx::BootInstant::from_nanos(123459),
1397                severity: 0x60, // FATAL
1398                arguments: vec![
1399                    Argument::message("Fatal with float pi"),
1400                    Argument::new("float_pi", std::f64::consts::PI),
1401                ],
1402            },
1403            1,
1404        );
1405
1406        let log_bytes4_zero = fn_encode(
1407            Record {
1408                timestamp: zx::BootInstant::from_nanos(123459),
1409                severity: 0x60, // FATAL
1410                arguments: vec![
1411                    Argument::message("Fatal with float zero"),
1412                    Argument::new("float_zero", 0.0f64),
1413                ],
1414            },
1415            1,
1416        );
1417
1418        let log_bytes4_large = fn_encode(
1419            Record {
1420                timestamp: zx::BootInstant::from_nanos(123459),
1421                severity: 0x60, // FATAL
1422                arguments: vec![
1423                    Argument::message("Fatal with float large"),
1424                    Argument::new("float_large", 123456.789f64),
1425                ],
1426            },
1427            1,
1428        );
1429
1430        let all_records = [
1431            &log_bytes1,
1432            &log_bytes2,
1433            &log_bytes3_signed,
1434            &log_bytes3_unsigned,
1435            &log_bytes3_bool,
1436            &log_bytes3_str,
1437            &log_bytes4_pi,
1438            &log_bytes4_zero,
1439            &log_bytes4_large,
1440        ];
1441
1442        // 1. JSON setup and execution using converted FXT messages
1443        let buffers_json = TestBuffers::default();
1444        let stdout_json = JsonWriter::<LogEntry>::new_test(None, &buffers_json);
1445        let mut formatter_json =
1446            DefaultLogFormatter::new(LogFilterCriteria::default(), stdout_json, options.clone());
1447        formatter_json.set_boot_timestamp(Timestamp::from_nanos(0));
1448
1449        let source = MonikerWithUrl {
1450            moniker: ExtendedMoniker::parse_str("core/foo").unwrap(),
1451            url: FlyStr::new("fuchsia-pkg://foo"),
1452        };
1453
1454        let (sender_json, receiver_json) = zx::Socket::create_stream();
1455        let mut json_stream_bytes = Vec::new();
1456        for record_bytes in &all_records {
1457            let target_log =
1458                diagnostics_message::from_structured(source.clone(), record_bytes).unwrap();
1459            serde_json::to_writer(&mut json_stream_bytes, &target_log).unwrap();
1460            json_stream_bytes.push(b'\n');
1461        }
1462        sender_json.write(&json_stream_bytes).expect("failed to write target logs");
1463        drop(sender_json);
1464
1465        super::dump_logs_from_socket(
1466            flex_client::socket_to_async(receiver_json),
1467            &mut formatter_json,
1468            &symbolizer,
1469            false,
1470        )
1471        .await
1472        .unwrap();
1473
1474        // 2. FXT setup and execution
1475        let buffers_fxt = TestBuffers::default();
1476        let stdout_fxt = JsonWriter::<LogEntry>::new_test(None, &buffers_fxt);
1477        let mut formatter_fxt =
1478            DefaultLogFormatter::new(LogFilterCriteria::default(), stdout_fxt, options.clone());
1479        formatter_fxt.set_boot_timestamp(Timestamp::from_nanos(0));
1480
1481        let (sender_fxt, receiver_fxt) = zx::Socket::create_stream();
1482        sender_fxt.write(&manifest_bytes).unwrap();
1483        for record_bytes in &all_records {
1484            sender_fxt.write(record_bytes).unwrap();
1485        }
1486        drop(sender_fxt);
1487
1488        super::dump_fxt_logs_from_socket(
1489            flex_client::socket_to_async(receiver_fxt),
1490            &mut formatter_fxt,
1491            &symbolizer,
1492            false,
1493        )
1494        .await
1495        .unwrap();
1496
1497        // 3. Verify identity
1498        assert_eq!(buffers_json.stdout.into_string(), buffers_fxt.stdout.into_string());
1499    }
1500
1501    #[fuchsia::test]
1502    async fn test_default_formatter_tail_limit() {
1503        let buffers = TestBuffers::default();
1504        let stdout = JsonWriter::<LogEntry>::new_test(None, &buffers);
1505        let options = LogFormatterOptions { tail: Some(2), ..Default::default() };
1506        let mut formatter = DefaultLogFormatter::new(LogFilterCriteria::default(), stdout, options);
1507
1508        let entry1 = LogEntry {
1509            data: LogData::TargetLog(
1510                logs_data_builder().add_tag("tag1").set_message("msg1").build(),
1511            ),
1512        };
1513        let entry2 = LogEntry {
1514            data: LogData::TargetLog(
1515                logs_data_builder().add_tag("tag1").set_message("msg2").build(),
1516            ),
1517        };
1518        let entry3 = LogEntry {
1519            data: LogData::TargetLog(
1520                logs_data_builder().add_tag("tag1").set_message("msg3").build(),
1521            ),
1522        };
1523
1524        formatter.push_log(entry1).await.unwrap();
1525        formatter.push_log(entry2).await.unwrap();
1526        formatter.push_log(entry3).await.unwrap();
1527
1528        // Before flushing, nothing should be outputted because tail buffering delays output
1529        let stdout_snapshot = buffers.stdout.clone().into_string();
1530        assert_eq!(stdout_snapshot, "");
1531
1532        formatter.flush().await.unwrap();
1533
1534        // After flushing, only the last 2 entries should be outputted
1535        let output = buffers.into_stdout_str();
1536        assert!(!output.contains("msg1"));
1537        assert!(output.contains("msg2"));
1538        assert!(output.contains("msg3"));
1539    }
1540
1541    #[fuchsia::test]
1542    async fn test_default_formatter_sorts_when_enabled() {
1543        let buffers = TestBuffers::default();
1544        let stdout = JsonWriter::<LogEntry>::new_test(None, &buffers);
1545        let options = LogFormatterOptions {
1546            display: Some(LogTextDisplayOptions {
1547                color: LogTextColor::BySeverity,
1548                show_metadata: false,
1549                ..Default::default()
1550            }),
1551            sort: true,
1552            ..Default::default()
1553        };
1554        let mut formatter = DefaultLogFormatter::new(LogFilterCriteria::default(), stdout, options);
1555
1556        formatter
1557            .push_log(LogEntry { data: LogData::TargetLog(make_log_with_timestamp(1_300_000_000)) })
1558            .await
1559            .unwrap();
1560        formatter
1561            .push_log(LogEntry { data: LogData::TargetLog(make_log_with_timestamp(1_100_000_000)) })
1562            .await
1563            .unwrap();
1564        formatter
1565            .push_log(LogEntry { data: LogData::TargetLog(make_log_with_timestamp(1_200_000_000)) })
1566            .await
1567            .unwrap();
1568
1569        assert_eq!(buffers.stdout.clone().into_string(), "");
1570        formatter.flush().await.unwrap();
1571
1572        assert_eq!(
1573            buffers.into_stdout_str(),
1574            concat!(
1575                "[00001.100000][ffx] INFO: Hello world 1100000000!\u{1b}[m\n",
1576                "[00001.200000][ffx] INFO: Hello world 1200000000!\u{1b}[m\n",
1577                "[00001.300000][ffx] INFO: Hello world 1300000000!\u{1b}[m\n",
1578            )
1579        );
1580    }
1581
1582    #[fuchsia::test]
1583    async fn test_default_formatter_sort_window_flushes_progressively() {
1584        let buffers = TestBuffers::default();
1585        let stdout = JsonWriter::<LogEntry>::new_test(None, &buffers);
1586        let options = LogFormatterOptions {
1587            display: Some(LogTextDisplayOptions { show_metadata: false, ..Default::default() }),
1588            sort: true,
1589            ..Default::default()
1590        };
1591        let mut formatter = DefaultLogFormatter::new(LogFilterCriteria::default(), stdout, options);
1592
1593        formatter
1594            .push_log(LogEntry { data: LogData::TargetLog(make_log_with_timestamp(1_200_000_000)) })
1595            .await
1596            .unwrap();
1597        formatter
1598            .push_log(LogEntry { data: LogData::TargetLog(make_log_with_timestamp(1_100_000_000)) })
1599            .await
1600            .unwrap();
1601        for _ in 0..64 {
1602            formatter
1603                .push_log(LogEntry {
1604                    data: LogData::TargetLog(make_log_with_timestamp(1_300_000_000)),
1605                })
1606                .await
1607                .unwrap();
1608        }
1609
1610        // Before flush() is called, the oldest 2 entries beyond the 64-entry reorder window
1611        // should already be emitted in sorted order.
1612        assert_eq!(
1613            buffers.stdout.clone().into_string(),
1614            concat!(
1615                "[00001.100000][ffx] INFO: Hello world 1100000000!\n",
1616                "[00001.200000][ffx] INFO: Hello world 1200000000!\n",
1617            )
1618        );
1619    }
1620
1621    #[fuchsia::test]
1622    async fn test_default_formatter_does_not_reorder_bogus_timestamps() {
1623        let buffers = TestBuffers::default();
1624        let stdout = JsonWriter::<LogEntry>::new_test(None, &buffers);
1625        let options = LogFormatterOptions {
1626            display: Some(LogTextDisplayOptions {
1627                color: LogTextColor::BySeverity,
1628                show_metadata: false,
1629                ..Default::default()
1630            }),
1631            sort: true,
1632            ..Default::default()
1633        };
1634        let mut formatter = DefaultLogFormatter::new(LogFilterCriteria::default(), stdout, options);
1635
1636        for ts in [
1637            10_000_000_000,
1638            999_000_000_000, // far-future outlier (> 1s skew): stay in place
1639            999_100_000_000, // second far-future outlier: stay in place
1640            10_200_000_000,
1641            10_100_000_000, // within 1s skew of 10.2s: reorder before 10.2s
1642            1_000_000_000,  // far-past outlier (> 1s skew): stay in place and highlight
1643            10_300_000_000,
1644        ] {
1645            formatter
1646                .push_log(LogEntry { data: LogData::TargetLog(make_log_with_timestamp(ts)) })
1647                .await
1648                .unwrap();
1649        }
1650        formatter.flush().await.unwrap();
1651
1652        assert_eq!(
1653            buffers.into_stdout_str(),
1654            concat!(
1655                "[00010.000000][ffx] INFO: Hello world 10000000000!\u{1b}[m\n",
1656                "[00999.000000][ffx] INFO: Hello world 999000000000!\u{1b}[m\n",
1657                "[00999.100000][ffx] INFO: Hello world 999100000000!\u{1b}[m\n",
1658                "\u{1b}[38;5;6m[00010.100000]\u{1b}[m[ffx] INFO: Hello world 10100000000!\u{1b}[m\n",
1659                "[00010.200000][ffx] INFO: Hello world 10200000000!\u{1b}[m\n",
1660                "\u{1b}[38;5;6m[00001.000000]\u{1b}[m[ffx] INFO: Hello world 1000000000!\u{1b}[m\n",
1661                "[00010.300000][ffx] INFO: Hello world 10300000000!\u{1b}[m\n",
1662            )
1663        );
1664    }
1665
1666    #[fuchsia::test]
1667    async fn test_default_formatter_sorts_before_tail_limit() {
1668        let buffers = TestBuffers::default();
1669        let stdout = JsonWriter::<LogEntry>::new_test(None, &buffers);
1670        let options = LogFormatterOptions {
1671            display: Some(LogTextDisplayOptions { show_metadata: false, ..Default::default() }),
1672            sort: true,
1673            tail: Some(2),
1674            ..Default::default()
1675        };
1676        let mut formatter = DefaultLogFormatter::new(LogFilterCriteria::default(), stdout, options);
1677
1678        formatter
1679            .push_log(LogEntry { data: LogData::TargetLog(make_log_with_timestamp(1_300_000_000)) })
1680            .await
1681            .unwrap();
1682        formatter
1683            .push_log(LogEntry { data: LogData::TargetLog(make_log_with_timestamp(1_100_000_000)) })
1684            .await
1685            .unwrap();
1686        formatter
1687            .push_log(LogEntry { data: LogData::TargetLog(make_log_with_timestamp(1_200_000_000)) })
1688            .await
1689            .unwrap();
1690
1691        formatter.flush().await.unwrap();
1692
1693        assert_eq!(
1694            buffers.into_stdout_str(),
1695            concat!(
1696                "[00001.200000][ffx] INFO: Hello world 1200000000!\n",
1697                "[00001.300000][ffx] INFO: Hello world 1300000000!\n",
1698            )
1699        );
1700    }
1701
1702    #[fuchsia::test]
1703    async fn test_default_formatter_highlights_out_of_order_timestamps() {
1704        let buffers = TestBuffers::default();
1705        let stdout = JsonWriter::<LogEntry>::new_test(None, &buffers);
1706        let options = LogFormatterOptions {
1707            display: Some(LogTextDisplayOptions {
1708                color: LogTextColor::BySeverity,
1709                show_metadata: false,
1710                ..Default::default()
1711            }),
1712            sort: false,
1713            ..Default::default()
1714        };
1715        let mut formatter = DefaultLogFormatter::new(LogFilterCriteria::default(), stdout, options);
1716
1717        for ts in [
1718            1_300_000_000,
1719            1_100_000_000,
1720            1_200_000_000,
1721            1_400_000_000,
1722            999_000_000_000, // far-future outlier #1
1723            999_100_000_000, // far-future outlier #2
1724            5_500_000_000, // normal timestamp (< 999.1s, highlights transition then self-corrects)
1725            5_400_000_000, // out-of-order relative to 5.5s
1726            5_600_000_000, // in-order relative to 5.4s
1727        ] {
1728            formatter
1729                .push_log(LogEntry { data: LogData::TargetLog(make_log_with_timestamp(ts)) })
1730                .await
1731                .unwrap();
1732        }
1733
1734        // Rebooting resets state so a subsequent 1.0s log is not marked out-of-order.
1735        formatter.set_boot_timestamp(Timestamp::from_nanos(0));
1736        formatter
1737            .push_log(LogEntry { data: LogData::TargetLog(make_log_with_timestamp(1_000_000_000)) })
1738            .await
1739            .unwrap();
1740
1741        assert_eq!(
1742            buffers.into_stdout_str(),
1743            concat!(
1744                "[00001.300000][ffx] INFO: Hello world 1300000000!\u{1b}[m\n",
1745                "\u{1b}[38;5;6m[00001.100000]\u{1b}[m[ffx] INFO: Hello world 1100000000!\u{1b}[m\n",
1746                "[00001.200000][ffx] INFO: Hello world 1200000000!\u{1b}[m\n",
1747                "[00001.400000][ffx] INFO: Hello world 1400000000!\u{1b}[m\n",
1748                "[00999.000000][ffx] INFO: Hello world 999000000000!\u{1b}[m\n",
1749                "[00999.100000][ffx] INFO: Hello world 999100000000!\u{1b}[m\n",
1750                "\u{1b}[38;5;6m[00005.500000]\u{1b}[m[ffx] INFO: Hello world 5500000000!\u{1b}[m\n",
1751                "\u{1b}[38;5;6m[00005.400000]\u{1b}[m[ffx] INFO: Hello world 5400000000!\u{1b}[m\n",
1752                "[00005.600000][ffx] INFO: Hello world 5600000000!\u{1b}[m\n",
1753                "[00001.000000][ffx] INFO: Hello world 1000000000!\u{1b}[m\n",
1754            )
1755        );
1756    }
1757}
log_command/log_formatter.rs

log_command/
log_formatter.rs
