archivist_lib/
formatter.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.
4use crate::diagnostics::BatchIteratorConnectionStats;
5use crate::error::AccessorError;
6use crate::logs::servers::{ExtendRecordOpts, extend_fxt_record};
7use crate::logs::shared_buffer::FilterCursor;
8use diagnostics_log_encoding::encode::{Encoder, EncoderOpts, ResizableBuffer};
9use diagnostics_log_encoding::{Argument, Header, LOG_CONTROL_BIT, Record};
10use fidl_fuchsia_diagnostics_types::Severity;
11
12use fidl_fuchsia_diagnostics::{
13    DataType, Format, FormattedContent, MAXIMUM_ENTRIES_PER_BATCH, StreamMode,
14};
15
16use fuchsia_async as fasync;
17use futures::{Stream, StreamExt};
18use log::warn;
19use pin_project::pin_project;
20use serde::Serialize;
21use std::io::Cursor;
22use std::pin::Pin;
23use std::sync::Arc;
24use std::task::{Context, Poll, ready};
25use zerocopy::{FromBytes, IntoBytes};
26use zx;
27
28static SERIALIZED_DATA_VMO_NAME: zx::Name = zx::Name::new_lossy("archivist-serialized-data");
29static PACKET_BUFFER_VMO_NAME: zx::Name = zx::Name::new_lossy("archivist-packet-buffer");
30
31const SNAPSHOT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(15);
32
33pub type FormattedStream =
34    Pin<Box<dyn Stream<Item = Vec<Result<FormattedContent, AccessorError>>> + Send>>;
35
36#[pin_project]
37pub struct FormattedContentBatcher<C> {
38    #[pin]
39    items: C,
40    stats: Arc<BatchIteratorConnectionStats>,
41}
42
43/// Make a new `FormattedContentBatcher` with a chunking strategy depending on stream mode.
44///
45/// In snapshot mode, batched items will not be flushed to the client until the batch is complete,
46/// the underlying stream has terminated, or `SNAPSHOT_TIMEOUT` has passed since the first
47/// item was discovered (in order to prevent client starvation).
48///
49/// In subscribe or snapshot-then-subscribe mode, batched items will be flushed whenever the
50/// underlying stream is pending, ensuring clients always receive latest results.
51pub fn new_batcher<I, T, E>(
52    items: I,
53    stats: Arc<BatchIteratorConnectionStats>,
54    mode: StreamMode,
55) -> FormattedStream
56where
57    I: Stream<Item = Result<T, E>> + Send + 'static,
58    T: Into<FormattedContent> + Send,
59    E: Into<AccessorError> + Send,
60{
61    match mode {
62        StreamMode::Subscribe | StreamMode::SnapshotThenSubscribe => {
63            Box::pin(FormattedContentBatcher {
64                items: items.ready_chunks(MAXIMUM_ENTRIES_PER_BATCH as _),
65                stats,
66            })
67        }
68        StreamMode::Snapshot => Box::pin(FormattedContentBatcher {
69            items: items.time_limited_chunks(MAXIMUM_ENTRIES_PER_BATCH as usize, SNAPSHOT_TIMEOUT),
70            stats,
71        }),
72    }
73}
74
75impl<I, T, E> Stream for FormattedContentBatcher<I>
76where
77    I: Stream<Item = Vec<Result<T, E>>>,
78    T: Into<FormattedContent>,
79    E: Into<AccessorError>,
80{
81    type Item = Vec<Result<FormattedContent, AccessorError>>;
82
83    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
84        let this = self.project();
85        match this.items.poll_next(cx) {
86            Poll::Ready(Some(chunk)) => {
87                // loop over chunk instead of into_iter/map because we can't move `this`
88                let mut batch = vec![];
89                for item in chunk {
90                    let result = match item {
91                        Ok(i) => Ok(i.into()),
92                        Err(e) => {
93                            this.stats.add_result_error();
94                            Err(e.into())
95                        }
96                    };
97                    batch.push(result);
98                }
99                Poll::Ready(Some(batch))
100            }
101            Poll::Ready(None) => Poll::Ready(None),
102            Poll::Pending => Poll::Pending,
103        }
104    }
105}
106
107#[derive(Debug, Clone, Copy, PartialEq, Eq)]
108enum TimerState {
109    Idle,
110    Running,
111}
112
113pub trait TimeLimitedChunksExt: Stream {
114    fn time_limited_chunks(
115        self,
116        capacity: usize,
117        timeout: std::time::Duration,
118    ) -> TimeLimitedChunks<Self>
119    where
120        Self: Sized,
121    {
122        TimeLimitedChunks::new(self, capacity, timeout)
123    }
124}
125
126impl<T: Stream> TimeLimitedChunksExt for T {}
127
128/// A stream adapter that yields chunks of items from the underlying stream.
129///
130/// Chunks are yielded when they reach the specified capacity or when the specified timeout
131/// expires since the first item in the chunk was received.
132#[pin_project]
133pub struct TimeLimitedChunks<S: Stream> {
134    #[pin]
135    stream: S,
136    capacity: usize,
137    timeout: std::time::Duration,
138    buffer: Vec<S::Item>,
139    #[pin]
140    timer: fasync::Timer,
141    state: TimerState,
142}
143
144impl<S: Stream> TimeLimitedChunks<S> {
145    pub fn new(stream: S, capacity: usize, timeout: std::time::Duration) -> Self {
146        Self {
147            stream,
148            capacity,
149            timeout,
150            buffer: Vec::with_capacity(capacity),
151            timer: fasync::Timer::new(std::time::Instant::now() + timeout),
152            state: TimerState::Idle,
153        }
154    }
155}
156
157impl<S: Stream> Stream for TimeLimitedChunks<S> {
158    type Item = Vec<S::Item>;
159
160    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
161        let mut this = self.project();
162
163        loop {
164            match this.stream.as_mut().poll_next(cx) {
165                Poll::Ready(Some(item)) => {
166                    if this.buffer.is_empty() {
167                        let deadline = fasync::MonotonicInstant::after(
168                            zx::MonotonicDuration::from_nanos(this.timeout.as_nanos() as i64),
169                        );
170                        this.timer.as_mut().reset(deadline);
171                        *this.state = TimerState::Running;
172                    }
173                    this.buffer.push(item);
174                    if this.buffer.len() >= *this.capacity {
175                        *this.state = TimerState::Idle;
176                        return Poll::Ready(Some(std::mem::take(this.buffer)));
177                    }
178                }
179                Poll::Ready(None) => {
180                    if !this.buffer.is_empty() {
181                        *this.state = TimerState::Idle;
182                        return Poll::Ready(Some(std::mem::take(this.buffer)));
183                    }
184                    return Poll::Ready(None);
185                }
186                Poll::Pending => {
187                    if !this.buffer.is_empty() && *this.state == TimerState::Running {
188                        use std::future::Future;
189                        if this.timer.as_mut().poll(cx).is_ready() {
190                            *this.state = TimerState::Idle;
191                            return Poll::Ready(Some(std::mem::take(this.buffer)));
192                        }
193                    }
194                    return Poll::Pending;
195                }
196            }
197        }
198    }
199}
200
201/// Holds a VMO containing valid serialized data as well as the size of that data.
202pub struct SerializedVmo {
203    pub vmo: zx::Vmo,
204    pub size: u64,
205    format: Format,
206}
207
208impl SerializedVmo {
209    pub fn serialize(
210        source: &impl Serialize,
211        data_type: DataType,
212        format: Format,
213    ) -> Result<Self, AccessorError> {
214        let initial_buffer_capacity = match data_type {
215            DataType::Inspect => inspect_format::constants::DEFAULT_VMO_SIZE_BYTES,
216            // Logs won't go through this codepath anyway, but in case we ever want to serialize a
217            // single log instance it makes sense to start at the page size.
218            DataType::Logs => 4096, // page size
219        };
220        let mut buffer = Vec::with_capacity(initial_buffer_capacity);
221        match format {
222            Format::Json => {
223                serde_json::to_writer(&mut buffer, source).map_err(AccessorError::Serialization)?
224            }
225            Format::Cbor => ciborium::into_writer(source, &mut buffer)
226                .map_err(|err| AccessorError::CborSerialization(err.into()))?,
227            Format::Text => unreachable!("We'll never get Text"),
228            Format::Fxt => unreachable!("We'll never get FXT"),
229        }
230        let vmo = zx::Vmo::create(buffer.len() as u64).unwrap();
231        let _ = vmo.set_name(&SERIALIZED_DATA_VMO_NAME);
232        vmo.write(&buffer, 0).unwrap();
233        Ok(Self { vmo, size: buffer.len() as u64, format })
234    }
235}
236
237pub fn encode_rolled_out(count: u64) -> Vec<u8> {
238    let mut encoder =
239        Encoder::new(Cursor::new(ResizableBuffer::from(Vec::new())), EncoderOpts::default());
240    let record = Record {
241        timestamp: zx::BootInstant::from_nanos(0),
242        severity: Severity::Info.into_primitive(),
243        arguments: vec![Argument::new(
244            "rolled_out",
245            diagnostics_log_encoding::Value::UnsignedInt(count),
246        )],
247    };
248    // This write should not fail as ResizableBuffer will grow as needed.
249    encoder.write_record(record).unwrap();
250
251    let mut buffer = encoder.take().into_inner().into_inner();
252    if buffer.len() >= 8 {
253        let mut header = Header::read_from_bytes(&buffer[0..8]).unwrap();
254        header.set_tag(LOG_CONTROL_BIT);
255        buffer[0..8].copy_from_slice(header.as_bytes());
256    }
257    buffer
258}
259
260impl From<SerializedVmo> for FormattedContent {
261    fn from(content: SerializedVmo) -> FormattedContent {
262        match content.format {
263            Format::Json => {
264                // set_content_size() is redundant, but consumers may expect the size there.
265                content
266                    .vmo
267                    .set_content_size(&content.size)
268                    .expect("set_content_size always returns Ok");
269                FormattedContent::Json(fidl_fuchsia_mem::Buffer {
270                    vmo: content.vmo,
271                    size: content.size,
272                })
273            }
274            Format::Cbor => {
275                content
276                    .vmo
277                    .set_content_size(&content.size)
278                    .expect("set_content_size always returns Ok");
279                FormattedContent::Cbor(content.vmo)
280            }
281            Format::Fxt => {
282                content
283                    .vmo
284                    .set_content_size(&content.size)
285                    .expect("set_content_size always returns Ok");
286                FormattedContent::Fxt(content.vmo)
287            }
288            Format::Text => unreachable!("We'll never get Text"),
289        }
290    }
291}
292
293trait PacketFormat {
294    const FORMAT: Format;
295    const HEADER: &[u8] = &[];
296    const FOOTER: &[u8] = &[];
297
298    /// Writes an item in the required format.  Returns `Poll::Ready(Some(<separator length>))` if
299    /// an item was written, and `Poll::Ready(None)` if there are no more items.  If `first` is true,
300    /// this is the first item in this batch.
301    fn write_item(
302        self: Pin<&mut Self>,
303        cx: &mut Context<'_>,
304        first: bool,
305        buffer: &mut Vec<u8>,
306    ) -> Poll<Option<usize>>;
307}
308
309#[pin_project]
310pub struct PacketSerializer<T> {
311    stats: Option<Arc<BatchIteratorConnectionStats>>,
312    max_packet_size: u64,
313    #[pin]
314    format: T,
315    overflow: Vec<u8>,
316    finished: bool,
317}
318
319impl<T> PacketSerializer<T> {
320    fn with_format(
321        stats: Option<Arc<BatchIteratorConnectionStats>>,
322        max_packet_size: u64,
323        format: T,
324    ) -> Self {
325        Self { stats, max_packet_size, format, overflow: Vec::new(), finished: false }
326    }
327}
328
329impl<T: PacketFormat> Stream for PacketSerializer<T> {
330    type Item = Result<SerializedVmo, AccessorError>;
331
332    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
333        if self.finished {
334            return Poll::Ready(None);
335        }
336
337        // Limit packet size to prevent unbounded memory use.
338        const MAX_PACKET_SIZE_LIMIT: u64 = 1 << 20; // 1 MiB
339        let max_packet_size = std::cmp::min(self.max_packet_size, MAX_PACKET_SIZE_LIMIT);
340        let mut this = self.project();
341
342        let mut buffer = Vec::with_capacity(256 * 1024);
343        buffer.extend_from_slice(T::HEADER);
344
345        let mut results = 0;
346
347        if !this.overflow.is_empty() {
348            buffer.append(this.overflow);
349            results += 1;
350        }
351
352        let mut vmo = None;
353        let mut vmo_len = 0;
354
355        loop {
356            // Copy to the VMO if the room in the buffer drops below a threshold.
357            if buffer.capacity() - buffer.len() < 512 {
358                vmo.get_or_insert_with(|| {
359                    let v = zx::Vmo::create(max_packet_size).unwrap();
360                    let _ = v.set_name(&PACKET_BUFFER_VMO_NAME);
361                    v
362                })
363                .write(&buffer, vmo_len as u64)
364                .unwrap();
365                vmo_len += buffer.len();
366                buffer.clear();
367            }
368
369            let last_len = buffer.len();
370
371            let separator_len = match this.format.as_mut().write_item(cx, results == 0, &mut buffer)
372            {
373                Poll::Ready(Some(separator_len)) => separator_len,
374                Poll::Ready(None) => {
375                    *this.finished = true;
376                    if results == 0 {
377                        return Poll::Ready(None);
378                    } else {
379                        break;
380                    }
381                }
382                Poll::Pending => {
383                    if results == 0 {
384                        return Poll::Pending;
385                    } else {
386                        break;
387                    }
388                }
389            };
390
391            let item_len = buffer.len() - last_len - separator_len;
392
393            if (item_len + T::HEADER.len() + T::FOOTER.len()) as u64 >= max_packet_size {
394                warn!("dropping oversize item (limit={max_packet_size} len={item_len})");
395                buffer.truncate(last_len);
396            } else {
397                if (vmo_len + buffer.len() + T::FOOTER.len()) as u64 > max_packet_size {
398                    // Last item put us over the maximum packet size, keep it for the next batch.
399                    // We should have at least one item because otherwise we should have gone
400                    // through the branch above.
401                    assert!(results > 0);
402                    this.overflow.extend_from_slice(&buffer[last_len + separator_len..]);
403                    buffer.truncate(last_len);
404                    break;
405                }
406
407                results += 1;
408            }
409        }
410
411        if let Some(stats) = &this.stats {
412            stats.add_result(results);
413        }
414
415        buffer.extend_from_slice(T::FOOTER);
416
417        let vmo = match vmo {
418            Some(vmo) => {
419                vmo.set_stream_size((vmo_len + buffer.len()) as u64).unwrap();
420                vmo
421            }
422            None => {
423                let v = zx::Vmo::create(buffer.len() as u64).unwrap();
424                let _ = v.set_name(&PACKET_BUFFER_VMO_NAME);
425                v
426            }
427        };
428        vmo.write(&buffer, vmo_len as u64).unwrap();
429        vmo_len += buffer.len();
430        Poll::Ready(Some(Ok(SerializedVmo { vmo, size: vmo_len as u64, format: T::FORMAT })))
431    }
432}
433
434#[pin_project]
435pub struct FxtPacketFormat {
436    #[pin]
437    pub cursor: FilterCursor,
438    pub subscribe_to_manifest: bool,
439    pub sent_tags: std::collections::HashMap<u64, Arc<crate::identity::ComponentIdentity>>,
440}
441
442impl PacketFormat for FxtPacketFormat {
443    const FORMAT: Format = Format::Fxt;
444
445    fn write_item(
446        self: Pin<&mut Self>,
447        cx: &mut Context<'_>,
448        _first: bool,
449        buffer: &mut Vec<u8>,
450    ) -> Poll<Option<usize>> {
451        let mut this = self.project();
452        loop {
453            let Some(message) = ready!(this.cursor.as_mut().poll_next(cx)) else {
454                return Poll::Ready(None);
455            };
456
457            let (identity, header, data) = match message.parse() {
458                Ok(res) => res,
459                Err(_) => {
460                    // If we fail to parse the message, just ignore it and move on to the next
461                    // message.
462                    continue;
463                }
464            };
465            let tag = header.tag() as u64;
466
467            if *this.subscribe_to_manifest {
468                let send_manifest = match this.sent_tags.entry(tag) {
469                    std::collections::hash_map::Entry::Vacant(e) => {
470                        e.insert(Arc::clone(identity));
471                        true
472                    }
473                    std::collections::hash_map::Entry::Occupied(mut e) => {
474                        if !Arc::ptr_eq(e.get(), identity) && **e.get() != **identity {
475                            e.insert(Arc::clone(identity));
476                            true
477                        } else {
478                            false
479                        }
480                    }
481                };
482
483                if message.dropped > 0 {
484                    let rolled_out_buffer = encode_rolled_out(message.dropped);
485                    buffer.extend_from_slice(&rolled_out_buffer);
486                }
487
488                if send_manifest {
489                    use diagnostics_log_encoding::encode::{Encoder, EncoderOpts, ResizableBuffer};
490                    use diagnostics_log_encoding::{Argument, LOG_CONTROL_BIT, Record};
491                    use fidl_fuchsia_diagnostics_types::Severity;
492                    use std::io::Cursor;
493
494                    let mut encoder = Encoder::new(
495                        Cursor::new(ResizableBuffer::from(Vec::new())),
496                        EncoderOpts::default(),
497                    );
498                    let record = Record {
499                        timestamp: zx::BootInstant::from_nanos(0),
500                        severity: Severity::Info.into_primitive(),
501                        arguments: vec![
502                            Argument::other("moniker", identity.moniker.to_string()),
503                            Argument::other("url", identity.url.as_str()),
504                        ],
505                    };
506                    encoder.write_record(record).unwrap();
507                    let mut manifest_buffer = encoder.take().into_inner().into_inner();
508                    if manifest_buffer.len() >= 8 {
509                        let mut header_manifest =
510                            Header::read_from_bytes(&manifest_buffer[0..8]).unwrap();
511                        header_manifest.set_tag((tag as u32) | LOG_CONTROL_BIT);
512                        manifest_buffer[0..8].copy_from_slice(header_manifest.as_bytes());
513                    }
514                    buffer.extend_from_slice(&manifest_buffer);
515                }
516            }
517
518            buffer.extend_from_slice(header.as_bytes());
519            buffer.extend_from_slice(&data[8..]);
520
521            extend_fxt_record(
522                identity,
523                message.dropped,
524                &ExtendRecordOpts {
525                    component_url: !*this.subscribe_to_manifest,
526                    moniker: !*this.subscribe_to_manifest,
527                    rolled_out: !*this.subscribe_to_manifest,
528                },
529                buffer,
530            );
531            return Poll::Ready(Some(0));
532        }
533    }
534}
535
536pub type FxtPacketSerializer = PacketSerializer<FxtPacketFormat>;
537
538impl FxtPacketSerializer {
539    pub fn new(
540        stats: Arc<BatchIteratorConnectionStats>,
541        max_packet_size: u64,
542        cursor: FilterCursor,
543        subscribe_to_manifest: bool,
544    ) -> Self {
545        Self::with_format(
546            Some(stats),
547            max_packet_size,
548            FxtPacketFormat {
549                cursor,
550                subscribe_to_manifest,
551                sent_tags: std::collections::HashMap::new(),
552            },
553        )
554    }
555}
556
557#[pin_project]
558pub struct JsonPacketFormat<I>(#[pin] I);
559
560impl<I: Stream<Item = impl Serialize>> PacketFormat for JsonPacketFormat<I> {
561    const FORMAT: Format = Format::Json;
562    const HEADER: &[u8] = b"[";
563    const FOOTER: &[u8] = b"]";
564
565    fn write_item(
566        self: Pin<&mut Self>,
567        cx: &mut Context<'_>,
568        first: bool,
569        buffer: &mut Vec<u8>,
570    ) -> Poll<Option<usize>> {
571        const SEPARATOR: &[u8] = b",\n";
572
573        if let Some(item) = ready!(self.project().0.poll_next(cx)) {
574            let separator_len = if !first {
575                buffer.extend_from_slice(SEPARATOR);
576                SEPARATOR.len()
577            } else {
578                0
579            };
580            // We don't expect serialization to fail because we should always be able to write to
581            // `buffer` and `item` is a type we control which we know should always be serializable.
582            serde_json::to_writer(buffer, &item).expect("failed to serialize item");
583            Poll::Ready(Some(separator_len))
584        } else {
585            Poll::Ready(None)
586        }
587    }
588}
589
590pub type JsonPacketSerializer<I> = PacketSerializer<JsonPacketFormat<I>>;
591
592impl<I> JsonPacketSerializer<I> {
593    pub fn new(stats: Arc<BatchIteratorConnectionStats>, max_packet_size: u64, items: I) -> Self {
594        Self::with_format(Some(stats), max_packet_size, JsonPacketFormat(items))
595    }
596
597    pub fn new_without_stats(max_packet_size: u64, items: I) -> Self {
598        Self::with_format(None, max_packet_size, JsonPacketFormat(items))
599    }
600}
601
602#[cfg(test)]
603mod tests {
604    use super::*;
605    use crate::diagnostics::AccessorStats;
606    use futures::stream::iter;
607
608    #[fuchsia::test]
609    async fn time_limited_chunks_yields_on_capacity() {
610        let stream = futures::stream::iter(vec![1, 2, 3, 4]);
611        let mut chunks =
612            Box::pin(TimeLimitedChunks::new(stream, 2, std::time::Duration::from_secs(30)));
613
614        assert_eq!(chunks.next().await, Some(vec![1, 2]));
615        assert_eq!(chunks.next().await, Some(vec![3, 4]));
616        assert_eq!(chunks.next().await, None);
617    }
618
619    #[fuchsia::test]
620    async fn time_limited_chunks_yields_on_stream_end() {
621        let stream = futures::stream::iter(vec![1, 2, 3]);
622        let mut chunks =
623            Box::pin(TimeLimitedChunks::new(stream, 2, std::time::Duration::from_secs(30)));
624
625        assert_eq!(chunks.next().await, Some(vec![1, 2]));
626        assert_eq!(chunks.next().await, Some(vec![3]));
627        assert_eq!(chunks.next().await, None);
628    }
629
630    #[fuchsia::test]
631    async fn time_limited_chunks_yields_on_timeout() {
632        let (tx, rx) = futures::channel::mpsc::unbounded();
633        let mut chunks =
634            Box::pin(TimeLimitedChunks::new(rx, 2, std::time::Duration::from_millis(10)));
635
636        tx.unbounded_send(1).unwrap();
637
638        let start = std::time::Instant::now();
639        assert_eq!(chunks.next().await, Some(vec![1]));
640        assert!(start.elapsed() >= std::time::Duration::from_millis(10));
641
642        tx.unbounded_send(2).unwrap();
643        tx.unbounded_send(3).unwrap();
644        assert_eq!(chunks.next().await, Some(vec![2, 3]));
645
646        drop(tx);
647        assert_eq!(chunks.next().await, None);
648    }
649
650    #[fuchsia::test]
651    async fn two_items_joined_and_split() {
652        let inputs = &[&"FFFFFFFFFF", &"GGGGGGGGGG"];
653        let joined = &["[\"FFFFFFFFFF\",\n\"GGGGGGGGGG\"]"];
654        let split = &[r#"["FFFFFFFFFF"]"#, r#"["GGGGGGGGGG"]"#];
655        let smallest_possible_joined_len = joined[0].len() as u64;
656
657        let make_packets = |max| async move {
658            let node = fuchsia_inspect::Node::default();
659            let accessor_stats = Arc::new(AccessorStats::new(node));
660            let test_stats = Arc::new(accessor_stats.new_logs_batch_iterator());
661            JsonPacketSerializer::new(test_stats, max, iter(inputs.iter()))
662                .collect::<Vec<_>>()
663                .await
664                .into_iter()
665                .map(|r| {
666                    let result = r.unwrap();
667                    let mut buf = vec![0; result.size as usize];
668                    result.vmo.read(&mut buf, 0).expect("reading vmo");
669                    std::str::from_utf8(&buf).unwrap().to_string()
670                })
671                .collect::<Vec<_>>()
672        };
673
674        let actual_joined = make_packets(smallest_possible_joined_len).await;
675        assert_eq!(&actual_joined[..], joined);
676
677        let actual_split = make_packets(smallest_possible_joined_len - 1).await;
678        assert_eq!(&actual_split[..], split);
679    }
680
681    #[fuchsia::test]
682    async fn overflow_separator_added() {
683        let inputs = &[&"A", &"B", &"C"];
684        // "[" + "A" + "]" = 4 bytes.
685        // "[" + "A" + ",\n" + "B" + "]" = 4 + 2 + 3 = 9 bytes.
686        // If max is 8, "B" will overflow.
687        // Second packet starts with "B" from overflow.
688        // "[" + "B" + ",\n" + "C" + "]" = 4 + 2 + 3 = 9 bytes.
689        // If max is 8, "C" will overflow.
690        // Third packet starts with "C" from overflow.
691
692        let make_packets = |max| async move {
693            let node = fuchsia_inspect::Node::default();
694            let accessor_stats = Arc::new(AccessorStats::new(node));
695            let test_stats = Arc::new(accessor_stats.new_logs_batch_iterator());
696            JsonPacketSerializer::new(test_stats, max, iter(inputs.iter()))
697                .collect::<Vec<_>>()
698                .await
699                .into_iter()
700                .map(|r| {
701                    let result = r.unwrap();
702                    let mut buf = vec![0; result.size as usize];
703                    result.vmo.read(&mut buf, 0).expect("reading vmo");
704                    std::str::from_utf8(&buf).unwrap().to_string()
705                })
706                .collect::<Vec<_>>()
707        };
708
709        let packets = make_packets(8).await;
710        assert_eq!(packets.len(), 3);
711        assert_eq!(packets[0], r#"["A"]"#);
712        assert_eq!(packets[1], r#"["B"]"#);
713        assert_eq!(packets[2], r#"["C"]"#);
714
715        let packets = make_packets(10).await;
716        assert_eq!(packets.len(), 2);
717        assert_eq!(packets[0], "[\"A\",\n\"B\"]");
718        assert_eq!(packets[1], r#"["C"]"#);
719    }
720
721    #[fuchsia::test]
722    async fn oversize_item_not_dropped_incorrectly() {
723        let inputs = &[&"A", &"BCDEF"];
724        // Packet 1: ["A"] (4 bytes)
725        // Item 2: "BCDEF" (7 bytes)
726        // "[" + "A" + ",\n" + "BCDEF" + "]" = 1 + 3 + 2 + 7 + 1 = 14 bytes.
727        // If max is 11:
728        // "A" fits (4 bytes).
729        // "BCDEF" overflows.
730        // NEXT packet:
731        // "[" + "BCDEF" + "]" = 9 bytes.
732        // 9 fits in 11.
733
734        let make_packets = |max| async move {
735            let node = fuchsia_inspect::Node::default();
736            let accessor_stats = Arc::new(AccessorStats::new(node));
737            let test_stats = Arc::new(accessor_stats.new_logs_batch_iterator());
738            JsonPacketSerializer::new(test_stats, max, iter(inputs.iter()))
739                .collect::<Vec<_>>()
740                .await
741                .into_iter()
742                .map(|r| {
743                    let result = r.unwrap();
744                    let mut buf = vec![0; result.size as usize];
745                    result.vmo.read(&mut buf, 0).expect("reading vmo");
746                    std::str::from_utf8(&buf).unwrap().to_string()
747                })
748                .collect::<Vec<_>>()
749        };
750
751        let packets = make_packets(11).await;
752        assert_eq!(packets.len(), 2);
753        assert_eq!(packets[0], r#"["A"]"#);
754        assert_eq!(packets[1], r#"["BCDEF"]"#);
755    }
756
757    #[fuchsia::test]
758    async fn item_too_big_for_packet_is_dropped() {
759        let inputs = &[&"ABCDE"]; // 7 bytes
760        // "[" + 7 + "]" = 9 bytes.
761        // If max is 8, it should be dropped.
762
763        let make_packets = |max| async move {
764            let node = fuchsia_inspect::Node::default();
765            let accessor_stats = Arc::new(AccessorStats::new(node));
766            let test_stats = Arc::new(accessor_stats.new_logs_batch_iterator());
767            JsonPacketSerializer::new(test_stats, max, iter(inputs.iter()))
768                .collect::<Vec<_>>()
769                .await
770                .into_iter()
771                .map(|r| {
772                    let result = r.unwrap();
773                    let mut buf = vec![0; result.size as usize];
774                    result.vmo.read(&mut buf, 0).expect("reading vmo");
775                    std::str::from_utf8(&buf).unwrap().to_string()
776                })
777                .collect::<Vec<_>>()
778        };
779
780        let packets = make_packets(8).await;
781        // Item should be dropped, so we get no packets.
782        assert_eq!(packets.len(), 0);
783    }
784
785    #[fuchsia::test]
786    async fn fxt_packet_serializer_subscribe_to_manifest() {
787        use crate::identity::ComponentIdentity;
788        use crate::logs::shared_buffer::{SharedBuffer, create_ring_buffer};
789        use crate::logs::stats::LogStreamStats;
790        use diagnostics_log_encoding::encode::{Encoder, EncoderOpts};
791        use diagnostics_log_encoding::{Argument, Header, LOG_CONTROL_BIT, Record};
792        use fidl_fuchsia_diagnostics::StreamMode;
793        use fidl_fuchsia_diagnostics_types::Severity;
794        use fuchsia_inspect::Node;
795        use std::io::Cursor;
796        use zerocopy::{FromBytes, IntoBytes};
797
798        let identity1 = Arc::new(ComponentIdentity::unknown());
799        let mut identity2_inner = ComponentIdentity::unknown();
800        identity2_inner.url =
801            flyweights::FlyStr::new("fuchsia-pkg://fuchsia.com/test#meta/test.cm");
802        let identity2 = Arc::new(identity2_inner);
803
804        let buffer = Arc::new(SharedBuffer::new(
805            create_ring_buffer(65536),
806            Box::new(|_| {}),
807            Default::default(),
808            &Node::default(),
809        ));
810        let stats1 = Arc::new(LogStreamStats::new(&Node::default(), &identity1));
811        let container1 = buffer.new_container_buffer(Arc::clone(&identity1), stats1);
812        let stats2 = Arc::new(LogStreamStats::new(&Node::default(), &identity2));
813        let container2 = buffer.new_container_buffer(Arc::clone(&identity2), stats2);
814
815        // create message 1
816        let mut buffer_out = Cursor::new(vec![0u8; 128]);
817        let mut encoder = Encoder::new(&mut buffer_out, EncoderOpts::default());
818        encoder
819            .write_record(Record {
820                timestamp: zx::BootInstant::from_nanos(100),
821                severity: Severity::Info.into_primitive(),
822                arguments: vec![Argument::other("msg", "hello")],
823            })
824            .unwrap();
825        let end = encoder.inner().position() as usize;
826        let mut msg1_bytes = encoder.inner().get_ref()[..end].to_vec();
827        let mut header = Header::read_from_bytes(&msg1_bytes[0..8]).unwrap();
828        header.set_tag(1);
829        msg1_bytes[0..8].copy_from_slice(header.as_bytes());
830        container1.push_back(&msg1_bytes);
831
832        // create message 2
833        let mut buffer_out = Cursor::new(vec![0u8; 128]);
834        let mut encoder = Encoder::new(&mut buffer_out, EncoderOpts::default());
835        encoder
836            .write_record(Record {
837                timestamp: zx::BootInstant::from_nanos(200),
838                severity: Severity::Info.into_primitive(),
839                arguments: vec![Argument::other("msg", "world")],
840            })
841            .unwrap();
842        let end = encoder.inner().position() as usize;
843        let mut msg2_bytes = encoder.inner().get_ref()[..end].to_vec();
844        let mut header = Header::read_from_bytes(&msg2_bytes[0..8]).unwrap();
845        header.set_tag(2);
846        msg2_bytes[0..8].copy_from_slice(header.as_bytes());
847        container2.push_back(&msg2_bytes);
848
849        // create message 3 (same tag/identity as message 2)
850        let mut buffer_out = Cursor::new(vec![0u8; 128]);
851        let mut encoder = Encoder::new(&mut buffer_out, EncoderOpts::default());
852        encoder
853            .write_record(Record {
854                timestamp: zx::BootInstant::from_nanos(300),
855                severity: Severity::Info.into_primitive(),
856                arguments: vec![Argument::other("msg", "again")],
857            })
858            .unwrap();
859        let end = encoder.inner().position() as usize;
860        let mut msg3_bytes = encoder.inner().get_ref()[..end].to_vec();
861        let mut header = Header::read_from_bytes(&msg3_bytes[0..8]).unwrap();
862        header.set_tag(2);
863        msg3_bytes[0..8].copy_from_slice(header.as_bytes());
864        container2.push_back(&msg3_bytes);
865
866        let cursor = buffer.cursor(StreamMode::Snapshot, vec![]);
867
868        let node = Node::default();
869        let accessor_stats = Arc::new(AccessorStats::new(node));
870        let test_stats = Arc::new(accessor_stats.new_logs_batch_iterator());
871
872        // Test with subscribe_to_manifest = true
873        let packets: Vec<_> =
874            FxtPacketSerializer::new(Arc::clone(&test_stats), 1024 * 1024, cursor, true)
875                .collect::<Vec<_>>()
876                .await
877                .into_iter()
878                .map(|r| {
879                    let result = r.unwrap();
880                    let mut buf = vec![0; result.size as usize];
881                    result.vmo.read(&mut buf, 0).expect("reading vmo");
882                    buf
883                })
884                .collect();
885
886        assert_eq!(packets.len(), 1);
887        let output = &packets[0];
888
889        let mut current_slice = output.as_slice();
890        let mut records = vec![];
891
892        while !current_slice.is_empty() {
893            let (record, remaining) =
894                diagnostics_log_encoding::parse::parse_record(current_slice).unwrap();
895            let header =
896                diagnostics_log_encoding::Header::read_from_bytes(&current_slice[..8]).unwrap();
897            records.push((header, record));
898            current_slice = remaining;
899        }
900
901        assert_eq!(records.len(), 5);
902
903        // Record 0: Manifest for tag 0
904        assert_ne!(records[0].0.tag() & LOG_CONTROL_BIT, 0);
905        assert_eq!(records[0].0.tag() & !LOG_CONTROL_BIT, 0);
906
907        // Record 1: Msg 1
908        assert_eq!(records[1].0.tag(), 0);
909        assert_eq!(records[1].1.arguments.len(), 1); // no moniker/url injected
910
911        // Record 2: Manifest for tag 1
912        assert_ne!(records[2].0.tag() & LOG_CONTROL_BIT, 0);
913        assert_eq!(records[2].0.tag() & !LOG_CONTROL_BIT, 1);
914
915        // Record 3: Msg 2
916        assert_eq!(records[3].0.tag(), 1);
917        assert_eq!(records[3].1.arguments.len(), 1); // no moniker/url injected
918
919        // Record 4: Msg 3
920        assert_eq!(records[4].0.tag(), 1);
921        assert_eq!(records[4].1.arguments.len(), 1); // no moniker/url injected
922    }
923}
archivist_lib/formatter.rs

archivist_lib/
formatter.rs
