archivist_lib/logs/shared_buffer/
cursor.rs

1// Copyright 2025 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 super::{ContainerId, Inner, InnerGuard, SharedBuffer};
6use crate::identity::ComponentIdentity;
7use diagnostics_data::{LogError, LogsData};
8use diagnostics_log_encoding::parse::ParseError;
9use diagnostics_log_encoding::{Header, TRACING_FORMAT_LOG_RECORD_TYPE};
10use diagnostics_message::error::MessageError;
11use fidl_fuchsia_diagnostics::ComponentSelector;
12use fuchsia_async::condition::{ConditionGuard, WakerEntry};
13use futures::Stream;
14use futures::stream::FusedStream;
15use pin_project::pin_project;
16use ring_buffer::ring_buffer_record_len;
17use selectors::matches_selectors;
18use std::cmp::Ordering;
19use std::collections::BinaryHeap;
20use std::marker::PhantomData;
21use std::pin::Pin;
22use std::sync::Arc;
23use std::task::{Context, Poll, ready};
24use zerocopy::FromBytes;
25
26/// FilterCursor is a cursor that returns all logs optionally filtered by component selectors.
27#[pin_project]
28pub struct FilterCursor {
29    buffer: Arc<SharedBuffer>,
30    index: u64,
31    message_id: u64,
32    end: Option<u64>,
33    selectors: Vec<ComponentSelector>,
34    messages: BinaryHeap<MessageRef>,
35    flush_sockets_for_snapshot: bool,
36    #[pin]
37    waker_entry: WakerEntry<Inner>,
38}
39
40impl FilterCursor {
41    pub fn new(
42        buffer: Arc<SharedBuffer>,
43        index: u64,
44        message_id: u64,
45        snapshot: bool,
46        selectors: Vec<ComponentSelector>,
47    ) -> Self {
48        let waker_entry = buffer.inner.waker_entry();
49        Self {
50            buffer,
51            index,
52            message_id,
53            end: None,
54            selectors,
55            messages: BinaryHeap::new(),
56            waker_entry,
57            flush_sockets_for_snapshot: snapshot,
58        }
59    }
60
61    /// Polls for the next message.  We can't use the Stream trait because this is a lending
62    /// iterator; it returns a reference to a message (and will hold a lock).  Depending on the use,
63    /// this can be more efficient than using a Stream which will involve copying the message.
64    pub fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Message<'_>>> {
65        let this = self.project();
66
67        if *this.flush_sockets_for_snapshot {
68            let mut inner = InnerGuard::new(this.buffer);
69            *this.end = Some(this.buffer.flush_sockets(&mut inner));
70            *this.flush_sockets_for_snapshot = false;
71        }
72
73        let mut inner = this.buffer.inner.lock();
74
75        // Comparing monikers can be expensive, so memoize the results.
76        struct ContainerIds {
77            ids: [ContainerId; 8],
78            result: u8,
79            pos: usize,
80        }
81
82        impl ContainerIds {
83            /// Memoizes the result of `predicate` for a given `id`.
84            fn memoize(&mut self, id: ContainerId, predicate: impl FnOnce() -> bool) -> bool {
85                if let Some(pos) = self.ids.iter().position(|i| i == &id) {
86                    self.result & (1 << pos) != 0
87                } else {
88                    let result = predicate();
89                    self.ids[self.pos] = id;
90                    if result {
91                        self.result |= 1 << self.pos;
92                    } else {
93                        self.result &= !(1 << self.pos);
94                    }
95                    self.pos = (self.pos + 1) % self.ids.len();
96                    result
97                }
98            }
99        }
100
101        let mut container_ids = ContainerIds { ids: [ContainerId(0xffff); 8], result: 0, pos: 0 };
102
103        // NOTE: If messages are dropped the dropped count won't account for filtering: it will
104        // include messages that have been dropped that don't match the filter.  Fixing this is
105        // difficult and not worth the effort.  Dropped messages should be rare and the common case
106        // is that there is no filtering.
107        let mut dropped = 0;
108        if *this.index < inner.tail {
109            *this.index = inner.tail;
110            dropped += inner.tail_message_id - *this.message_id;
111            *this.message_id = inner.tail_message_id;
112        }
113        while *this.index < inner.last_scanned && this.end.is_none_or(|e| *this.index < e) {
114            // SAFETY: We've checked index >= inner.tail, so the range must be valid.  `msg`
115            // will remain valid whilst we're holding the lock.
116            let (component, msg, timestamp) =
117                unsafe { inner.parse_message(*this.index..inner.last_scanned) };
118
119            if let Some(timestamp) = timestamp
120                && let Some(container) = inner.containers.get(component)
121                && (this.selectors.is_empty()
122                    || container_ids.memoize(component, || {
123                        matches_selectors(&container.identity.moniker, this.selectors)
124                    }))
125            {
126                this.messages.push(MessageRef { index: *this.index, timestamp });
127            }
128
129            *this.index += ring_buffer_record_len(msg.len()) as u64;
130            *this.message_id += 1;
131        }
132        while let Some(message) = this.messages.pop() {
133            if message.index < inner.tail {
134                // See note above regarding filtering.
135                dropped += 1;
136                continue;
137            }
138            return Poll::Ready(Some(Message { inner, message, dropped }));
139        }
140        if this.end.is_some_and(|e| *this.index >= e) || inner.terminated {
141            *this.index = u64::MAX;
142            Poll::Ready(None)
143        } else {
144            inner.add_waker(this.waker_entry, cx.waker().clone());
145            Poll::Pending
146        }
147    }
148
149    fn is_terminated(&self) -> bool {
150        self.index == u64::MAX
151    }
152}
153
154#[derive(Eq)]
155struct MessageRef {
156    timestamp: zx::BootInstant,
157    index: u64,
158}
159
160impl PartialEq for MessageRef {
161    fn eq(&self, other: &Self) -> bool {
162        self.timestamp == other.timestamp && self.index == other.index
163    }
164}
165
166impl PartialOrd for MessageRef {
167    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
168        Some(self.cmp(other))
169    }
170}
171
172impl Ord for MessageRef {
173    fn cmp(&self, other: &Self) -> Ordering {
174        // BinaryHeap is a max-heap, but we want min, hence this ordering.
175        other.timestamp.cmp(&self.timestamp).then_with(|| other.index.cmp(&self.index))
176    }
177}
178
179pub struct Message<'a> {
180    inner: ConditionGuard<'a, Inner>,
181    message: MessageRef,
182    pub dropped: u64,
183}
184
185impl Message<'_> {
186    /// Returns the component and FXT bytes.  The FXT record is validated to be the correct length
187    /// and type.
188    pub fn parse(&self) -> Result<(&Arc<ComponentIdentity>, Header, &[u8]), MessageError> {
189        // SAFETY: We hold a lock which prevents the buffer from being drained so
190        // it should be safe to read this range.
191        let (container, data, _) =
192            unsafe { self.inner.parse_message(self.message.index..self.inner.last_scanned) };
193        let container_id = container;
194        let (mut header, _) = Header::read_from_prefix(data)
195            .map_err(|_| MessageError::from(ParseError::InvalidHeader))?;
196        let msg_len = header.size_words() as usize * 8;
197        if msg_len > data.len() || msg_len < 16 {
198            return Err(ParseError::ValueOutOfValidRange.into());
199        }
200        if header.raw_type() != TRACING_FORMAT_LOG_RECORD_TYPE {
201            return Err(ParseError::InvalidRecordType.into());
202        }
203        header.set_tag(container_id.0);
204        let container = self.inner.containers.get(container).unwrap();
205        Ok((&container.identity, header, &data[..msg_len]))
206    }
207}
208
209impl TryFrom<Message<'_>> for LogsData {
210    type Error = MessageError;
211
212    fn try_from(value: Message<'_>) -> Result<Self, Self::Error> {
213        let (container, _header, data) = value.parse()?;
214        let mut data = diagnostics_message::from_structured(container.as_ref().into(), data)?;
215        if value.dropped > 0 {
216            data.metadata
217                .errors
218                .get_or_insert(vec![])
219                .push(LogError::RolledOutLogs { count: value.dropped });
220        }
221        Ok(data)
222    }
223}
224
225/// Like FilterCursor but returns a stream of LogsData.
226#[pin_project]
227pub struct FilterCursorStream<T> {
228    #[pin]
229    cursor: FilterCursor,
230    phantom: PhantomData<T>,
231}
232
233impl<T> Stream for FilterCursorStream<T>
234where
235    T: for<'a> TryFrom<Message<'a>>,
236{
237    type Item = T;
238
239    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
240        let mut this = self.project();
241        loop {
242            match ready!(this.cursor.as_mut().poll_next(cx)) {
243                Some(item) => {
244                    if let Ok(data) = T::try_from(item) {
245                        return Poll::Ready(Some(data));
246                    }
247                    // The message is bad, just ignore it.
248                }
249                None => return Poll::Ready(None),
250            }
251        }
252    }
253}
254
255impl<T> FusedStream for FilterCursorStream<T>
256where
257    T: for<'a> TryFrom<Message<'a>>,
258{
259    fn is_terminated(&self) -> bool {
260        self.cursor.is_terminated()
261    }
262}
263
264impl<T> From<FilterCursor> for FilterCursorStream<T> {
265    fn from(cursor: FilterCursor) -> Self {
266        Self { cursor, phantom: PhantomData }
267    }
268}
269
270#[cfg(test)]
271mod tests {
272    use super::*;
273    use crate::logs::shared_buffer::{
274        InnerGuard, SharedBuffer, SharedBufferOptions, create_ring_buffer,
275    };
276    use crate::logs::stats::LogStreamStats;
277    use crate::logs::testing::make_message;
278    use assert_matches::assert_matches;
279    use fidl_fuchsia_diagnostics::StreamMode;
280    use futures::StreamExt;
281    use selectors::{FastError, parse_component_selector};
282    use std::future::poll_fn;
283    use std::pin::pin;
284    use std::time::Duration;
285
286    fn test_stats() -> Arc<LogStreamStats> {
287        Arc::new(LogStreamStats::new(
288            &fuchsia_inspect::Node::default(),
289            &ComponentIdentity::unknown(),
290        ))
291    }
292
293    #[fuchsia::test]
294    async fn cursor_basic() {
295        let buffer = SharedBuffer::new(
296            create_ring_buffer(65536),
297            Box::new(|_| {}),
298            Default::default(),
299            &fuchsia_inspect::Node::default(),
300        );
301        let container = buffer.new_container_buffer(Arc::new(vec!["a"].into()), test_stats());
302        let msg = make_message("a", None, zx::BootInstant::from_nanos(1));
303        container.push_back(msg.bytes());
304
305        let cursor = buffer.cursor(StreamMode::Snapshot, vec![]);
306        let mut stream = pin!(FilterCursorStream::<LogsData>::from(cursor));
307
308        let item = stream.next().await.unwrap();
309        assert_eq!(item.msg().unwrap(), "a");
310        assert!(stream.next().await.is_none());
311    }
312
313    #[fuchsia::test]
314    async fn cursor_filter() {
315        let buffer = SharedBuffer::new(
316            create_ring_buffer(65536),
317            Box::new(|_| {}),
318            Default::default(),
319            &fuchsia_inspect::Node::default(),
320        );
321        let container_a = buffer.new_container_buffer(Arc::new(vec!["a"].into()), test_stats());
322        let container_b = buffer.new_container_buffer(Arc::new(vec!["b"].into()), test_stats());
323
324        container_a.push_back(make_message("msg_a", None, zx::BootInstant::from_nanos(1)).bytes());
325        container_b.push_back(make_message("msg_b", None, zx::BootInstant::from_nanos(2)).bytes());
326
327        let selector = parse_component_selector::<FastError>("a").unwrap();
328        let cursor = buffer.cursor(StreamMode::Snapshot, vec![selector]);
329        let mut stream = pin!(FilterCursorStream::<LogsData>::from(cursor));
330
331        let item = stream.next().await.unwrap();
332        assert_eq!(item.msg().unwrap(), "msg_a");
333        assert!(stream.next().await.is_none());
334    }
335
336    #[fuchsia::test]
337    async fn cursor_subscribe() {
338        let buffer = SharedBuffer::new(
339            create_ring_buffer(65536),
340            Box::new(|_| {}),
341            Default::default(),
342            &fuchsia_inspect::Node::default(),
343        );
344        let container = buffer.new_container_buffer(Arc::new(vec!["a"].into()), test_stats());
345
346        let cursor = buffer.cursor(StreamMode::Subscribe, vec![]);
347        let mut stream = pin!(FilterCursorStream::<LogsData>::from(cursor));
348
349        assert!(futures::poll!(stream.next()).is_pending());
350
351        container.push_back(make_message("msg", None, zx::BootInstant::from_nanos(1)).bytes());
352
353        let item = stream.next().await.unwrap();
354        assert_eq!(item.msg().unwrap(), "msg");
355    }
356
357    #[fuchsia::test]
358    async fn cursor_snapshot_then_subscribe() {
359        let buffer = SharedBuffer::new(
360            create_ring_buffer(65536),
361            Box::new(|_| {}),
362            Default::default(),
363            &fuchsia_inspect::Node::default(),
364        );
365        let container = buffer.new_container_buffer(Arc::new(vec!["a"].into()), test_stats());
366
367        container.push_back(make_message("msg1", None, zx::BootInstant::from_nanos(1)).bytes());
368
369        let cursor = buffer.cursor(StreamMode::SnapshotThenSubscribe, vec![]);
370        let mut stream = pin!(FilterCursorStream::<LogsData>::from(cursor));
371
372        let item = stream.next().await.unwrap();
373        assert_eq!(item.msg().unwrap(), "msg1");
374
375        assert!(futures::poll!(stream.next()).is_pending());
376
377        container.push_back(make_message("msg2", None, zx::BootInstant::from_nanos(2)).bytes());
378
379        let item = stream.next().await.unwrap();
380        assert_eq!(item.msg().unwrap(), "msg2");
381    }
382
383    #[fuchsia::test]
384    async fn cursor_dropped_logs() {
385        let buffer = SharedBuffer::new(
386            create_ring_buffer(65536),
387            Box::new(|_| {}),
388            SharedBufferOptions { sleep_time: Duration::ZERO, ..Default::default() },
389            &fuchsia_inspect::Node::default(),
390        );
391        let container_a = buffer.new_container_buffer(Arc::new(vec!["a"].into()), test_stats());
392        let container_b = buffer.new_container_buffer(Arc::new(vec!["b"].into()), test_stats());
393
394        let msg = make_message("msg", None, zx::BootInstant::from_nanos(1));
395
396        container_a.push_back(msg.bytes()); // 1
397        container_a.push_back(msg.bytes()); // 2
398        container_a.push_back(msg.bytes()); // 3
399
400        let cursor = buffer.cursor(StreamMode::SnapshotThenSubscribe, vec![]);
401        let mut stream = pin!(FilterCursorStream::<LogsData>::from(cursor));
402
403        // Consume one
404        let item = stream.next().await.unwrap();
405        assert_eq!(item.msg().unwrap(), "msg");
406
407        let has_messages = async |component| {
408            let mut cursor = pin!(buffer.cursor(
409                StreamMode::Snapshot,
410                vec![parse_component_selector::<FastError>(component).unwrap()],
411            ));
412            poll_fn(|cx| cursor.as_mut().poll_next(cx).map(|i| i.is_some())).await
413        };
414
415        // Force roll out of remaining A messages
416        while has_messages("a").await {
417            container_b.push_back(msg.bytes());
418            // Force the buffer to pop old messages.
419            {
420                let mut inner = InnerGuard::new(&buffer);
421                inner.check_space(inner.ring_buffer.head());
422            }
423        }
424
425        // Next item should indicate the two dropped messages for container A.
426        let item = stream.next().await.unwrap();
427        assert_eq!(item.metadata.errors.as_ref().unwrap().len(), 1);
428        assert_matches!(
429            item.metadata.errors.as_ref().unwrap()[0],
430            LogError::RolledOutLogs { count } if count == 2
431        );
432    }
433
434    #[fuchsia::test]
435    async fn terminate_terminates_cursor() {
436        let buffer = SharedBuffer::new(
437            create_ring_buffer(65536),
438            Box::new(|_| {}),
439            SharedBufferOptions { sleep_time: Duration::from_secs(10), ..Default::default() },
440            &fuchsia_inspect::Node::default(),
441        );
442        let cursor = pin!(FilterCursorStream::<LogsData>::from(
443            buffer.cursor(StreamMode::SnapshotThenSubscribe, vec![])
444        ));
445        let container = buffer.new_container_buffer(Arc::new(vec!["a"].into()), test_stats());
446        container.push_back(make_message("msg", None, zx::BootInstant::from_nanos(1)).bytes());
447        drop(buffer.terminate());
448        assert_eq!(cursor.count().await, 1);
449    }
450}
archivist_lib/logs/shared_buffer/cursor.rs

archivist_lib/logs/shared_buffer/
cursor.rs
