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

log_command_ctf/
log_formatter.rs
