| // Copyright 2018 The Fuchsia Authors. All rights reserved. |
| // Use of this source code is governed by a BSD-style license that can be |
| // found in the LICENSE file. |
| |
| use anyhow::{Context as _, Error}; |
| use fidl::endpoints::ClientEnd; |
| use fidl_fuchsia_logger::{ |
| LogFilterOptions, LogListenerSafeMarker, LogRequest, LogRequestStream, LogSinkRequest, |
| LogSinkRequestStream, |
| }; |
| use fidl_fuchsia_sys_internal::{ |
| LogConnection, LogConnectionListenerRequest, LogConnectorProxy, SourceIdentity, |
| }; |
| use fuchsia_async as fasync; |
| use fuchsia_inspect as inspect; |
| use fuchsia_inspect_derive::Inspect; |
| use fuchsia_zircon as zx; |
| use futures::{ |
| channel::mpsc, future, lock::Mutex, sink::SinkExt, FutureExt, StreamExt, TryStreamExt, |
| }; |
| use log::{error, warn}; |
| use std::sync::Arc; |
| |
| mod buffer; |
| mod debuglog; |
| mod error; |
| mod listener; |
| pub mod message; |
| mod socket; |
| mod stats; |
| |
| pub use debuglog::{convert_debuglog_to_log_message, KernelDebugLog}; |
| use listener::{pool::Pool, pretend_scary_listener_is_safe, Listener}; |
| use message::Message; |
| use socket::LogMessageSocket; |
| use stats::LogSource; |
| |
| /// Store 4 MB of log messages and delete on FIFO basis. |
| const OLD_MSGS_BUF_SIZE: usize = 4 * 1024 * 1024; |
| |
| /// The `LogManager` is responsible for brokering all logging in the archivist. |
| #[derive(Clone, Inspect)] |
| pub struct LogManager { |
| #[inspect(forward)] |
| inner: Arc<Mutex<ManagerInner>>, |
| } |
| |
| #[derive(Inspect)] |
| struct ManagerInner { |
| #[inspect(skip)] |
| listeners: Pool, |
| #[inspect(rename = "buffer_stats")] |
| log_msg_buffer: buffer::MemoryBoundedBuffer<Message>, |
| stats: stats::LogManagerStats, |
| inspect_node: inspect::Node, |
| } |
| |
| impl LogManager { |
| pub fn new() -> Self { |
| Self { |
| inner: Arc::new(Mutex::new(ManagerInner { |
| listeners: Pool::default(), |
| log_msg_buffer: buffer::MemoryBoundedBuffer::new(OLD_MSGS_BUF_SIZE), |
| stats: stats::LogManagerStats::new_detached(), |
| inspect_node: inspect::Node::default(), |
| })), |
| } |
| } |
| |
| /// Spawn a task to read from the kernel's debuglog. The returned future completes once existing |
| /// messages have been ingested. |
| pub async fn spawn_debuglog_drainer<K>(&self, klog_reader: K) -> Result<(), Error> |
| where |
| K: debuglog::DebugLog + Send + Sync + 'static, |
| { |
| let mut kernel_logger = debuglog::DebugLogBridge::create(klog_reader); |
| for log_msg in |
| kernel_logger.existing_logs().await.context("reading from kernel log iterator")? |
| { |
| self.ingest_message(log_msg, LogSource::Kernel).await; |
| } |
| |
| let manager = self.clone(); |
| fasync::spawn(async move { |
| let drain_result = kernel_logger |
| .listen() |
| .try_for_each(|log_msg| manager.ingest_message(log_msg, LogSource::Kernel).map(Ok)); |
| |
| if let Err(e) = drain_result.await { |
| error!( |
| "important logs may be missing from system log.\ |
| failed to drain kernel log: {:?}", |
| e |
| ); |
| } |
| }); |
| |
| Ok(()) |
| } |
| |
| /// Spawn a log sink for messages sent by the archivist itself. |
| pub fn spawn_internal_sink(&self, socket: zx::Socket, name: &str) -> Result<(), Error> { |
| // TODO(50105): Figure out how to properly populate SourceIdentity |
| let mut source = SourceIdentity::empty(); |
| source.component_name = Some(name.to_owned()); |
| let log_stream = LogMessageSocket::new(socket, source)?; |
| let manager = self.clone(); |
| fasync::spawn(async move { |
| manager.drain_messages(log_stream).await; |
| unreachable!(); |
| }); |
| Ok(()) |
| } |
| |
| /// Spawn a task which attempts to act as `LogConnectionListener` for its parent realm, eventually |
| /// passing `LogSink` connections into the manager. |
| /// TODO(49764): Make this run on main future |
| pub fn spawn_log_consumer(&self, connector: LogConnectorProxy) { |
| let handler = self.clone(); |
| fasync::spawn(async move { |
| match connector.take_log_connection_listener().await { |
| Ok(Some(listener)) => { |
| let mut connections = |
| listener.into_stream().expect("getting request stream from server end"); |
| while let Ok(Some(connection)) = connections.try_next().await { |
| match connection { |
| LogConnectionListenerRequest::OnNewConnection { |
| connection: LogConnection { log_request, source_identity }, |
| control_handle: _, |
| } => { |
| let fut = handler.clone().process_log_sink( |
| log_request |
| .into_stream() |
| .expect("getting LogSinkRequestStream from serverend"), |
| source_identity, |
| ); |
| |
| fasync::spawn(fut); |
| } |
| }; |
| } |
| } |
| Ok(None) => { |
| warn!("local realm already gave out LogConnectionListener, skipping logs") |
| } |
| Err(why) => { |
| error!("error retrieving LogConnectionListener from LogConnector: {:?}", why) |
| } |
| } |
| }); |
| } |
| |
| /// Process LogSink protocol on `stream`. The future returned by this |
| /// function will not complete before all messages on this connection are |
| /// processed. |
| pub async fn process_log_sink(self, mut stream: LogSinkRequestStream, source: SourceIdentity) { |
| if source.component_name.is_none() { |
| self.inner.lock().await.stats.record_unattributed(); |
| } |
| let (mut log_stream_sender, recv) = mpsc::channel(0); |
| let log_sink_fut = async move { |
| while let Some(LogSinkRequest::Connect { socket, control_handle }) = |
| stream.try_next().await? |
| { |
| let source = SourceIdentity { |
| component_name: source.component_name.clone(), |
| component_url: source.component_url.clone(), |
| instance_id: source.instance_id.clone(), |
| realm_path: source.realm_path.clone(), |
| }; |
| let log_stream = match LogMessageSocket::new(socket, source) |
| .context("creating log stream from socket") |
| { |
| Ok(s) => s, |
| Err(e) => { |
| control_handle.shutdown(); |
| return Err(e); |
| } |
| }; |
| log_stream_sender.send(log_stream).await?; |
| } |
| Ok(()) |
| }; |
| |
| let log_sink_fut = async move { |
| if let Err(e) = log_sink_fut.await { |
| eprintln!("logsink stream errored: {:?}", e); |
| } |
| }; |
| |
| // move it into async block else two futures are no compatiable to join. |
| let drain_messages_fut = async move { |
| recv.for_each_concurrent(None, |log_stream| async { |
| self.clone().drain_messages(log_stream).await; |
| }) |
| .await; |
| }; |
| |
| // Join future processing LogSink and future processing sockets so as to |
| // drain all the messages. |
| future::join(log_sink_fut, drain_messages_fut).await; |
| } |
| |
| /// Drain a `LoggerStream` which wraps a socket from a component generating logs. |
| async fn drain_messages(self, mut log_stream: LogMessageSocket) { |
| while let Some(next) = log_stream.next().await { |
| match next { |
| Ok(log_msg) => { |
| self.ingest_message(log_msg, stats::LogSource::LogSink).await; |
| } |
| Err(e) => { |
| self.inner.lock().await.stats.record_closed_stream(); |
| warn!("closing socket from {:?} due to error: {}", log_stream.source(), e); |
| return; |
| } |
| } |
| } |
| } |
| |
| /// Spawn a task to handle requests from components reading the shared log. |
| pub fn spawn_log_handler(&self, stream: LogRequestStream) { |
| let handler = self.clone(); |
| fasync::spawn(async move { |
| if let Err(e) = handler.handle_log_requests(stream).await { |
| warn!("error handling Log requests: {:?}", e); |
| } |
| }); |
| } |
| |
| /// Handle requests to `fuchsia.logger.Log`. All request types read the whole backlog from |
| /// memory, `DumpLogs(Safe)` stops listening after that though. |
| async fn handle_log_requests(self, mut stream: LogRequestStream) -> Result<(), Error> { |
| while let Some(request) = stream.try_next().await? { |
| let (listener, options, dump_logs) = match request { |
| LogRequest::ListenSafe { log_listener, options, .. } => { |
| (log_listener, options, false) |
| } |
| LogRequest::DumpLogsSafe { log_listener, options, .. } => { |
| (log_listener, options, true) |
| } |
| |
| // TODO(fxb/48758) delete these methods! |
| LogRequest::Listen { log_listener, options, .. } => { |
| warn!("Use of fuchsia.logger.Log.Listen. Use ListenSafe."); |
| let listener = pretend_scary_listener_is_safe(log_listener)?; |
| (listener, options, false) |
| } |
| LogRequest::DumpLogs { log_listener, options, .. } => { |
| warn!("Use of fuchsia.logger.Log.DumpLogs. Use DumpLogsSafe."); |
| let listener = pretend_scary_listener_is_safe(log_listener)?; |
| (listener, options, true) |
| } |
| }; |
| |
| self.handle_listener(listener, options, dump_logs).await?; |
| } |
| Ok(()) |
| } |
| |
| /// Handle a new listener, sending it all cached messages and either calling `Done` if |
| /// `dump_logs` is true or adding it to the pool of ongoing listeners if not. |
| async fn handle_listener( |
| &self, |
| log_listener: ClientEnd<LogListenerSafeMarker>, |
| options: Option<Box<LogFilterOptions>>, |
| dump_logs: bool, |
| ) -> Result<(), Error> { |
| let mut listener = Listener::new(log_listener, options)?; |
| |
| let mut inner = self.inner.lock().await; |
| listener.backfill(inner.log_msg_buffer.iter()).await; |
| |
| if !listener.is_healthy() { |
| warn!("listener dropped before we finished"); |
| return Ok(()); |
| } |
| |
| if dump_logs { |
| listener.done(); |
| } else { |
| inner.listeners.add(listener); |
| } |
| Ok(()) |
| } |
| |
| /// Ingest an individual log message. |
| async fn ingest_message(&self, log_msg: Message, source: stats::LogSource) { |
| let mut inner = self.inner.lock().await; |
| |
| inner.stats.record_log(&log_msg, source); |
| inner.listeners.send(&log_msg).await; |
| inner.log_msg_buffer.push(log_msg); |
| } |
| } |
| |
| #[cfg(test)] |
| mod tests { |
| use { |
| super::*, |
| crate::logs::debuglog::tests::{TestDebugEntry, TestDebugLog}, |
| crate::logs::message::fx_log_packet_t, |
| fidl_fuchsia_logger::{ |
| LogFilterOptions, LogLevelFilter, LogMarker, LogMessage, LogProxy, LogSinkMarker, |
| LogSinkProxy, |
| }, |
| fuchsia_inspect::assert_inspect_tree, |
| fuchsia_zircon as zx, |
| validating_log_listener::{validate_log_dump, validate_log_stream}, |
| }; |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn test_log_manager_simple() { |
| TestHarness::new().manager_test(false).await; |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn test_log_manager_dump() { |
| TestHarness::new().manager_test(true).await; |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn unfiltered_stats() { |
| let first_packet = setup_default_packet(); |
| let first_message = LogMessage { |
| pid: first_packet.metadata.pid, |
| tid: first_packet.metadata.tid, |
| time: first_packet.metadata.time, |
| dropped_logs: first_packet.metadata.dropped_logs, |
| severity: first_packet.metadata.severity, |
| msg: String::from("BBBBB"), |
| tags: vec![String::from("AAAAA")], |
| }; |
| |
| let (mut second_packet, mut second_message) = (first_packet.clone(), first_message.clone()); |
| second_packet.metadata.pid = 0; |
| second_message.pid = second_packet.metadata.pid; |
| |
| let (mut third_packet, mut third_message) = (second_packet.clone(), second_message.clone()); |
| third_packet.metadata.severity = LogLevelFilter::Info.into_primitive().into(); |
| third_message.severity = third_packet.metadata.severity; |
| |
| let (fourth_packet, fourth_message) = (third_packet.clone(), third_message.clone()); |
| |
| let (mut fifth_packet, mut fifth_message) = (fourth_packet.clone(), fourth_message.clone()); |
| fifth_packet.metadata.severity = LogLevelFilter::Error.into_primitive().into(); |
| fifth_message.severity = fifth_packet.metadata.severity; |
| |
| let log_stats_tree = TestHarness::new() |
| .filter_test( |
| vec![first_message, second_message, third_message, fourth_message, fifth_message], |
| vec![first_packet, second_packet, third_packet, fourth_packet, fifth_packet], |
| None, |
| ) |
| .await; |
| |
| assert_inspect_tree!( |
| log_stats_tree, |
| root: { |
| log_stats: { |
| total_logs: 5u64, |
| kernel_logs: 0u64, |
| logsink_logs: 5u64, |
| trace_logs: 0u64, |
| debug_logs: 0u64, |
| info_logs: 2u64, |
| warning_logs: 2u64, |
| error_logs: 1u64, |
| fatal_logs: 0u64, |
| closed_streams: 0u64, |
| unattributed_log_sinks: 1u64, |
| buffer_stats: { |
| rolled_out_entries: 0u64, |
| } |
| }, |
| } |
| ); |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn test_filter_by_pid() { |
| let p = setup_default_packet(); |
| let mut p2 = p.clone(); |
| p2.metadata.pid = 0; |
| let lm = LogMessage { |
| pid: p.metadata.pid, |
| tid: p.metadata.tid, |
| time: p.metadata.time, |
| dropped_logs: p.metadata.dropped_logs, |
| severity: p.metadata.severity, |
| msg: String::from("BBBBB"), |
| tags: vec![String::from("AAAAA")], |
| }; |
| let options = LogFilterOptions { |
| filter_by_pid: true, |
| pid: 1, |
| filter_by_tid: false, |
| tid: 0, |
| min_severity: LogLevelFilter::None, |
| verbosity: 0, |
| tags: vec![], |
| }; |
| |
| TestHarness::new().filter_test(vec![lm], vec![p, p2], Some(options)).await; |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn test_filter_by_tid() { |
| let mut p = setup_default_packet(); |
| p.metadata.pid = 0; |
| let mut p2 = p.clone(); |
| p2.metadata.tid = 0; |
| let lm = LogMessage { |
| pid: p.metadata.pid, |
| tid: p.metadata.tid, |
| time: p.metadata.time, |
| dropped_logs: p.metadata.dropped_logs, |
| severity: p.metadata.severity, |
| msg: String::from("BBBBB"), |
| tags: vec![String::from("AAAAA")], |
| }; |
| let options = LogFilterOptions { |
| filter_by_pid: false, |
| pid: 1, |
| filter_by_tid: true, |
| tid: 1, |
| min_severity: LogLevelFilter::None, |
| verbosity: 0, |
| tags: vec![], |
| }; |
| |
| TestHarness::new().filter_test(vec![lm], vec![p, p2], Some(options)).await; |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn test_filter_by_min_severity() { |
| let p = setup_default_packet(); |
| let mut p2 = p.clone(); |
| p2.metadata.pid = 0; |
| p2.metadata.tid = 0; |
| p2.metadata.severity = LogLevelFilter::Error.into_primitive().into(); |
| let mut p3 = p.clone(); |
| p3.metadata.severity = LogLevelFilter::Info.into_primitive().into(); |
| let mut p4 = p.clone(); |
| p4.metadata.severity = -22; |
| let mut p5 = p.clone(); |
| p5.metadata.severity = LogLevelFilter::Fatal.into_primitive().into(); |
| let lm = LogMessage { |
| pid: p2.metadata.pid, |
| tid: p2.metadata.tid, |
| time: p2.metadata.time, |
| dropped_logs: p2.metadata.dropped_logs, |
| severity: p2.metadata.severity, |
| msg: String::from("BBBBB"), |
| tags: vec![String::from("AAAAA")], |
| }; |
| let options = LogFilterOptions { |
| filter_by_pid: false, |
| pid: 1, |
| filter_by_tid: false, |
| tid: 1, |
| min_severity: LogLevelFilter::Error, |
| verbosity: 0, |
| tags: vec![], |
| }; |
| |
| TestHarness::new().filter_test(vec![lm], vec![p, p2, p3, p4, p5], Some(options)).await; |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn test_filter_by_combination() { |
| let mut p = setup_default_packet(); |
| p.metadata.pid = 0; |
| p.metadata.tid = 0; |
| let mut p2 = p.clone(); |
| p2.metadata.severity = LogLevelFilter::Error.into_primitive().into(); |
| let mut p3 = p.clone(); |
| p3.metadata.pid = 1; |
| let lm = LogMessage { |
| pid: p2.metadata.pid, |
| tid: p2.metadata.tid, |
| time: p2.metadata.time, |
| dropped_logs: p2.metadata.dropped_logs, |
| severity: p2.metadata.severity, |
| msg: String::from("BBBBB"), |
| tags: vec![String::from("AAAAA")], |
| }; |
| let options = LogFilterOptions { |
| filter_by_pid: true, |
| pid: 0, |
| filter_by_tid: false, |
| tid: 1, |
| min_severity: LogLevelFilter::Error, |
| verbosity: 0, |
| tags: vec![], |
| }; |
| |
| TestHarness::new().filter_test(vec![lm], vec![p, p2, p3], Some(options)).await; |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn test_filter_by_tags() { |
| let mut p = setup_default_packet(); |
| let mut p2 = p.clone(); |
| // p tags - "DDDDD" |
| memset(&mut p.data[..], 1, 68, 5); |
| |
| p2.metadata.pid = 0; |
| p2.metadata.tid = 0; |
| p2.data[6] = 5; |
| // p2 tag - "AAAAA", "BBBBB" |
| // p2 msg - "CCCCC" |
| memset(&mut p2.data[..], 13, 67, 5); |
| |
| let lm1 = LogMessage { |
| pid: p.metadata.pid, |
| tid: p.metadata.tid, |
| time: p.metadata.time, |
| dropped_logs: p.metadata.dropped_logs, |
| severity: p.metadata.severity, |
| msg: String::from("BBBBB"), |
| tags: vec![String::from("DDDDD")], |
| }; |
| let lm2 = LogMessage { |
| pid: p2.metadata.pid, |
| tid: p2.metadata.tid, |
| time: p2.metadata.time, |
| dropped_logs: p2.metadata.dropped_logs, |
| severity: p2.metadata.severity, |
| msg: String::from("CCCCC"), |
| tags: vec![String::from("AAAAA"), String::from("BBBBB")], |
| }; |
| let options = LogFilterOptions { |
| filter_by_pid: false, |
| pid: 1, |
| filter_by_tid: false, |
| tid: 1, |
| min_severity: LogLevelFilter::None, |
| verbosity: 0, |
| tags: vec![String::from("BBBBB"), String::from("DDDDD")], |
| }; |
| |
| TestHarness::new().filter_test(vec![lm1, lm2], vec![p, p2], Some(options)).await; |
| } |
| |
| #[fasync::run_singlethreaded(test)] |
| async fn test_debuglog_drainer() { |
| let log1 = TestDebugEntry::new("log1".as_bytes()); |
| let log2 = TestDebugEntry::new("log2".as_bytes()); |
| let log3 = TestDebugEntry::new("log3".as_bytes()); |
| |
| let klog_reader = TestDebugLog::new(); |
| klog_reader.enqueue_read_entry(&log1); |
| klog_reader.enqueue_read_entry(&log2); |
| // logs recieved after kernel indicates no logs should be read |
| klog_reader.enqueue_read_fail(zx::Status::SHOULD_WAIT); |
| klog_reader.enqueue_read_entry(&log3); |
| klog_reader.enqueue_read_fail(zx::Status::SHOULD_WAIT); |
| |
| let expected_logs = vec![ |
| LogMessage { |
| pid: log1.pid, |
| tid: log1.tid, |
| time: log1.timestamp, |
| dropped_logs: 0, |
| severity: fidl_fuchsia_logger::LogLevelFilter::Info as i32, |
| msg: String::from("log1"), |
| tags: vec![String::from("klog")], |
| }, |
| LogMessage { |
| pid: log2.pid, |
| tid: log2.tid, |
| time: log2.timestamp, |
| dropped_logs: 0, |
| severity: fidl_fuchsia_logger::LogLevelFilter::Info as i32, |
| msg: String::from("log2"), |
| tags: vec![String::from("klog")], |
| }, |
| LogMessage { |
| pid: log3.pid, |
| tid: log3.tid, |
| time: log3.timestamp, |
| dropped_logs: 0, |
| severity: fidl_fuchsia_logger::LogLevelFilter::Info as i32, |
| msg: String::from("log3"), |
| tags: vec![String::from("klog")], |
| }, |
| ]; |
| |
| let klog_stats_tree = debuglog_test(expected_logs, klog_reader).await; |
| assert_inspect_tree!( |
| klog_stats_tree, |
| root: { |
| log_stats: contains { |
| total_logs: 3u64, |
| kernel_logs: 3u64, |
| logsink_logs: 0u64, |
| trace_logs: 0u64, |
| debug_logs: 0u64, |
| info_logs: 3u64, |
| warning_logs: 0u64, |
| error_logs: 0u64, |
| fatal_logs: 0u64, |
| closed_streams: 0u64, |
| unattributed_log_sinks: 0u64, |
| } |
| } |
| ); |
| } |
| |
| struct TestHarness { |
| log_proxy: LogProxy, |
| sin: zx::Socket, |
| inspector: inspect::Inspector, |
| _log_sink_proxy: LogSinkProxy, |
| } |
| |
| impl TestHarness { |
| fn new() -> Self { |
| let inspector = inspect::Inspector::new(); |
| let (sin, sout) = zx::Socket::create(zx::SocketOpts::DATAGRAM).unwrap(); |
| let inner = Arc::new(Mutex::new(ManagerInner { |
| listeners: Pool::default(), |
| log_msg_buffer: buffer::MemoryBoundedBuffer::new(OLD_MSGS_BUF_SIZE), |
| stats: stats::LogManagerStats::new_detached(), |
| inspect_node: inspect::Node::default(), |
| })); |
| |
| let mut lm = LogManager { inner }; |
| lm.iattach(inspector.root(), "log_stats").unwrap(); |
| |
| let (log_proxy, log_stream) = |
| fidl::endpoints::create_proxy_and_stream::<LogMarker>().unwrap(); |
| lm.spawn_log_handler(log_stream); |
| |
| let (log_sink_proxy, log_sink_stream) = |
| fidl::endpoints::create_proxy_and_stream::<LogSinkMarker>().unwrap(); |
| |
| let manager = lm.clone(); |
| fasync::spawn(async move { |
| manager.process_log_sink(log_sink_stream, SourceIdentity::empty()).await; |
| }); |
| |
| log_sink_proxy.connect(sout).expect("unable to connect out socket to log sink"); |
| |
| Self { log_proxy, _log_sink_proxy: log_sink_proxy, sin, inspector } |
| } |
| |
| /// Run a filter test, returning the Inspector to check Inspect output. |
| async fn filter_test( |
| self, |
| expected: impl IntoIterator<Item = LogMessage>, |
| packets: Vec<fx_log_packet_t>, |
| filter_options: Option<LogFilterOptions>, |
| ) -> inspect::Inspector { |
| for mut p in packets { |
| self.sin.write(to_u8_slice(&mut p)).unwrap(); |
| } |
| |
| validate_log_stream(expected, self.log_proxy, filter_options).await; |
| self.inspector |
| } |
| |
| async fn manager_test(self, test_dump_logs: bool) { |
| let mut p = setup_default_packet(); |
| self.sin.write(to_u8_slice(&mut p)).unwrap(); |
| |
| p.metadata.severity = LogLevelFilter::Info.into_primitive().into(); |
| self.sin.write(to_u8_slice(&mut p)).unwrap(); |
| |
| let mut lm1 = LogMessage { |
| time: p.metadata.time, |
| pid: p.metadata.pid, |
| tid: p.metadata.tid, |
| dropped_logs: p.metadata.dropped_logs, |
| severity: p.metadata.severity, |
| msg: String::from("BBBBB"), |
| tags: vec![String::from("AAAAA")], |
| }; |
| let lm2 = copy_log_message(&lm1); |
| lm1.severity = LogLevelFilter::Warn.into_primitive().into(); |
| let mut lm3 = copy_log_message(&lm2); |
| lm3.pid = 2; |
| |
| p.metadata.pid = 2; |
| self.sin.write(to_u8_slice(&mut p)).unwrap(); |
| |
| if test_dump_logs { |
| validate_log_dump(vec![lm1, lm2, lm3], self.log_proxy, None).await; |
| } else { |
| validate_log_stream(vec![lm1, lm2, lm3], self.log_proxy, None).await; |
| } |
| } |
| } |
| |
| /// Run a test on logs from klog, returning the inspector object. |
| async fn debuglog_test( |
| expected: impl IntoIterator<Item = LogMessage>, |
| debug_log: TestDebugLog, |
| ) -> inspect::Inspector { |
| let inspector = inspect::Inspector::new(); |
| let inner = Arc::new(Mutex::new(ManagerInner { |
| listeners: Pool::default(), |
| log_msg_buffer: buffer::MemoryBoundedBuffer::new(OLD_MSGS_BUF_SIZE), |
| stats: stats::LogManagerStats::new_detached(), |
| inspect_node: inspect::Node::default(), |
| })); |
| |
| let mut lm = LogManager { inner }; |
| lm.iattach(inspector.root(), "log_stats").unwrap(); |
| let (log_proxy, log_stream) = |
| fidl::endpoints::create_proxy_and_stream::<LogMarker>().unwrap(); |
| lm.spawn_log_handler(log_stream); |
| lm.spawn_debuglog_drainer(debug_log).await.unwrap(); |
| |
| validate_log_stream(expected, log_proxy, None).await; |
| inspector |
| } |
| |
| fn setup_default_packet() -> fx_log_packet_t { |
| let mut p: fx_log_packet_t = Default::default(); |
| p.metadata.pid = 1; |
| p.metadata.tid = 1; |
| p.metadata.severity = LogLevelFilter::Warn.into_primitive().into(); |
| p.metadata.dropped_logs = 2; |
| p.data[0] = 5; |
| memset(&mut p.data[..], 1, 65, 5); |
| memset(&mut p.data[..], 7, 66, 5); |
| return p; |
| } |
| |
| fn copy_log_message(log_message: &LogMessage) -> LogMessage { |
| LogMessage { |
| pid: log_message.pid, |
| tid: log_message.tid, |
| time: log_message.time, |
| severity: log_message.severity, |
| dropped_logs: log_message.dropped_logs, |
| tags: log_message.tags.clone(), |
| msg: log_message.msg.clone(), |
| } |
| } |
| |
| /// Function to convert fx_log_packet_t to &[u8]. |
| /// This function is safe as it works on `fx_log_packet_t` which |
| /// doesn't have any uninitialized padding bits. |
| fn to_u8_slice(p: &fx_log_packet_t) -> &[u8] { |
| // This code just converts to &[u8] so no need to explicity drop it as memory |
| // location would be freed as soon as p is dropped. |
| unsafe { |
| ::std::slice::from_raw_parts( |
| (p as *const fx_log_packet_t) as *const u8, |
| ::std::mem::size_of::<fx_log_packet_t>(), |
| ) |
| } |
| } |
| |
| fn memset<T: Copy>(x: &mut [T], offset: usize, value: T, size: usize) { |
| x[offset..(offset + size)].iter_mut().for_each(|x| *x = value); |
| } |
| } |
