| // Copyright 2019 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 { |
| crate::{ |
| container::ComponentDiagnostics, |
| error::Error, |
| events::types::UniqueKey, |
| identity::ComponentIdentity, |
| inspect::container::{InspectArtifactsContainer, UnpopulatedInspectDataContainer}, |
| lifecycle::container::{LifecycleArtifactsContainer, LifecycleDataContainer}, |
| logs::{ |
| budget::BudgetManager, |
| container::LogsArtifactsContainer, |
| debuglog::{DebugLog, DebugLogBridge, KERNEL_IDENTITY}, |
| error::LogsError, |
| listener::Listener, |
| multiplex::{Multiplexer, MultiplexerHandle}, |
| }, |
| ImmutableString, |
| }, |
| diagnostics_data::LogsData, |
| diagnostics_hierarchy::{ |
| trie::{self, TrieIterableNode}, |
| InspectHierarchyMatcher, |
| }, |
| fidl::endpoints::ProtocolMarker, |
| fidl_fuchsia_diagnostics::{ |
| self, LogInterestSelector, LogSettingsMarker, LogSettingsRequest, LogSettingsRequestStream, |
| Selector, StreamMode, |
| }, |
| fidl_fuchsia_io as fio, |
| fidl_fuchsia_logger::{LogMarker, LogRequest, LogRequestStream}, |
| fuchsia_async as fasync, fuchsia_inspect as inspect, fuchsia_zircon as zx, |
| futures::channel::mpsc, |
| futures::prelude::*, |
| io_util, |
| lazy_static::lazy_static, |
| parking_lot::{Mutex, RwLock}, |
| selectors, |
| std::{ |
| collections::{BTreeMap, HashMap}, |
| sync::{ |
| atomic::{AtomicUsize, Ordering}, |
| Arc, |
| }, |
| }, |
| tracing::{debug, error, warn}, |
| }; |
| |
| lazy_static! { |
| static ref CONNECTION_ID: AtomicUsize = AtomicUsize::new(0); |
| } |
| |
| /// DataRepo holds all diagnostics data and is a singleton wrapped by multiple |
| /// [`pipeline::Pipeline`]s in a given Archivist instance. |
| #[derive(Clone)] |
| pub struct DataRepo { |
| inner: Arc<RwLock<DataRepoState>>, |
| } |
| |
| impl std::ops::Deref for DataRepo { |
| type Target = RwLock<DataRepoState>; |
| fn deref(&self) -> &Self::Target { |
| &*self.inner |
| } |
| } |
| |
| #[cfg(test)] |
| impl Default for DataRepo { |
| fn default() -> Self { |
| let budget = BudgetManager::new(crate::constants::LEGACY_DEFAULT_MAXIMUM_CACHED_LOGS_BYTES); |
| DataRepo { inner: DataRepoState::new(budget, &Default::default()) } |
| } |
| } |
| |
| impl DataRepo { |
| pub fn new(logs_budget: &BudgetManager, parent: &fuchsia_inspect::Node) -> Self { |
| DataRepo { inner: DataRepoState::new(logs_budget.clone(), parent) } |
| } |
| |
| /// Drain the kernel's debug log. The returned future completes once |
| /// existing messages have been ingested. |
| pub async fn drain_debuglog<K>(self, klog_reader: K) |
| where |
| K: DebugLog + Send + Sync + 'static, |
| { |
| debug!("Draining debuglog."); |
| let container = self.write().get_log_container(KERNEL_IDENTITY.clone()); |
| let mut kernel_logger = DebugLogBridge::create(klog_reader); |
| let mut messages = match kernel_logger.existing_logs().await { |
| Ok(messages) => messages, |
| Err(e) => { |
| error!(%e, "failed to read from kernel log, important logs may be missing"); |
| return; |
| } |
| }; |
| messages.sort_by_key(|m| m.timestamp()); |
| for message in messages { |
| container.ingest_message(message); |
| } |
| |
| let res = kernel_logger |
| .listen() |
| .try_for_each(|message| async { |
| container.ingest_message(message); |
| Ok(()) |
| }) |
| .await; |
| if let Err(e) = res { |
| error!(%e, "failed to drain kernel log, important logs may be missing"); |
| } |
| } |
| |
| /// Spawn a task to handle requests from components reading the shared log. |
| pub fn handle_log( |
| self, |
| stream: LogRequestStream, |
| sender: mpsc::UnboundedSender<fasync::Task<()>>, |
| ) { |
| if let Err(e) = sender.clone().unbounded_send(fasync::Task::spawn(async move { |
| if let Err(e) = self.handle_log_requests(stream, sender).await { |
| warn!("error handling Log requests: {}", e); |
| } |
| })) { |
| warn!("Couldn't queue listener task: {:?}", e); |
| } |
| } |
| |
| /// Handle requests to `fuchsia.logger.Log`. All request types read the |
| /// whole backlog from memory, `DumpLogs(Safe)` stops listening after that. |
| async fn handle_log_requests( |
| self, |
| mut stream: LogRequestStream, |
| mut sender: mpsc::UnboundedSender<fasync::Task<()>>, |
| ) -> Result<(), LogsError> { |
| let connection_id = CONNECTION_ID.fetch_add(1, Ordering::Relaxed); |
| while let Some(request) = stream.next().await { |
| let request = request.map_err(|source| LogsError::HandlingRequests { |
| protocol: LogMarker::NAME, |
| source, |
| })?; |
| |
| let (listener, options, dump_logs, selectors) = match request { |
| LogRequest::ListenSafe { log_listener, options, .. } => { |
| (log_listener, options, false, None) |
| } |
| LogRequest::DumpLogsSafe { log_listener, options, .. } => { |
| (log_listener, options, true, None) |
| } |
| |
| LogRequest::ListenSafeWithSelectors { |
| log_listener, options, selectors, .. |
| } => (log_listener, options, false, Some(selectors)), |
| }; |
| |
| let listener = Listener::new(listener, options)?; |
| let mode = |
| if dump_logs { StreamMode::Snapshot } else { StreamMode::SnapshotThenSubscribe }; |
| let logs = self.logs_cursor(mode, None); |
| if let Some(s) = selectors { |
| self.write().update_logs_interest(connection_id, s); |
| } |
| |
| sender.send(listener.spawn(logs, dump_logs)).await.ok(); |
| } |
| self.write().finish_interest_connection(connection_id); |
| Ok(()) |
| } |
| |
| pub async fn handle_log_settings( |
| self, |
| mut stream: LogSettingsRequestStream, |
| ) -> Result<(), LogsError> { |
| let connection_id = CONNECTION_ID.fetch_add(1, Ordering::Relaxed); |
| while let Some(request) = stream.next().await { |
| let request = request.map_err(|source| LogsError::HandlingRequests { |
| protocol: LogSettingsMarker::NAME, |
| source, |
| })?; |
| match request { |
| LogSettingsRequest::RegisterInterest { selectors, .. } => { |
| self.write().update_logs_interest(connection_id, selectors); |
| } |
| } |
| } |
| self.write().finish_interest_connection(connection_id); |
| |
| Ok(()) |
| } |
| |
| pub fn logs_cursor( |
| &self, |
| mode: StreamMode, |
| selectors: Option<Vec<Selector>>, |
| ) -> impl Stream<Item = Arc<LogsData>> + Send + 'static { |
| let mut repo = self.write(); |
| let (mut merged, mpx_handle) = Multiplexer::new(); |
| if let Some(selectors) = selectors { |
| merged.set_selectors(selectors); |
| } |
| repo.data_directories |
| .iter() |
| .filter_map(|(_, c)| c) |
| .filter_map(|c| { |
| c.logs_cursor(mode).map(|cursor| (c.identity.relative_moniker.clone(), cursor)) |
| }) |
| .for_each(|(n, c)| { |
| mpx_handle.send(n, c); |
| }); |
| repo.logs_multiplexers.add(mode, mpx_handle); |
| merged.set_on_drop_id_sender(repo.logs_multiplexers.cleanup_sender()); |
| |
| merged |
| } |
| |
| /// Returns `true` if a container exists for the requested `identity` and that container either |
| /// corresponds to a running component or we've decided to still retain it. |
| pub fn is_live(&self, identity: &ComponentIdentity) -> bool { |
| let this = self.read(); |
| if let Some(containers) = this.data_directories.get(&identity.unique_key().into()) { |
| let diagnostics_containers = containers.get_values(); |
| diagnostics_containers.len() == 1 && diagnostics_containers[0].should_retain() |
| } else { |
| false |
| } |
| } |
| |
| /// Stop accepting new messages, ensuring that pending Cursors return Poll::Ready(None) after |
| /// consuming any messages received before this call. |
| pub fn terminate_logs(&self) { |
| let mut repo = self.write(); |
| for container in repo.data_directories.iter().filter_map(|(_, v)| v) { |
| container.terminate_logs(); |
| } |
| repo.logs_multiplexers.terminate(); |
| } |
| } |
| |
| pub struct DataRepoState { |
| pub data_directories: trie::Trie<String, ComponentDiagnostics>, |
| inspect_node: inspect::Node, |
| |
| /// A reference to the budget manager, kept to be passed to containers. |
| logs_budget: BudgetManager, |
| /// The current global interest in logs, as defined by the last client to send us selectors. |
| logs_interest: Vec<LogInterestSelector>, |
| /// BatchIterators for logs need to be made aware of new components starting and their logs. |
| logs_multiplexers: MultiplexerBroker, |
| |
| /// Interest registrations that we have received through fuchsia.logger.Log/ListWithSelectors |
| /// or through fuchsia.logger.LogSettings/RegisterInterest. |
| interest_registrations: BTreeMap<usize, Vec<LogInterestSelector>>, |
| } |
| |
| impl DataRepoState { |
| fn new(logs_budget: BudgetManager, parent: &fuchsia_inspect::Node) -> Arc<RwLock<Self>> { |
| Arc::new(RwLock::new(Self { |
| inspect_node: parent.create_child("sources"), |
| data_directories: trie::Trie::new(), |
| logs_budget, |
| logs_interest: vec![], |
| logs_multiplexers: MultiplexerBroker::new(), |
| interest_registrations: BTreeMap::new(), |
| })) |
| } |
| |
| pub fn mark_stopped(&mut self, key: &UniqueKey) { |
| if let Some(containers) = self.data_directories.get_mut(key) { |
| let diagnostics_containers = containers.get_values_mut(); |
| if diagnostics_containers.len() == 1 { |
| diagnostics_containers[0].mark_stopped(); |
| } |
| } |
| } |
| |
| pub fn add_new_component( |
| &mut self, |
| identity: ComponentIdentity, |
| event_timestamp: zx::Time, |
| ) -> Result<(), Error> { |
| let lifecycle_artifact_container = LifecycleArtifactsContainer { event_timestamp }; |
| |
| let unique_key: Vec<_> = identity.unique_key().into(); |
| let diag_repo_entry_opt = self.data_directories.get_mut(&unique_key); |
| match diag_repo_entry_opt { |
| Some(diag_repo_entry) => { |
| let diag_repo_entry_values: &mut [ComponentDiagnostics] = |
| diag_repo_entry.get_values_mut(); |
| |
| match &mut *diag_repo_entry_values { |
| [] => { |
| // An entry with no values implies that the somehow we observed the |
| // creation of a component lower in the topology before observing this |
| // one. If this is the case, just instantiate as though it's our first |
| // time encountering this moniker segment. |
| self.data_directories.insert( |
| unique_key, |
| ComponentDiagnostics::new_with_lifecycle( |
| Arc::new(identity), |
| lifecycle_artifact_container, |
| &self.inspect_node, |
| ), |
| ) |
| } |
| [existing_diagnostics_artifact_container] => { |
| // Races may occur between seeing diagnostics ready and seeing |
| // creation lifecycle events. Handle this here. |
| // TODO(fxbug.dev/52047): Remove once caching handles ordering issues. |
| existing_diagnostics_artifact_container.mark_started(); |
| if existing_diagnostics_artifact_container.lifecycle.is_none() { |
| existing_diagnostics_artifact_container.lifecycle = |
| Some(lifecycle_artifact_container); |
| } |
| } |
| _ => { |
| return Err(Error::MultipleArtifactContainers(unique_key)); |
| } |
| } |
| } |
| // This case is expected to be the most common case. We've seen a creation |
| // lifecycle event and it promotes the instantiation of a new data repository entry. |
| None => self.data_directories.insert( |
| unique_key, |
| ComponentDiagnostics::new_with_lifecycle( |
| Arc::new(identity), |
| lifecycle_artifact_container, |
| &self.inspect_node, |
| ), |
| ), |
| } |
| Ok(()) |
| } |
| |
| /// Returns a container for logs artifacts, constructing one and adding it to the trie if |
| /// necessary. |
| pub fn get_log_container( |
| &mut self, |
| identity: ComponentIdentity, |
| ) -> Arc<LogsArtifactsContainer> { |
| let trie_key: Vec<_> = identity.unique_key().into(); |
| |
| // we use a macro instead of a closure to avoid lifetime issues |
| macro_rules! insert_component { |
| () => {{ |
| let mut to_insert = |
| ComponentDiagnostics::empty(Arc::new(identity), &self.inspect_node); |
| let logs = to_insert.logs( |
| &self.logs_budget, |
| &self.logs_interest, |
| &mut self.logs_multiplexers, |
| ); |
| self.data_directories.insert(trie_key, to_insert); |
| logs |
| }}; |
| } |
| |
| match self.data_directories.get_mut(&trie_key) { |
| Some(component) => match &mut component.get_values_mut()[..] { |
| [] => insert_component!(), |
| [existing] => existing.logs( |
| &self.logs_budget, |
| &self.logs_interest, |
| &mut self.logs_multiplexers, |
| ), |
| _ => unreachable!("invariant: each trie node has 0-1 entries"), |
| }, |
| None => insert_component!(), |
| } |
| } |
| |
| pub fn update_logs_interest( |
| &mut self, |
| connection_id: usize, |
| selectors: Vec<LogInterestSelector>, |
| ) { |
| let previous_selectors = |
| self.interest_registrations.insert(connection_id, selectors).unwrap_or_default(); |
| // unwrap safe, we just inserted. |
| let new_selectors = self.interest_registrations.get(&connection_id).unwrap(); |
| for (_, dir) in self.data_directories.iter() { |
| if let Some(dir) = dir { |
| if let Some(logs) = &dir.logs { |
| logs.update_interest(new_selectors, &previous_selectors); |
| } |
| } |
| } |
| } |
| |
| pub fn finish_interest_connection(&mut self, connection_id: usize) { |
| let selectors = self.interest_registrations.remove(&connection_id); |
| if let Some(selectors) = selectors { |
| for (_, dir) in self.data_directories.iter() { |
| if let Some(dir) = dir { |
| if let Some(logs) = &dir.logs { |
| logs.reset_interest(&selectors); |
| } |
| } |
| } |
| } |
| } |
| |
| pub fn add_inspect_artifacts( |
| &mut self, |
| identity: ComponentIdentity, |
| directory_proxy: fio::DirectoryProxy, |
| event_timestamp: zx::Time, |
| ) -> Result<(), Error> { |
| let inspect_container = InspectArtifactsContainer { |
| component_diagnostics_proxy: directory_proxy, |
| event_timestamp, |
| }; |
| |
| self.insert_inspect_artifact_container(inspect_container, identity) |
| } |
| |
| // Inserts an InspectArtifactsContainer into the data repository. |
| fn insert_inspect_artifact_container( |
| &mut self, |
| inspect_container: InspectArtifactsContainer, |
| identity: ComponentIdentity, |
| ) -> Result<(), Error> { |
| let unique_key: Vec<_> = identity.unique_key().into(); |
| let diag_repo_entry_opt = self.data_directories.get_mut(&unique_key); |
| |
| match diag_repo_entry_opt { |
| Some(diag_repo_entry) => { |
| let diag_repo_entry_values: &mut [ComponentDiagnostics] = |
| diag_repo_entry.get_values_mut(); |
| |
| match &mut *diag_repo_entry_values { |
| [] => { |
| // An entry with no values implies that the somehow we observed the |
| // creation of a component lower in the topology before observing this |
| // one. If this is the case, just instantiate as though it's our first |
| // time encountering this moniker segment. |
| self.data_directories.insert( |
| unique_key, |
| ComponentDiagnostics::new_with_inspect( |
| Arc::new(identity), |
| inspect_container, |
| &self.inspect_node, |
| ), |
| ) |
| } |
| [existing_diagnostics_artifact_container] => { |
| // Races may occur between synthesized and real diagnostics_ready |
| // events, so we must handle de-duplication here. |
| // TODO(fxbug.dev/52047): Remove once caching handles ordering issues. |
| existing_diagnostics_artifact_container.mark_started(); |
| if existing_diagnostics_artifact_container.inspect.is_none() { |
| // This is expected to be the most common case. We've encountered the |
| // diagnostics_ready event for a component that has already been |
| // observed to be started/existing. We now must update the diagnostics |
| // artifact container with the inspect artifacts that accompanied the |
| // diagnostics_ready event. |
| existing_diagnostics_artifact_container.inspect = |
| Some(inspect_container); |
| } |
| } |
| _ => { |
| return Err(Error::MultipleArtifactContainers(unique_key)); |
| } |
| } |
| } |
| // This case is expected to be uncommon; we've encountered a diagnostics_ready |
| // event before a start or existing event! |
| None => self.data_directories.insert( |
| unique_key, |
| ComponentDiagnostics::new_with_inspect( |
| Arc::new(identity), |
| inspect_container, |
| &self.inspect_node, |
| ), |
| ), |
| } |
| Ok(()) |
| } |
| |
| pub fn fetch_lifecycle_event_data(&self) -> Vec<LifecycleDataContainer> { |
| self.data_directories.iter().fold( |
| Vec::new(), |
| |mut acc, (_, diagnostics_artifacts_container_opt)| { |
| match diagnostics_artifacts_container_opt { |
| None => acc, |
| Some(diagnostics_artifacts_container) => { |
| if let Some(lifecycle_artifacts) = |
| &diagnostics_artifacts_container.lifecycle |
| { |
| acc.push(LifecycleDataContainer::from_lifecycle_artifact( |
| lifecycle_artifacts, |
| diagnostics_artifacts_container.identity.clone(), |
| )); |
| } |
| |
| if let Some(inspect_artifacts) = &diagnostics_artifacts_container.inspect { |
| acc.push(LifecycleDataContainer::from_inspect_artifact( |
| inspect_artifacts, |
| diagnostics_artifacts_container.identity.clone(), |
| )); |
| } |
| |
| if let Some(log_artifacts) = &diagnostics_artifacts_container.logs { |
| acc.push(LifecycleDataContainer::from_logs_sink_connected_artifact( |
| log_artifacts, |
| diagnostics_artifacts_container.identity.clone(), |
| )); |
| } |
| |
| acc |
| } |
| } |
| }, |
| ) |
| } |
| |
| /// Return all of the DirectoryProxies that contain Inspect hierarchies |
| /// which contain data that should be selected from. |
| pub fn fetch_inspect_data( |
| &self, |
| component_selectors: &Option<Vec<Selector>>, |
| moniker_to_static_matcher_map: Option<&HashMap<ImmutableString, InspectHierarchyMatcher>>, |
| ) -> Vec<UnpopulatedInspectDataContainer> { |
| return self |
| .data_directories |
| .iter() |
| .filter_map(|(_, diagnostics_artifacts_container_opt)| { |
| let (diagnostics_artifacts_container, inspect_artifacts) = |
| match &diagnostics_artifacts_container_opt { |
| Some(diagnostics_artifacts_container) => { |
| match &diagnostics_artifacts_container.inspect { |
| Some(inspect_artifacts) => { |
| (diagnostics_artifacts_container, inspect_artifacts) |
| } |
| None => return None, |
| } |
| } |
| None => return None, |
| }; |
| |
| let optional_hierarchy_matcher = match moniker_to_static_matcher_map { |
| Some(map) => { |
| match map.get( |
| diagnostics_artifacts_container |
| .identity |
| .relative_moniker |
| .join("/") |
| .as_str(), |
| ) { |
| Some(inspect_matcher) => Some(inspect_matcher), |
| // Return early if there were static selectors, and none were for this |
| // moniker. |
| None => return None, |
| } |
| } |
| None => None, |
| }; |
| |
| // Verify that the dynamic selectors contain an entry that applies to |
| // this moniker as well. |
| if !match component_selectors { |
| Some(component_selectors) => component_selectors.iter().any(|s| { |
| selectors::match_component_moniker_against_selector( |
| &diagnostics_artifacts_container.identity.relative_moniker, |
| s, |
| ) |
| .ok() |
| .unwrap_or(false) |
| }), |
| None => true, |
| } { |
| return None; |
| } |
| |
| // This artifact contains inspect and matches a passed selector. |
| io_util::clone_directory( |
| &inspect_artifacts.component_diagnostics_proxy, |
| fio::OpenFlags::CLONE_SAME_RIGHTS, |
| ) |
| .ok() |
| .map(|directory| UnpopulatedInspectDataContainer { |
| identity: diagnostics_artifacts_container.identity.clone(), |
| component_diagnostics_proxy: directory, |
| inspect_matcher: optional_hierarchy_matcher.cloned(), |
| }) |
| }) |
| .collect(); |
| } |
| } |
| |
| type LiveIteratorsMap = HashMap<usize, (StreamMode, MultiplexerHandle<Arc<LogsData>>)>; |
| |
| /// Ensures that BatchIterators get access to logs from newly started components. |
| pub struct MultiplexerBroker { |
| live_iterators: Arc<Mutex<LiveIteratorsMap>>, |
| cleanup_sender: mpsc::UnboundedSender<usize>, |
| _live_iterators_cleanup_task: fasync::Task<()>, |
| } |
| |
| impl MultiplexerBroker { |
| fn new() -> Self { |
| let (cleanup_sender, mut receiver) = mpsc::unbounded(); |
| let live_iterators = Arc::new(Mutex::new(HashMap::new())); |
| let live_iterators_clone = live_iterators.clone(); |
| Self { |
| live_iterators, |
| cleanup_sender, |
| _live_iterators_cleanup_task: fasync::Task::spawn(async move { |
| while let Some(id) = receiver.next().await { |
| live_iterators_clone.lock().remove(&id); |
| } |
| }), |
| } |
| } |
| |
| fn cleanup_sender(&self) -> mpsc::UnboundedSender<usize> { |
| self.cleanup_sender.clone() |
| } |
| |
| /// A new BatchIterator has been created and must be notified when future log containers are |
| /// created. |
| fn add(&mut self, mode: StreamMode, recipient: MultiplexerHandle<Arc<LogsData>>) { |
| match mode { |
| // snapshot streams only want to know about what's currently available |
| StreamMode::Snapshot => recipient.close(), |
| StreamMode::SnapshotThenSubscribe | StreamMode::Subscribe => { |
| self.live_iterators.lock().insert(recipient.multiplexer_id(), (mode, recipient)); |
| } |
| } |
| } |
| |
| /// Notify existing BatchIterators of a new logs container so they can include its messages |
| /// in their results. |
| pub fn send(&mut self, container: &Arc<LogsArtifactsContainer>) { |
| self.live_iterators.lock().retain(|_, (mode, recipient)| { |
| recipient.send(container.identity.relative_moniker.clone(), container.cursor(*mode)) |
| }); |
| } |
| |
| /// Notify all multiplexers to terminate their streams once sub streams have terminated. |
| fn terminate(&mut self) { |
| for (_, (_, recipient)) in self.live_iterators.lock().drain() { |
| recipient.close(); |
| } |
| } |
| } |
| |
| #[cfg(test)] |
| mod tests { |
| use { |
| super::*, |
| crate::{events::types::ComponentIdentifier, logs::stored_message::StoredMessage}, |
| diagnostics_hierarchy::trie::TrieIterableNode, |
| diagnostics_log_encoding::{ |
| encode::Encoder, Argument, Record, Severity as StreamSeverity, Value, |
| }, |
| fuchsia_zircon as zx, |
| selectors::{self, FastError}, |
| std::{io::Cursor, time::Duration}, |
| }; |
| |
| const TEST_URL: &'static str = "fuchsia-pkg://test"; |
| |
| #[fuchsia::test] |
| async fn inspect_repo_disallows_duplicated_dirs() { |
| let inspect_repo = DataRepo::default(); |
| let mut inspect_repo = inspect_repo.write(); |
| let moniker = vec!["a", "b", "foo.cmx"].into(); |
| let instance_id = "1234".to_string(); |
| |
| let component_id = ComponentIdentifier::Legacy { instance_id, moniker }; |
| let identity = ComponentIdentity::from_identifier_and_url(component_id, TEST_URL); |
| |
| let (proxy, _) = fidl::endpoints::create_proxy::<fio::DirectoryMarker>() |
| .expect("create directory proxy"); |
| |
| inspect_repo |
| .add_inspect_artifacts(identity.clone(), proxy, zx::Time::from_nanos(0)) |
| .expect("add to repo"); |
| |
| let (proxy, _) = fidl::endpoints::create_proxy::<fio::DirectoryMarker>() |
| .expect("create directory proxy"); |
| |
| inspect_repo |
| .add_inspect_artifacts(identity.clone(), proxy, zx::Time::from_nanos(0)) |
| .expect("add to repo"); |
| |
| assert_eq!( |
| inspect_repo |
| .data_directories |
| .get(&identity.unique_key().into()) |
| .unwrap() |
| .get_values() |
| .len(), |
| 1 |
| ); |
| } |
| |
| #[fuchsia::test] |
| async fn data_repo_updates_existing_entry_to_hold_inspect_data() { |
| let data_repo = DataRepo::default(); |
| let mut data_repo = data_repo.write(); |
| let moniker = vec!["a", "b", "foo.cmx"].into(); |
| let instance_id = "1234".to_string(); |
| |
| let component_id = ComponentIdentifier::Legacy { instance_id, moniker }; |
| let identity = ComponentIdentity::from_identifier_and_url(component_id, TEST_URL); |
| |
| data_repo |
| .add_new_component(identity.clone(), zx::Time::from_nanos(0)) |
| .expect("instantiated new component."); |
| |
| let (proxy, _) = fidl::endpoints::create_proxy::<fio::DirectoryMarker>() |
| .expect("create directory proxy"); |
| |
| data_repo |
| .add_inspect_artifacts(identity.clone(), proxy, zx::Time::from_nanos(0)) |
| .expect("add to repo"); |
| |
| assert_eq!( |
| data_repo |
| .data_directories |
| .get(&identity.unique_key().into()) |
| .unwrap() |
| .get_values() |
| .len(), |
| 1 |
| ); |
| let entry = |
| &data_repo.data_directories.get(&identity.unique_key().into()).unwrap().get_values()[0]; |
| assert!(entry.inspect.is_some()); |
| assert_eq!(entry.identity.url, TEST_URL); |
| } |
| |
| #[fuchsia::test] |
| async fn data_repo_tolerates_duplicate_new_component_insertions() { |
| let data_repo = DataRepo::default(); |
| let mut data_repo = data_repo.write(); |
| let moniker = vec!["a", "b", "foo.cmx"].into(); |
| let instance_id = "1234".to_string(); |
| |
| let component_id = ComponentIdentifier::Legacy { instance_id, moniker }; |
| let identity = ComponentIdentity::from_identifier_and_url(component_id.clone(), TEST_URL); |
| |
| data_repo |
| .add_new_component(identity.clone(), zx::Time::from_nanos(0)) |
| .expect("instantiated new component."); |
| |
| let duplicate_new_component_insertion = |
| data_repo.add_new_component(identity.clone(), zx::Time::from_nanos(1)); |
| |
| assert!(duplicate_new_component_insertion.is_ok()); |
| |
| let repo_values = |
| data_repo.data_directories.get(&identity.unique_key().into()).unwrap().get_values(); |
| assert_eq!(repo_values.len(), 1); |
| let entry = &repo_values[0]; |
| assert!(entry.lifecycle.is_some()); |
| assert_eq!(entry.identity.relative_moniker, component_id.relative_moniker_for_selectors()); |
| assert_eq!(entry.identity.url, TEST_URL); |
| } |
| |
| #[fuchsia::test] |
| async fn running_components_provide_start_time() { |
| let data_repo = DataRepo::default(); |
| let mut data_repo = data_repo.write(); |
| let moniker = vec!["a", "b", "foo.cmx"].into(); |
| let instance_id = "1234".to_string(); |
| |
| let component_id = ComponentIdentifier::Legacy { instance_id, moniker }; |
| let identity = ComponentIdentity::from_identifier_and_url(component_id.clone(), TEST_URL); |
| |
| let component_insertion = |
| data_repo.add_new_component(identity.clone(), zx::Time::from_nanos(1)); |
| |
| assert!(component_insertion.is_ok()); |
| |
| let repo_values = |
| data_repo.data_directories.get(&identity.unique_key().into()).unwrap().get_values(); |
| assert_eq!(repo_values.len(), 1); |
| let entry = &repo_values[0]; |
| assert!(entry.lifecycle.is_some()); |
| assert_eq!(entry.identity.relative_moniker, component_id.relative_moniker_for_selectors()); |
| assert_eq!(entry.identity.url, TEST_URL); |
| } |
| |
| #[fuchsia::test] |
| async fn data_repo_tolerant_of_new_component_calls_if_diagnostics_ready_already_processed() { |
| let data_repo = DataRepo::default(); |
| let mut data_repo = data_repo.write(); |
| let moniker = vec!["a", "b", "foo.cmx"].into(); |
| let instance_id = "1234".to_string(); |
| |
| let component_id = ComponentIdentifier::Legacy { instance_id, moniker }; |
| let identity = ComponentIdentity::from_identifier_and_url(component_id, TEST_URL); |
| |
| let (proxy, _) = fidl::endpoints::create_proxy::<fio::DirectoryMarker>() |
| .expect("create directory proxy"); |
| |
| data_repo |
| .add_inspect_artifacts(identity.clone(), proxy, zx::Time::from_nanos(0)) |
| .expect("add to repo"); |
| |
| let false_new_component_result = |
| data_repo.add_new_component(identity.clone(), zx::Time::from_nanos(0)); |
| assert!(false_new_component_result.is_ok()); |
| |
| // We shouldn't have overwritten the entry. There should still be an inspect |
| // artifacts container. |
| assert_eq!( |
| data_repo |
| .data_directories |
| .get(&identity.unique_key().into()) |
| .unwrap() |
| .get_values() |
| .len(), |
| 1 |
| ); |
| let entry = |
| &data_repo.data_directories.get(&identity.unique_key().into()).unwrap().get_values()[0]; |
| assert_eq!(entry.identity.url, TEST_URL); |
| assert!(entry.inspect.is_some()); |
| assert!(entry.lifecycle.is_some()); |
| } |
| |
| #[fuchsia::test] |
| async fn diagnostics_repo_cant_have_more_than_one_diagnostics_data_container_per_component() { |
| let data_repo = DataRepo::default(); |
| let mut data_repo = data_repo.write(); |
| let moniker = vec!["a", "b", "foo.cmx"].into(); |
| let instance_id = "1234".to_string(); |
| |
| let component_id = ComponentIdentifier::Legacy { instance_id, moniker }; |
| let identity = ComponentIdentity::from_identifier_and_url(component_id, TEST_URL); |
| |
| data_repo |
| .add_new_component(identity.clone(), zx::Time::from_nanos(0)) |
| .expect("insertion will succeed."); |
| |
| assert_eq!( |
| data_repo |
| .data_directories |
| .get(&identity.unique_key().into()) |
| .unwrap() |
| .get_values() |
| .len(), |
| 1 |
| ); |
| |
| let mutable_values = data_repo |
| .data_directories |
| .get_mut(&identity.unique_key().into()) |
| .unwrap() |
| .get_values_mut(); |
| |
| mutable_values |
| .push(ComponentDiagnostics::empty(Arc::new(identity.clone()), &Default::default())); |
| |
| let (proxy, _) = fidl::endpoints::create_proxy::<fio::DirectoryMarker>() |
| .expect("create directory proxy"); |
| |
| assert!(data_repo.add_inspect_artifacts(identity, proxy, zx::Time::from_nanos(0)).is_err()); |
| } |
| |
| #[fuchsia::test] |
| async fn data_repo_filters_inspect_by_selectors() { |
| let data_repo = DataRepo::default(); |
| let realm_path = vec!["a".to_string(), "b".to_string()]; |
| let instance_id = "1234".to_string(); |
| |
| let mut moniker = realm_path.clone(); |
| moniker.push("foo.cmx".to_string()); |
| let component_id = ComponentIdentifier::Legacy { instance_id, moniker: moniker.into() }; |
| let identity = ComponentIdentity::from_identifier_and_url(component_id, TEST_URL); |
| |
| data_repo |
| .write() |
| .add_new_component(identity.clone(), zx::Time::from_nanos(0)) |
| .expect("insertion will succeed."); |
| |
| data_repo |
| .write() |
| .add_inspect_artifacts( |
| identity, |
| io_util::open_directory_in_namespace("/tmp", io_util::OpenFlags::RIGHT_READABLE) |
| .expect("open root"), |
| zx::Time::from_nanos(0), |
| ) |
| .expect("add inspect artifacts"); |
| |
| let mut moniker = realm_path; |
| moniker.push("foo2.cmx".to_string()); |
| let component_id2 = ComponentIdentifier::Legacy { |
| instance_id: "12345".to_string(), |
| moniker: moniker.into(), |
| }; |
| let identity2 = ComponentIdentity::from_identifier_and_url(component_id2, TEST_URL); |
| |
| data_repo |
| .write() |
| .add_new_component(identity2.clone(), zx::Time::from_nanos(0)) |
| .expect("insertion will succeed."); |
| |
| data_repo |
| .write() |
| .add_inspect_artifacts( |
| identity2, |
| io_util::open_directory_in_namespace("/tmp", io_util::OpenFlags::RIGHT_READABLE) |
| .expect("open root"), |
| zx::Time::from_nanos(0), |
| ) |
| .expect("add inspect artifacts"); |
| |
| assert_eq!(2, data_repo.read().fetch_inspect_data(&None, None).len()); |
| |
| let selectors = Some(vec![ |
| selectors::parse_selector::<FastError>("a/b/foo.cmx:root").expect("parse selector") |
| ]); |
| assert_eq!(1, data_repo.read().fetch_inspect_data(&selectors, None).len()); |
| |
| let selectors = Some(vec![ |
| selectors::parse_selector::<FastError>("a/b/f*.cmx:root").expect("parse selector") |
| ]); |
| assert_eq!(2, data_repo.read().fetch_inspect_data(&selectors, None).len()); |
| |
| let selectors = Some(vec![ |
| selectors::parse_selector::<FastError>("foo.cmx:root").expect("parse selector") |
| ]); |
| assert_eq!(0, data_repo.read().fetch_inspect_data(&selectors, None).len()); |
| } |
| |
| #[fuchsia::test] |
| async fn data_repo_filters_logs_by_selectors() { |
| let repo = DataRepo::default(); |
| let foo_container = |
| repo.write().get_log_container(ComponentIdentity::from_identifier_and_url( |
| ComponentIdentifier::parse_from_moniker("./foo").unwrap(), |
| "fuchsia-pkg://foo", |
| )); |
| let bar_container = |
| repo.write().get_log_container(ComponentIdentity::from_identifier_and_url( |
| ComponentIdentifier::parse_from_moniker("./bar").unwrap(), |
| "fuchsia-pkg://bar", |
| )); |
| |
| foo_container.ingest_message(make_message("a", 1)); |
| bar_container.ingest_message(make_message("b", 2)); |
| foo_container.ingest_message(make_message("c", 3)); |
| |
| let stream = repo.logs_cursor(StreamMode::Snapshot, None); |
| |
| let results = |
| stream.map(|value| value.msg().unwrap().to_string()).collect::<Vec<_>>().await; |
| assert_eq!(results, vec!["a".to_string(), "b".to_string(), "c".to_string()]); |
| |
| let filtered_stream = repo.logs_cursor( |
| StreamMode::Snapshot, |
| Some(vec![selectors::parse_selector::<FastError>("foo:root").unwrap()]), |
| ); |
| |
| let results = |
| filtered_stream.map(|value| value.msg().unwrap().to_string()).collect::<Vec<_>>().await; |
| assert_eq!(results, vec!["a".to_string(), "c".to_string()]); |
| } |
| |
| #[fuchsia::test] |
| async fn multiplexer_broker_cleanup() { |
| let repo = DataRepo::default(); |
| let stream = repo.logs_cursor(StreamMode::SnapshotThenSubscribe, None); |
| |
| assert_eq!(repo.read().logs_multiplexers.live_iterators.lock().len(), 1); |
| |
| // When the multiplexer goes away it must be forgotten by the broker. |
| drop(stream); |
| loop { |
| fasync::Timer::new(Duration::from_millis(100)).await; |
| if repo.read().logs_multiplexers.live_iterators.lock().len() == 0 { |
| break; |
| } |
| } |
| } |
| |
| fn make_message(msg: &str, timestamp: i64) -> StoredMessage { |
| let record = Record { |
| timestamp, |
| severity: StreamSeverity::Debug, |
| arguments: vec![ |
| Argument { name: "pid".to_string(), value: Value::UnsignedInt(1) }, |
| Argument { name: "tid".to_string(), value: Value::UnsignedInt(2) }, |
| Argument { name: "message".to_string(), value: Value::Text(msg.to_string()) }, |
| ], |
| }; |
| let mut buffer = Cursor::new(vec![0u8; 1024]); |
| let mut encoder = Encoder::new(&mut buffer); |
| encoder.write_record(&record).unwrap(); |
| let encoded = &buffer.get_ref()[..buffer.position() as usize]; |
| StoredMessage::structured(encoded, Default::default()).unwrap() |
| } |
| } |
