diagnostics_reader/
lib.rs

1// Copyright 2020 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
5#![deny(missing_docs)]
6
7//! A library for reading Inspect and Log data from
8//! the ArchiveAccessor FIDL protocol.
9
10use async_stream::stream;
11use diagnostics_data::{DiagnosticsData, LogsData};
12#[cfg(fuchsia_api_level_less_than = "HEAD")]
13use diagnostics_message as _;
14#[cfg(fuchsia_api_level_at_least = "HEAD")]
15use diagnostics_message::{RustMessageFormatter, from_extended_record};
16use fidl_fuchsia_diagnostics::{
17    ArchiveAccessorMarker, ArchiveAccessorProxy, BatchIteratorMarker, BatchIteratorProxy,
18    ClientSelectorConfiguration, Format, FormattedContent, PerformanceConfiguration, ReaderError,
19    Selector, SelectorArgument, StreamMode, StreamParameters,
20};
21use fuchsia_async::{self as fasync, DurationExt, TimeoutExt};
22use fuchsia_component::client;
23#[cfg(fuchsia_api_level_at_least = "HEAD")]
24use fuchsia_sync::Mutex;
25use futures::channel::mpsc;
26use futures::prelude::*;
27use futures::sink::SinkExt;
28use futures::stream::FusedStream;
29use pin_project::pin_project;
30use serde::Deserialize;
31use std::future::ready;
32use std::marker::PhantomData;
33use std::pin::Pin;
34use std::sync::Arc;
35use std::task::{Context, Poll};
36use thiserror::Error;
37use zx::{self as zx, MonotonicDuration};
38
39/// Alias for ArchiveReader<Logs>. Used for reading logs.
40pub type LogsArchiveReader = ArchiveReader<Logs>;
41
42/// Alias for ArchiveReader<Inspect>. Used for reading inspect.
43pub type InspectArchiveReader = ArchiveReader<Inspect>;
44
45pub use diagnostics_data::{Data, Inspect, Logs, Severity};
46pub use diagnostics_hierarchy::{DiagnosticsHierarchy, Property, hierarchy};
47
48const RETRY_DELAY_MS: i64 = 300;
49
50#[cfg(fuchsia_api_level_at_least = "HEAD")]
51const FORMAT: Format = Format::Cbor;
52#[cfg(fuchsia_api_level_less_than = "HEAD")]
53const FORMAT: Format = Format::Json;
54
55/// Errors that this library can return
56#[derive(Debug, Error)]
57pub enum Error {
58    /// Failed to connect to the archive accessor
59    #[error("Failed to connect to the archive accessor")]
60    ConnectToArchive(#[source] anyhow::Error),
61
62    /// Failed to create the BatchIterator channel ends
63    #[error("Failed to create the BatchIterator channel ends")]
64    CreateIteratorProxy(#[source] fidl::Error),
65
66    /// Failed to stream diagnostics from the accessor
67    #[error("Failed to stream diagnostics from the accessor")]
68    StreamDiagnostics(#[source] fidl::Error),
69
70    /// Failed to call iterator server
71    #[error("Failed to call iterator server")]
72    GetNextCall(#[source] fidl::Error),
73
74    /// Received error from the GetNext response
75    #[error("Received error from the GetNext response: {0:?}")]
76    GetNextReaderError(ReaderError),
77
78    /// Failed to read json received
79    #[error("Failed to read json received")]
80    ReadJson(#[source] serde_json::Error),
81
82    /// Failed to read cbor received
83    #[cfg(fuchsia_api_level_at_least = "HEAD")]
84    #[error("Failed to read cbor received")]
85    ReadCbor(#[source] anyhow::Error),
86
87    /// Failed to parse the diagnostics data from the json received
88    #[error("Failed to parse the diagnostics data from the json received")]
89    ParseDiagnosticsData(#[source] serde_json::Error),
90
91    /// Failed to read vmo from the response
92    #[error("Failed to read vmo from the response")]
93    ReadVmo(#[source] zx::Status),
94}
95
96/// An inspect tree selector for a component.
97pub struct ComponentSelector {
98    moniker: Vec<String>,
99    tree_selectors: Vec<String>,
100}
101
102impl ComponentSelector {
103    /// Create a new component event selector.
104    /// By default it will select the whole tree unless tree selectors are provided.
105    /// `moniker` is the realm path relative to the realm of the running component plus the
106    /// component name. For example: [a, b, component].
107    pub fn new(moniker: Vec<String>) -> Self {
108        Self { moniker, tree_selectors: Vec::new() }
109    }
110
111    /// Select a section of the inspect tree.
112    pub fn with_tree_selector(mut self, tree_selector: impl Into<String>) -> Self {
113        self.tree_selectors.push(tree_selector.into());
114        self
115    }
116
117    fn moniker_str(&self) -> String {
118        self.moniker.join("/")
119    }
120}
121
122/// Trait used for things that can be converted to selector arguments.
123pub trait ToSelectorArguments {
124    /// Converts this to selector arguments.
125    fn to_selector_arguments(self) -> Box<dyn Iterator<Item = SelectorArgument>>;
126}
127
128/// Trait used for things that can be converted to component selector arguments.
129pub trait ToComponentSelectorArguments {
130    /// Converts this to selector arguments.
131    fn to_component_selector_arguments(self) -> ComponentSelector;
132}
133
134impl ToComponentSelectorArguments for &str {
135    fn to_component_selector_arguments(self) -> ComponentSelector {
136        if self.contains("\\:") {
137            // String is already escaped, don't escape it.
138            ComponentSelector::new(self.split("/").map(|value| value.to_string()).collect())
139        } else {
140            // String isn't escaped, escape it
141            ComponentSelector::new(
142                selectors::sanitize_moniker_for_selectors(self)
143                    .split("/")
144                    .map(|value| value.to_string())
145                    .collect(),
146            )
147            .with_tree_selector("[...]root")
148        }
149    }
150}
151
152impl ToComponentSelectorArguments for String {
153    fn to_component_selector_arguments(self) -> ComponentSelector {
154        self.as_str().to_component_selector_arguments()
155    }
156}
157
158impl ToComponentSelectorArguments for ComponentSelector {
159    fn to_component_selector_arguments(self) -> ComponentSelector {
160        self
161    }
162}
163
164impl ToSelectorArguments for String {
165    fn to_selector_arguments(self) -> Box<dyn Iterator<Item = SelectorArgument>> {
166        Box::new([SelectorArgument::RawSelector(self)].into_iter())
167    }
168}
169
170impl ToSelectorArguments for &str {
171    fn to_selector_arguments(self) -> Box<dyn Iterator<Item = SelectorArgument>> {
172        Box::new([SelectorArgument::RawSelector(self.to_string())].into_iter())
173    }
174}
175
176impl ToSelectorArguments for ComponentSelector {
177    fn to_selector_arguments(self) -> Box<dyn Iterator<Item = SelectorArgument>> {
178        let moniker = self.moniker_str();
179        // If not tree selectors were provided, select the full tree.
180        if self.tree_selectors.is_empty() {
181            Box::new([SelectorArgument::RawSelector(format!("{moniker}:root"))].into_iter())
182        } else {
183            Box::new(
184                self.tree_selectors
185                    .into_iter()
186                    .map(move |s| SelectorArgument::RawSelector(format!("{moniker}:{s}"))),
187            )
188        }
189    }
190}
191
192impl ToSelectorArguments for Selector {
193    fn to_selector_arguments(self) -> Box<dyn Iterator<Item = SelectorArgument>> {
194        Box::new([SelectorArgument::StructuredSelector(self)].into_iter())
195    }
196}
197
198/// Before unsealing this, consider whether your code belongs in this file.
199pub trait SerializableValue: private::Sealed {
200    /// The Format of this SerializableValue. Either Logs or Inspect.
201    const FORMAT_OF_VALUE: Format;
202}
203
204/// Trait used to verify that a JSON payload has a valid diagnostics payload.
205pub trait CheckResponse: private::Sealed {
206    /// Returns true if the response has a valid payload.
207    fn has_payload(&self) -> bool;
208}
209
210// The "sealed trait" pattern.
211//
212// https://rust-lang.github.io/api-guidelines/future-proofing.html
213mod private {
214    pub trait Sealed {}
215}
216impl private::Sealed for serde_json::Value {}
217impl private::Sealed for ciborium::Value {}
218impl<D: DiagnosticsData> private::Sealed for Data<D> {}
219
220impl<D: DiagnosticsData> CheckResponse for Data<D> {
221    fn has_payload(&self) -> bool {
222        self.payload.is_some()
223    }
224}
225
226impl SerializableValue for serde_json::Value {
227    const FORMAT_OF_VALUE: Format = Format::Json;
228}
229
230impl CheckResponse for serde_json::Value {
231    fn has_payload(&self) -> bool {
232        match self {
233            serde_json::Value::Object(obj) => {
234                obj.get("payload").map(|p| !matches!(p, serde_json::Value::Null)).is_some()
235            }
236            _ => false,
237        }
238    }
239}
240
241#[cfg(fuchsia_api_level_at_least = "HEAD")]
242impl SerializableValue for ciborium::Value {
243    const FORMAT_OF_VALUE: Format = Format::Cbor;
244}
245
246impl CheckResponse for ciborium::Value {
247    fn has_payload(&self) -> bool {
248        match self {
249            ciborium::Value::Map(m) => {
250                let payload_key = ciborium::Value::Text("payload".into());
251                m.iter().any(|(key, _)| *key == payload_key)
252            }
253            _ => false,
254        }
255    }
256}
257
258/// Retry configuration for ArchiveReader
259#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
260pub enum RetryConfig {
261    /// The minimum schema count required for a successful read.
262    /// This guarantees that a read will contain at least MinSchemaCount
263    /// results.
264    MinSchemaCount(usize),
265}
266
267impl RetryConfig {
268    /// Always retry
269    pub fn always() -> Self {
270        Self::MinSchemaCount(1)
271    }
272
273    /// Never retry
274    pub fn never() -> Self {
275        Self::MinSchemaCount(0)
276    }
277
278    /// Retry result_count times
279    fn should_retry(&self, result_count: usize) -> bool {
280        match self {
281            Self::MinSchemaCount(min) => *min > result_count,
282        }
283    }
284}
285
286/// A trait representing a type of diagnostics data.
287pub trait DiagnosticsDataType: private::Sealed {}
288
289impl private::Sealed for Logs {}
290
291impl private::Sealed for Inspect {}
292
293impl DiagnosticsDataType for Logs {}
294
295impl DiagnosticsDataType for Inspect {}
296
297/// Utility for reading inspect data of a running component using the injected Archive
298/// Reader service.
299pub struct ArchiveReader<T> {
300    archive: Option<ArchiveAccessorProxy>,
301    selectors: Vec<SelectorArgument>,
302    retry_config: RetryConfig,
303    timeout: Option<MonotonicDuration>,
304    batch_retrieval_timeout_seconds: Option<i64>,
305    max_aggregated_content_size_bytes: Option<u64>,
306    format: Option<Format>,
307    _phantom: PhantomData<T>,
308}
309
310impl<T: DiagnosticsDataType> ArchiveReader<T> {
311    /// Initializes the ArchiveReader with a custom connection to an ArchiveAccessor.
312    /// By default, the connection will be initialized by connecting to
313    /// fuchsia.diagnostics.ArchiveAccessor
314    pub fn with_archive(&mut self, archive: ArchiveAccessorProxy) -> &mut Self {
315        self.archive = Some(archive);
316        self
317    }
318
319    /// Sets the minimum number of schemas expected in a result in order for the
320    /// result to be considered a success.
321    pub fn with_minimum_schema_count(&mut self, minimum_schema_count: usize) -> &mut Self {
322        self.retry_config = RetryConfig::MinSchemaCount(minimum_schema_count);
323        self
324    }
325
326    /// Sets a custom retry configuration. By default we always retry.
327    pub fn retry(&mut self, config: RetryConfig) -> &mut Self {
328        self.retry_config = config;
329        self
330    }
331
332    /// Sets the maximum time to wait for a response from the Archive.
333    /// Do not use in tests unless timeout is the expected behavior.
334    pub fn with_timeout(&mut self, duration: MonotonicDuration) -> &mut Self {
335        self.timeout = Some(duration);
336        self
337    }
338
339    /// Filters logs for a specific component or component selector.
340    /// If string input, the string may be either a component selector string
341    /// or a moniker, or a ComponentSelector may be passed directly.
342    pub fn select_all_for_component(
343        &mut self,
344        component: impl ToComponentSelectorArguments,
345    ) -> &mut Self {
346        self.selectors.extend(component.to_component_selector_arguments().to_selector_arguments());
347        self
348    }
349
350    /// Connects to the ArchiveAccessor and returns data matching provided selectors.
351    async fn snapshot_shared<D>(&self, format: Format) -> Result<Vec<Data<D>>, Error>
352    where
353        D: DiagnosticsData + 'static,
354    {
355        let data_future = self.snapshot_inner::<D, Data<D>>(format);
356        let data = match self.timeout {
357            Some(timeout) => data_future.on_timeout(timeout.after_now(), || Ok(Vec::new())).await?,
358            None => data_future.await?,
359        };
360        Ok(data)
361    }
362
363    async fn snapshot_inner<D, Y>(&self, format: Format) -> Result<Vec<Y>, Error>
364    where
365        D: DiagnosticsData,
366        Y: for<'a> Deserialize<'a> + CheckResponse + Send + 'static,
367    {
368        loop {
369            let iterator = self.batch_iterator::<D>(StreamMode::Snapshot, format)?;
370            let result = drain_batch_iterator::<Y>(Arc::new(iterator))
371                .filter_map(|value| ready(value.ok()))
372                .collect::<Vec<_>>()
373                .await;
374
375            if self.retry_config.should_retry(result.len()) {
376                fasync::Timer::new(fasync::MonotonicInstant::after(
377                    zx::MonotonicDuration::from_millis(RETRY_DELAY_MS),
378                ))
379                .await;
380            } else {
381                return Ok(result);
382            }
383        }
384    }
385
386    fn batch_iterator<D>(
387        &self,
388        mode: StreamMode,
389        format: Format,
390    ) -> Result<BatchIteratorProxy, Error>
391    where
392        D: DiagnosticsData,
393    {
394        let archive = match &self.archive {
395            Some(archive) => archive.clone(),
396            None => client::connect_to_protocol::<ArchiveAccessorMarker>()
397                .map_err(Error::ConnectToArchive)?,
398        };
399
400        let (iterator, server_end) = fidl::endpoints::create_proxy::<BatchIteratorMarker>();
401        let stream_parameters = StreamParameters {
402            stream_mode: Some(mode),
403            data_type: Some(D::DATA_TYPE),
404            format: Some(format),
405            client_selector_configuration: if self.selectors.is_empty() {
406                Some(ClientSelectorConfiguration::SelectAll(true))
407            } else {
408                Some(ClientSelectorConfiguration::Selectors(self.selectors.to_vec()))
409            },
410            performance_configuration: Some(PerformanceConfiguration {
411                max_aggregate_content_size_bytes: self.max_aggregated_content_size_bytes,
412                batch_retrieval_timeout_seconds: self.batch_retrieval_timeout_seconds,
413                ..Default::default()
414            }),
415            ..Default::default()
416        };
417
418        archive
419            .stream_diagnostics(&stream_parameters, server_end)
420            .map_err(Error::StreamDiagnostics)?;
421        Ok(iterator)
422    }
423}
424
425impl ArchiveReader<Logs> {
426    /// Creates an ArchiveReader for reading logs
427    pub fn logs() -> Self {
428        ArchiveReader::<Logs> {
429            timeout: None,
430            format: None,
431            selectors: vec![],
432            retry_config: RetryConfig::always(),
433            archive: None,
434            batch_retrieval_timeout_seconds: None,
435            max_aggregated_content_size_bytes: None,
436            _phantom: PhantomData,
437        }
438    }
439
440    #[doc(hidden)]
441    pub fn with_format(&mut self, format: Format) -> &mut Self {
442        self.format = Some(format);
443        self
444    }
445
446    #[inline]
447    fn format(&self) -> Format {
448        match self.format {
449            Some(f) => f,
450            None => {
451                #[cfg(fuchsia_api_level_at_least = "HEAD")]
452                let ret = Format::LegacyFxt;
453                #[cfg(fuchsia_api_level_less_than = "HEAD")]
454                let ret = Format::Json;
455                ret
456            }
457        }
458    }
459
460    /// Connects to the ArchiveAccessor and returns data matching provided selectors.
461    pub async fn snapshot(&self) -> Result<Vec<Data<Logs>>, Error> {
462        let fut = async {
463            loop {
464                let iterator = self.batch_iterator::<Logs>(StreamMode::Snapshot, self.format())?;
465                let result = drain_batch_iterator_for_logs(Arc::new(iterator), Some(self.format()))
466                    .filter_map(|value| ready(value.ok()))
467                    .collect::<Vec<_>>()
468                    .await;
469                if self.retry_config.should_retry(result.len()) {
470                    fasync::Timer::new(fasync::MonotonicInstant::after(
471                        zx::MonotonicDuration::from_millis(RETRY_DELAY_MS),
472                    ))
473                    .await;
474                } else {
475                    return Ok(result);
476                }
477            }
478        };
479        match self.timeout {
480            Some(timeout) => fut.on_timeout(timeout.after_now(), || Ok(Vec::new())).await,
481            None => fut.await,
482        }
483    }
484
485    /// Connects to the ArchiveAccessor and returns a stream of data containing a snapshot of the
486    /// current buffer in the Archivist as well as new data that arrives.
487    pub fn snapshot_then_subscribe(&self) -> Result<Subscription, Error> {
488        let iterator =
489            self.batch_iterator::<Logs>(StreamMode::SnapshotThenSubscribe, self.format())?;
490        Ok(Subscription::new_with_format(iterator, self.format()))
491    }
492}
493
494impl ArchiveReader<Inspect> {
495    /// Creates an ArchiveReader for reading Inspect data.
496    pub fn inspect() -> Self {
497        ArchiveReader::<Inspect> {
498            timeout: None,
499            format: None,
500            selectors: vec![],
501            retry_config: RetryConfig::always(),
502            archive: None,
503            batch_retrieval_timeout_seconds: None,
504            max_aggregated_content_size_bytes: None,
505            _phantom: PhantomData,
506        }
507    }
508
509    /// Set the maximum time to wait for a wait for a single component
510    /// to have its diagnostics data "pumped".
511    pub fn with_batch_retrieval_timeout_seconds(&mut self, timeout: i64) -> &mut Self {
512        self.batch_retrieval_timeout_seconds = Some(timeout);
513        self
514    }
515
516    /// Sets the total number of bytes allowed in a single VMO read.
517    pub fn with_aggregated_result_bytes_limit(&mut self, limit_bytes: u64) -> &mut Self {
518        self.max_aggregated_content_size_bytes = Some(limit_bytes);
519        self
520    }
521
522    /// Connects to the ArchiveAccessor and returns inspect data matching provided selectors.
523    /// Returns the raw json for each hierarchy fetched. This is used for CTF compatibility
524    /// tests (which test various implementation details of the JSON format),
525    /// and use beyond such tests is discouraged.
526    pub async fn snapshot_raw<T>(&self) -> Result<T, Error>
527    where
528        T: for<'a> Deserialize<'a>
529            + SerializableValue
530            + From<Vec<T>>
531            + CheckResponse
532            + 'static
533            + Send,
534    {
535        let data_future = self.snapshot_inner::<Inspect, T>(T::FORMAT_OF_VALUE);
536        let data = match self.timeout {
537            Some(timeout) => data_future.on_timeout(timeout.after_now(), || Ok(Vec::new())).await?,
538            None => data_future.await?,
539        };
540        Ok(T::from(data))
541    }
542
543    /// Adds selectors used for performing filtering inspect hierarchies.
544    /// This may be called multiple times to add additional selectors.
545    pub fn add_selectors<T, S>(&mut self, selectors: T) -> &mut Self
546    where
547        T: Iterator<Item = S>,
548        S: ToSelectorArguments,
549    {
550        for selector in selectors {
551            self.add_selector(selector);
552        }
553        self
554    }
555
556    /// Requests a single component tree (or sub-tree).
557    pub fn add_selector(&mut self, selector: impl ToSelectorArguments) -> &mut Self {
558        self.selectors.extend(selector.to_selector_arguments());
559        self
560    }
561
562    /// Sets the format to use when reading inspect data.
563    pub fn with_format(&mut self, format: Format) -> &mut Self {
564        self.format = Some(format);
565        self
566    }
567
568    #[inline]
569    fn format(&self) -> Format {
570        match self.format {
571            Some(f) => f,
572            None => FORMAT,
573        }
574    }
575
576    /// Connects to the ArchiveAccessor and returns data matching provided selectors.
577    pub async fn snapshot(&self) -> Result<Vec<Data<Inspect>>, Error> {
578        self.snapshot_shared::<Inspect>(self.format()).await
579    }
580}
581
582#[derive(Debug, Deserialize)]
583#[serde(untagged)]
584enum OneOrMany<T> {
585    Many(Vec<T>),
586    One(T),
587}
588
589fn stream_batch<T>(
590    iterator: Arc<BatchIteratorProxy>,
591    process_content: impl Fn(FormattedContent) -> Result<OneOrMany<T>, Error>,
592) -> impl Stream<Item = Result<T, Error>>
593where
594    T: for<'a> Deserialize<'a> + Send + 'static,
595{
596    stream! {
597        loop {
598            let next_batch = iterator
599                .get_next()
600                .await
601                .map_err(Error::GetNextCall)?
602                .map_err(Error::GetNextReaderError)?;
603            if next_batch.is_empty() {
604                // End of stream
605                return;
606            }
607            for formatted_content in next_batch {
608                let output = process_content(formatted_content)?;
609                match output {
610                    OneOrMany::One(data) => yield Ok(data),
611                    OneOrMany::Many(datas) => {
612                        for data in datas {
613                            yield Ok(data);
614                        }
615                    }
616                }
617            }
618        }
619    }
620}
621
622/// Drain a batch iterator.
623pub fn drain_batch_iterator<T>(
624    iterator: Arc<BatchIteratorProxy>,
625) -> impl Stream<Item = Result<T, Error>>
626where
627    T: for<'a> Deserialize<'a> + Send + 'static,
628{
629    stream_batch(iterator, |formatted_content| match formatted_content {
630        FormattedContent::Json(data) => {
631            let mut buf = vec![0; data.size as usize];
632            data.vmo.read(&mut buf, 0).map_err(Error::ReadVmo)?;
633            serde_json::from_slice(&buf).map_err(Error::ReadJson)
634        }
635        #[cfg(fuchsia_api_level_at_least = "HEAD")]
636        FormattedContent::Cbor(vmo) => {
637            let mut buf = vec![0; vmo.get_content_size().expect("Always returns Ok") as usize];
638            vmo.read(&mut buf, 0).map_err(Error::ReadVmo)?;
639            Ok(ciborium::from_reader(buf.as_slice()).map_err(|err| Error::ReadCbor(err.into()))?)
640        }
641        #[cfg(fuchsia_api_level_at_least = "HEAD")]
642        FormattedContent::Fxt(_) => unreachable!("We never expect FXT for Inspect"),
643        FormattedContent::__SourceBreaking { unknown_ordinal: _ } => {
644            unreachable!("Received unrecognized FIDL message")
645        }
646    })
647}
648
649fn drain_batch_iterator_for_logs(
650    iterator: Arc<BatchIteratorProxy>,
651    _format: Option<Format>,
652) -> impl Stream<Item = Result<LogsData, Error>> {
653    #[cfg(fuchsia_api_level_at_least = "HEAD")]
654    let parser = Arc::new(Mutex::new(diagnostics_message::MessageParser::default()));
655    stream_batch::<LogsData>(iterator, move |formatted_content| match formatted_content {
656        FormattedContent::Json(data) => {
657            let mut buf = vec![0; data.size as usize];
658            data.vmo.read(&mut buf, 0).map_err(Error::ReadVmo)?;
659            serde_json::from_slice(&buf).map_err(Error::ReadJson)
660        }
661        #[cfg(fuchsia_api_level_at_least = "HEAD")]
662        FormattedContent::Fxt(vmo) => {
663            let mut buf = vec![0; vmo.get_content_size().expect("Always returns Ok") as usize];
664            vmo.read(&mut buf, 0).map_err(Error::ReadVmo)?;
665            let mut current_slice: &[u8] = &buf;
666            let mut items = vec![];
667            let mut parser = parser.lock();
668
669            while !current_slice.is_empty() {
670                if _format == Some(Format::Fxt) {
671                    match parser.parse_next(current_slice, RustMessageFormatter) {
672                        Ok((maybe_data, remaining)) => {
673                            assert!(remaining.len() < current_slice.len(), "Parser must advance");
674                            if let Some(data) = maybe_data {
675                                items.push(data);
676                            }
677                            current_slice = remaining;
678                        }
679                        Err(_) => {
680                            // This can happen if we are reading a truncated record.
681                            // Stop parsing this buffer.
682                            break;
683                        }
684                    }
685                } else {
686                    match from_extended_record(current_slice) {
687                        Ok((data, remaining)) => {
688                            items.push(data);
689                            current_slice = remaining;
690                        }
691                        Err(_) => {
692                            // This can happen if we are reading a truncated record.
693                            // Stop parsing this buffer.
694                            break;
695                        }
696                    }
697                }
698            }
699            Ok(OneOrMany::Many(items))
700        }
701        #[cfg(fuchsia_api_level_at_least = "HEAD")]
702        FormattedContent::Cbor(_) => unreachable!("We never expect CBOR"),
703        FormattedContent::__SourceBreaking { unknown_ordinal: _ } => {
704            unreachable!("Received unrecognized FIDL message")
705        }
706    })
707}
708
709/// A subscription used for reading logs.
710#[pin_project]
711pub struct Subscription {
712    #[pin]
713    recv: Pin<Box<dyn FusedStream<Item = Result<LogsData, Error>> + Send>>,
714    iterator: Arc<BatchIteratorProxy>,
715}
716
717const DATA_CHANNEL_SIZE: usize = 32;
718const ERROR_CHANNEL_SIZE: usize = 2;
719
720impl Subscription {
721    /// Creates a new subscription stream to a batch iterator.
722    /// The stream will return diagnostics data structures.
723    pub fn new(iterator: BatchIteratorProxy) -> Self {
724        let iterator = Arc::new(iterator);
725        Subscription {
726            recv: Box::pin(drain_batch_iterator_for_logs(iterator.clone(), None).fuse()),
727            iterator,
728        }
729    }
730
731    /// Creates a new subscription stream to a batch iterator.
732    /// The stream will return diagnostics data structures.
733    pub fn new_with_format(iterator: BatchIteratorProxy, format: Format) -> Self {
734        let iterator = Arc::new(iterator);
735        Subscription {
736            recv: Box::pin(drain_batch_iterator_for_logs(iterator.clone(), Some(format)).fuse()),
737            iterator,
738        }
739    }
740
741    /// Wait for the connection with the server to be established.
742    pub async fn wait_for_ready(&self) {
743        self.iterator.wait_for_ready().await.expect("doesn't disconnect");
744    }
745
746    /// Splits the subscription into two separate streams: results and errors.
747    pub fn split_streams(mut self) -> (SubscriptionResultsStream<LogsData>, mpsc::Receiver<Error>) {
748        let (mut errors_sender, errors) = mpsc::channel(ERROR_CHANNEL_SIZE);
749        let (mut results_sender, recv) = mpsc::channel(DATA_CHANNEL_SIZE);
750        let _drain_task = fasync::Task::spawn(async move {
751            while let Some(result) = self.next().await {
752                match result {
753                    Ok(value) => {
754                        if results_sender.send(value).await.is_err() {
755                            break;
756                        }
757                    }
758                    Err(e) => {
759                        // Use try_send so that if the error receiver is full or unpolled,
760                        // data stream processing does not deadlock.
761                        let _ = errors_sender.try_send(e);
762                    }
763                };
764            }
765        });
766        (SubscriptionResultsStream { recv, _drain_task }, errors)
767    }
768}
769
770impl Stream for Subscription {
771    type Item = Result<LogsData, Error>;
772
773    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
774        let this = self.project();
775        this.recv.poll_next(cx)
776    }
777}
778
779impl FusedStream for Subscription {
780    fn is_terminated(&self) -> bool {
781        self.recv.is_terminated()
782    }
783}
784
785/// A stream for reading diagnostics data
786#[pin_project]
787pub struct SubscriptionResultsStream<T> {
788    #[pin]
789    recv: mpsc::Receiver<T>,
790    _drain_task: fasync::Task<()>,
791}
792
793impl<T> Stream for SubscriptionResultsStream<T>
794where
795    T: for<'a> Deserialize<'a>,
796{
797    type Item = T;
798
799    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
800        let this = self.project();
801        this.recv.poll_next(cx)
802    }
803}
804
805impl<T> FusedStream for SubscriptionResultsStream<T>
806where
807    T: for<'a> Deserialize<'a>,
808{
809    fn is_terminated(&self) -> bool {
810        self.recv.is_terminated()
811    }
812}
813
814#[cfg(test)]
815mod tests {
816    use super::*;
817    use assert_matches::assert_matches;
818    use diagnostics_assertions::assert_data_tree;
819    use diagnostics_log::{Publisher, PublisherOptions};
820    use fidl::endpoints::ServerEnd;
821    use fidl_fuchsia_diagnostics as fdiagnostics;
822    use fuchsia_component_test::{
823        Capability, ChildOptions, RealmBuilder, RealmInstance, Ref, Route,
824    };
825    use futures::TryStreamExt;
826    use log::{error, info};
827
828    const TEST_COMPONENT_URL: &str = "#meta/inspect_test_component.cm";
829
830    struct ComponentOptions {
831        publish_n_trees: u64,
832    }
833
834    async fn start_component(opts: ComponentOptions) -> Result<RealmInstance, anyhow::Error> {
835        let builder = RealmBuilder::new().await?;
836        let test_component = builder
837            .add_child("test_component", TEST_COMPONENT_URL, ChildOptions::new().eager())
838            .await?;
839        builder
840            .add_route(
841                Route::new()
842                    .capability(Capability::protocol_by_name("fuchsia.logger.LogSink"))
843                    .from(Ref::parent())
844                    .to(&test_component),
845            )
846            .await?;
847        builder.init_mutable_config_to_empty(&test_component).await.unwrap();
848        builder
849            .set_config_value(&test_component, "publish_n_trees", opts.publish_n_trees.into())
850            .await
851            .unwrap();
852        let instance = builder.build().await?;
853        Ok(instance)
854    }
855
856    // All selectors in this test select against all tree names, in order to ensure the expected
857    // number of trees are published
858    #[fuchsia::test]
859    async fn inspect_data_for_component() -> Result<(), anyhow::Error> {
860        let instance = start_component(ComponentOptions { publish_n_trees: 1 }).await?;
861        let moniker = format!("realm_builder:{}/test_component", instance.root.child_name());
862        let component_selector = selectors::sanitize_moniker_for_selectors(&moniker);
863        let results = ArchiveReader::inspect()
864            .add_selector(format!("{component_selector}:[...]root"))
865            .snapshot()
866            .await?;
867        assert_eq!(results.len(), 1);
868        assert_data_tree!(results[0].payload.as_ref().unwrap(), root: {
869            "tree-0": 0u64,
870            int: 3u64,
871            "lazy-node": {
872                a: "test",
873                child: {
874                    double: 3.25,
875                },
876            }
877        });
878        // add_selector can take either a String or a Selector.
879        let lazy_property_selector = Selector {
880            component_selector: Some(fdiagnostics::ComponentSelector {
881                moniker_segments: Some(vec![
882                    fdiagnostics::StringSelector::ExactMatch(format!(
883                        "realm_builder:{}",
884                        instance.root.child_name()
885                    )),
886                    fdiagnostics::StringSelector::ExactMatch("test_component".into()),
887                ]),
888                ..Default::default()
889            }),
890            tree_selector: Some(fdiagnostics::TreeSelector::PropertySelector(
891                fdiagnostics::PropertySelector {
892                    node_path: vec![
893                        fdiagnostics::StringSelector::ExactMatch("root".into()),
894                        fdiagnostics::StringSelector::ExactMatch("lazy-node".into()),
895                    ],
896                    target_properties: fdiagnostics::StringSelector::ExactMatch("a".into()),
897                },
898            )),
899            tree_names: Some(fdiagnostics::TreeNames::All(fdiagnostics::All {})),
900            ..Default::default()
901        };
902        let int_property_selector = format!("{component_selector}:[...]root:int");
903        let mut reader = ArchiveReader::inspect();
904        reader.add_selector(int_property_selector).add_selector(lazy_property_selector);
905        let response = reader.snapshot().await?;
906        assert_eq!(response.len(), 1);
907        assert_eq!(response[0].moniker.to_string(), moniker);
908        assert_data_tree!(response[0].payload.as_ref().unwrap(), root: {
909            int: 3u64,
910            "lazy-node": {
911                a: "test"
912            }
913        });
914        Ok(())
915    }
916
917    #[fuchsia::test]
918    async fn select_all_for_moniker() {
919        let instance = start_component(ComponentOptions { publish_n_trees: 1 })
920            .await
921            .expect("component started");
922        let moniker = format!("realm_builder:{}/test_component", instance.root.child_name());
923        let results = ArchiveReader::inspect()
924            .select_all_for_component(moniker)
925            .snapshot()
926            .await
927            .expect("snapshotted");
928        assert_eq!(results.len(), 1);
929        assert_data_tree!(results[0].payload.as_ref().unwrap(), root: {
930            "tree-0": 0u64,
931            int: 3u64,
932            "lazy-node": {
933                a: "test",
934                child: {
935                    double: 3.25,
936                },
937            }
938        });
939    }
940
941    #[fuchsia::test]
942    async fn timeout() -> Result<(), anyhow::Error> {
943        let instance = start_component(ComponentOptions { publish_n_trees: 1 }).await?;
944
945        let mut reader = ArchiveReader::inspect();
946        reader
947            .add_selector(format!(
948                "realm_builder\\:{}/test_component:root",
949                instance.root.child_name()
950            ))
951            .with_timeout(zx::MonotonicDuration::from_nanos(0));
952        let result = reader.snapshot().await;
953        assert!(result.unwrap().is_empty());
954        Ok(())
955    }
956
957    #[fuchsia::test]
958    async fn component_selector() {
959        let selector = ComponentSelector::new(vec!["a".to_string()]);
960        assert_eq!(selector.moniker_str(), "a");
961        let arguments: Vec<_> = selector.to_selector_arguments().collect();
962        assert_eq!(arguments, vec![SelectorArgument::RawSelector("a:root".to_string())]);
963
964        let selector =
965            ComponentSelector::new(vec!["b".to_string(), "c".to_string(), "a".to_string()]);
966        assert_eq!(selector.moniker_str(), "b/c/a");
967
968        let selector = selector.with_tree_selector("root/b/c:d").with_tree_selector("root/e:f");
969        let arguments: Vec<_> = selector.to_selector_arguments().collect();
970        assert_eq!(
971            arguments,
972            vec![
973                SelectorArgument::RawSelector("b/c/a:root/b/c:d".into()),
974                SelectorArgument::RawSelector("b/c/a:root/e:f".into()),
975            ]
976        );
977    }
978
979    #[fuchsia::test]
980    async fn custom_archive() {
981        let proxy = spawn_fake_archive(serde_json::json!({
982            "moniker": "moniker",
983            "version": 1,
984            "data_source": "Inspect",
985            "metadata": {
986              "component_url": "component-url",
987              "timestamp": 0,
988              "filename": "filename",
989            },
990            "payload": {
991                "root": {
992                    "x": 1,
993                }
994            }
995        }));
996        let result =
997            ArchiveReader::inspect().with_archive(proxy).snapshot().await.expect("got result");
998        assert_eq!(result.len(), 1);
999        assert_data_tree!(result[0].payload.as_ref().unwrap(), root: { x: 1u64 });
1000    }
1001
1002    #[fuchsia::test]
1003    async fn handles_lists_correctly_on_snapshot_raw() {
1004        let value = serde_json::json!({
1005            "moniker": "moniker",
1006            "version": 1,
1007            "data_source": "Inspect",
1008            "metadata": {
1009            "component_url": "component-url",
1010            "timestamp": 0,
1011            "filename": "filename",
1012            },
1013            "payload": {
1014                "root": {
1015                    "x": 1,
1016                }
1017            }
1018        });
1019        let proxy = spawn_fake_archive(serde_json::json!([value.clone()]));
1020        let mut reader = ArchiveReader::inspect();
1021        reader.with_archive(proxy);
1022        let json_result = reader.snapshot_raw::<serde_json::Value>().await.expect("got result");
1023        match json_result {
1024            serde_json::Value::Array(values) => {
1025                assert_eq!(values.len(), 1);
1026                assert_eq!(values[0], value);
1027            }
1028            result => panic!("unexpected result: {result:?}"),
1029        }
1030        let cbor_result = reader.snapshot_raw::<ciborium::Value>().await.expect("got result");
1031        match cbor_result {
1032            ciborium::Value::Array(values) => {
1033                assert_eq!(values.len(), 1);
1034                let json_result =
1035                    values[0].deserialized::<serde_json::Value>().expect("convert to json");
1036                assert_eq!(json_result, value);
1037            }
1038            result => panic!("unexpected result: {result:?}"),
1039        }
1040    }
1041
1042    #[fuchsia::test(logging = false)]
1043    async fn snapshot_then_subscribe() {
1044        let (_instance, publisher, reader) = init_isolated_logging().await;
1045        let (mut stream, _errors) =
1046            reader.snapshot_then_subscribe().expect("subscribed to logs").split_streams();
1047        publisher.register_logger(None).unwrap();
1048        info!("hello from test");
1049        error!("error from test");
1050        let log = stream.next().await.unwrap();
1051        assert_eq!(log.msg().unwrap(), "hello from test");
1052        let log = stream.next().await.unwrap();
1053        assert_eq!(log.msg().unwrap(), "error from test");
1054    }
1055
1056    #[fuchsia::test]
1057    async fn read_many_trees_with_filtering() {
1058        let instance = start_component(ComponentOptions { publish_n_trees: 2 })
1059            .await
1060            .expect("component started");
1061        let selector = format!(
1062            "realm_builder\\:{}/test_component:[name=tree-0]root",
1063            instance.root.child_name()
1064        );
1065        let results = ArchiveReader::inspect()
1066            .add_selector(selector)
1067            // Only one schema since empty schemas are filtered out
1068            .with_minimum_schema_count(1)
1069            .snapshot()
1070            .await
1071            .expect("snapshotted");
1072        assert_matches!(results.iter().find(|v| v.metadata.name.as_ref() == "tree-1"), None);
1073        let should_have_data =
1074            results.into_iter().find(|v| v.metadata.name.as_ref() == "tree-0").unwrap();
1075        assert_data_tree!(should_have_data.payload.unwrap(), root: contains {
1076            "tree-0": 0u64,
1077        });
1078    }
1079
1080    fn spawn_fake_archive(data_to_send: serde_json::Value) -> fdiagnostics::ArchiveAccessorProxy {
1081        let (proxy, mut stream) =
1082            fidl::endpoints::create_proxy_and_stream::<fdiagnostics::ArchiveAccessorMarker>();
1083        fasync::Task::spawn(async move {
1084            while let Some(request) = stream.try_next().await.expect("stream request") {
1085                match request {
1086                    fdiagnostics::ArchiveAccessorRequest::StreamDiagnostics {
1087                        result_stream,
1088                        ..
1089                    } => {
1090                        let data = data_to_send.clone();
1091                        fasync::Task::spawn(handle_batch_iterator(data, result_stream)).detach();
1092                    }
1093                    fdiagnostics::ArchiveAccessorRequest::WaitForReady { responder, .. } => {
1094                        let _ = responder.send();
1095                    }
1096                    fdiagnostics::ArchiveAccessorRequest::_UnknownMethod { .. } => {
1097                        unreachable!("Unexpected method call");
1098                    }
1099                    fidl_fuchsia_diagnostics::ArchiveAccessorRequest::StreamDiagnosticsToSocket { .. } => {
1100                        unreachable!("Unexpected method call");
1101                    },
1102                }
1103            }
1104        })
1105        .detach();
1106        proxy
1107    }
1108
1109    async fn handle_batch_iterator(
1110        data: serde_json::Value,
1111        result_stream: ServerEnd<fdiagnostics::BatchIteratorMarker>,
1112    ) {
1113        let mut called = false;
1114        let mut stream = result_stream.into_stream();
1115        while let Some(req) = stream.try_next().await.expect("stream request") {
1116            match req {
1117                fdiagnostics::BatchIteratorRequest::WaitForReady { responder } => {
1118                    let _ = responder.send();
1119                }
1120                fdiagnostics::BatchIteratorRequest::GetNext { responder } => {
1121                    if called {
1122                        responder.send(Ok(Vec::new())).expect("send response");
1123                        continue;
1124                    }
1125                    called = true;
1126                    let content = serde_json::to_string_pretty(&data).expect("json pretty");
1127                    let vmo_size = content.len() as u64;
1128                    let vmo = zx::Vmo::create(vmo_size).expect("create vmo");
1129                    vmo.write(content.as_bytes(), 0).expect("write vmo");
1130                    let buffer = fidl_fuchsia_mem::Buffer { vmo, size: vmo_size };
1131                    responder
1132                        .send(Ok(vec![fdiagnostics::FormattedContent::Json(buffer)]))
1133                        .expect("send response");
1134                }
1135                fdiagnostics::BatchIteratorRequest::_UnknownMethod { .. } => {
1136                    unreachable!("Unexpected method call");
1137                }
1138            }
1139        }
1140    }
1141
1142    async fn create_realm() -> RealmBuilder {
1143        let builder = RealmBuilder::new().await.expect("create realm builder");
1144        let archivist = builder
1145            .add_child("archivist", "#meta/archivist-for-embedding.cm", ChildOptions::new().eager())
1146            .await
1147            .expect("add child archivist");
1148        builder
1149            .add_route(
1150                Route::new()
1151                    .capability(Capability::protocol_by_name("fuchsia.logger.LogSink"))
1152                    .capability(
1153                        Capability::protocol_by_name("fuchsia.tracing.provider.Registry")
1154                            .optional(),
1155                    )
1156                    .capability(Capability::event_stream("stopped"))
1157                    .capability(Capability::event_stream("capability_requested"))
1158                    .from(Ref::parent())
1159                    .to(&archivist),
1160            )
1161            .await
1162            .expect("added routes from parent to archivist");
1163        builder
1164            .add_route(
1165                Route::new()
1166                    .capability(Capability::protocol_by_name("fuchsia.logger.LogSink"))
1167                    .from(&archivist)
1168                    .to(Ref::parent()),
1169            )
1170            .await
1171            .expect("routed LogSink from archivist to parent");
1172        builder
1173            .add_route(
1174                Route::new()
1175                    .capability(Capability::protocol_by_name("fuchsia.diagnostics.ArchiveAccessor"))
1176                    .from(Ref::dictionary(&archivist, "diagnostics-accessors"))
1177                    .to(Ref::parent()),
1178            )
1179            .await
1180            .expect("routed ArchiveAccessor from archivist to parent");
1181        builder
1182    }
1183
1184    async fn init_isolated_logging() -> (RealmInstance, Publisher, ArchiveReader<Logs>) {
1185        let instance = create_realm().await.build().await.unwrap();
1186        let log_sink_client = instance.root.connect_to_protocol_at_exposed_dir().unwrap();
1187        let accessor_proxy = instance.root.connect_to_protocol_at_exposed_dir().unwrap();
1188        let mut reader = ArchiveReader::logs();
1189        reader.with_archive(accessor_proxy);
1190        let options = PublisherOptions::default().use_log_sink(log_sink_client);
1191        let publisher = Publisher::new_async(options).await.unwrap();
1192        (instance, publisher, reader)
1193    }
1194
1195    #[fuchsia::test]
1196    fn retry_config_behavior() {
1197        let config = RetryConfig::MinSchemaCount(1);
1198        let got = 0;
1199
1200        assert!(config.should_retry(got));
1201
1202        let config = RetryConfig::MinSchemaCount(1);
1203        let got = 1;
1204
1205        assert!(!config.should_retry(got));
1206
1207        let config = RetryConfig::MinSchemaCount(1);
1208        let got = 2;
1209
1210        assert!(!config.should_retry(got));
1211
1212        let config = RetryConfig::MinSchemaCount(0);
1213        let got = 1;
1214
1215        assert!(!config.should_retry(got));
1216
1217        let config = RetryConfig::always();
1218        let got = 0;
1219
1220        assert!(config.should_retry(got));
1221
1222        let config = RetryConfig::never();
1223        let got = 0;
1224
1225        assert!(!config.should_retry(got));
1226    }
1227}
diagnostics_reader/lib.rs

diagnostics_reader/
lib.rs
