src/diagnostics/archivist/src/events/types.rs - fuchsia - Git at Google

 // Copyright 2020 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::ComponentIdentity,
         events::error::{EventError, MonikerError},
     },
     async_trait::async_trait,
     fidl::endpoints::{ServerEnd, ServiceMarker},
     fidl_fuchsia_inspect::TreeProxy,
     fidl_fuchsia_inspect_deprecated::InspectProxy,
     fidl_fuchsia_io::DirectoryProxy,
     fidl_fuchsia_logger::{LogSinkMarker, LogSinkRequestStream},
     fidl_fuchsia_sys2::{self as fsys, Event, EventHeader},
     fidl_fuchsia_sys_internal::SourceIdentity,
     fidl_table_validation::*,
     fuchsia_zircon as zx,
     futures::{channel::mpsc, stream::BoxStream},
     io_util,
     std::{convert::TryFrom, ops::Deref},
 };

 #[async_trait]
 pub trait EventSource: Sync + Send {
     async fn listen(&mut self, sender: mpsc::Sender<ComponentEvent>) -> Result<(), EventError>;
 }

 /// The capacity for bounded channels used by this implementation.
 pub static CHANNEL_CAPACITY: usize = 1024;

 #[derive(Clone, Debug, Eq, Hash, PartialEq)]
 pub struct Moniker(Vec<String>);

 impl Deref for Moniker {
     type Target = Vec<String>;
     fn deref(&self) -> &Self::Target {
         &self.0
     }
 }

 impl Into<Moniker> for Vec<&str> {
     fn into(self) -> Moniker {
         Moniker(self.into_iter().map(|s| s.to_string()).collect())
     }
 }

 impl Into<Moniker> for Vec<String> {
     fn into(self) -> Moniker {
         Moniker(self)
     }
 }

 #[derive(Clone, Debug, Eq, PartialEq)]
 pub struct UniqueKey(Vec<String>);

 impl Deref for UniqueKey {
     type Target = Vec<String>;
     fn deref(&self) -> &Self::Target {
         &self.0
     }
 }

 impl Into<UniqueKey> for Vec<&str> {
     fn into(self) -> UniqueKey {
         UniqueKey(self.into_iter().map(|s| s.to_string()).collect())
     }
 }

 /// Represents the ID of a component.
 #[derive(Debug, Clone, Eq, Hash, PartialEq)]
 pub enum ComponentIdentifier {
     Legacy {
         /// The realm path plus the name of the component.
         moniker: Moniker,

         /// The instance ID of the component.
         instance_id: String,
     },
     Moniker(Vec<MonikerSegment>),
 }

 impl ComponentIdentifier {
     /// Returns the relative moniker to be used for selectors.
     /// For legacy components (v1), this is the relative moniker with respect to the root realm.
     pub fn relative_moniker_for_selectors(&self) -> Moniker {
         match self {
             Self::Legacy { moniker, .. } => moniker.clone(),
             Self::Moniker(segments) => {
                 if segments.is_empty() {
                     Moniker(vec![])
                 } else {
                     Moniker(segments.iter().map(|s| s.to_string()).collect())
                 }
             }
         }
     }

     pub fn unique_key(&self) -> UniqueKey {
         match self {
             Self::Legacy { instance_id, .. } => {
                 let mut key = self.relative_moniker_for_selectors().0;
                 key.push(instance_id.clone());
                 UniqueKey(key)
             }
             Self::Moniker(segments) => {
                 let mut key = vec![];
                 for segment in segments {
                     key.push(segment.to_string());
                     key.push(segment.instance_id.clone());
                 }
                 UniqueKey(key)
             }
         }
     }

     pub fn parse_from_moniker(moniker: &str) -> Result<Self, MonikerError> {
         if moniker == "<component_manager>" {
             return Ok(ComponentIdentifier::Moniker(vec![MonikerSegment {
                 collection: None,
                 name: "<component_manager>".to_string(),
                 instance_id: "0".to_string(),
             }]));
         }

         if moniker == "." {
             return Ok(ComponentIdentifier::Moniker(vec![]));
         }

         let without_root = moniker
             .strip_prefix("./")
             .ok_or_else(|| MonikerError::InvalidMonikerPrefix(moniker.to_string()))?;

         let mut segments = vec![];
         for raw_segment in without_root.split("/") {
             let mut parts = raw_segment.split(":");
             let segment = match (parts.next(), parts.next(), parts.next()) {
                 // we have a collection, a component name, and an instance id
                 (Some(c), Some(n), Some(i)) => MonikerSegment {
                     collection: Some(c.to_string()),
                     name: n.to_string(),
                     instance_id: i.to_string(),
                 },
                 // we have a name and an instance id, no collection
                 (Some(n), Some(i), None) => MonikerSegment {
                     collection: None,
                     name: n.to_string(),
                     instance_id: i.to_string(),
                 },
                 _ => return Err(MonikerError::InvalidSegment(raw_segment.to_string())),
             };
             segments.push(segment);
         }

         Ok(ComponentIdentifier::Moniker(segments))
     }
 }

 impl std::fmt::Display for ComponentIdentifier {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
             Self::Legacy { moniker, instance_id } => {
                 for (i, segment) in moniker.iter().enumerate() {
                     if i > 0 {
                         write!(f, "/")?;
                     }
                     write!(f, "{}", segment)?;
                 }
                 write!(f, ":{}", instance_id)
             }
             Self::Moniker(segments) => {
                 if segments.is_empty() {
                     write!(f, ".")
                 } else {
                     for (i, segment) in segments.iter().enumerate() {
                         if i > 0 {
                             write!(f, "/")?;
                         }
                         write!(f, "{}", segment)?;
                     }
                     Ok(())
                 }
             }
         }
     }
 }

 /// A single segment in the moniker of a component.
 #[derive(Debug, Clone, Eq, Hash, PartialEq)]
 pub struct MonikerSegment {
     /// The name of the component's collection, if any.
     pub collection: Option<String>,
     /// The name of the component.
     pub name: String,
     /// The instance of the component.
     pub instance_id: String,
 }

 impl std::fmt::Display for MonikerSegment {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         if let Some(collection) = &self.collection {
             write!(f, "{}:", collection)?;
         }
         write!(f, "{}", self.name)
     }
 }

 #[derive(Debug, Clone, ValidFidlTable, PartialEq)]
 #[fidl_table_src(SourceIdentity)]
 pub struct ValidatedSourceIdentity {
     pub realm_path: Vec<String>,
     pub component_url: String,
     pub component_name: String,
     pub instance_id: String,
 }

 #[derive(Debug, ValidFidlTable)]
 #[fidl_table_src(EventHeader)]
 pub struct ValidatedEventHeader {
     pub event_type: fsys::EventType,
     pub component_url: String,
     pub moniker: String,
     pub timestamp: i64,
 }

 #[derive(Debug, ValidFidlTable)]
 #[fidl_table_src(Event)]
 pub struct ValidatedEvent {
     /// Information about the component for which this event was generated.
     pub header: ValidatedEventHeader,

     /// Optional payload for some event types
     #[fidl_field_type(optional)]
     pub event_result: Option<fsys::EventResult>,
 }

 /// The ID of a component as used in components V1.
 /// Represents the shared data associated with
 /// all component events.
 #[derive(Debug, PartialEq, Clone)]
 pub struct EventMetadata {
     pub identity: ComponentIdentity,

     pub timestamp: zx::Time,
 }

 impl EventMetadata {
     pub fn new(identity: ComponentIdentity) -> Self {
         Self { identity, timestamp: zx::Time::get_monotonic() }
     }
 }

 impl TryFrom<SourceIdentity> for EventMetadata {
     type Error = EventError;
     fn try_from(component: SourceIdentity) -> Result<Self, Self::Error> {
         Ok(EventMetadata {
             identity: ComponentIdentity::try_from(component)?,
             timestamp: zx::Time::get_monotonic(),
         })
     }
 }

 /// Represents the diagnostics data associated
 /// with a component being observed starting.
 #[derive(Debug, PartialEq)]
 pub struct StartEvent {
     pub metadata: EventMetadata,
 }

 /// Represents the diagnostics data associated
 /// with a component being observed running.
 #[derive(Debug, PartialEq)]
 pub struct RunningEvent {
     pub metadata: EventMetadata,

     pub component_start_time: zx::Time,
 }

 /// Represents the diagnostics data associated
 /// with a component being observed stopping.
 #[derive(Debug, PartialEq)]
 pub struct StopEvent {
     pub metadata: EventMetadata,
 }

 /// Represents the diagnostics data associated
 /// with a new Diagnostics Directory being
 /// made available.
 #[derive(Debug)]
 pub struct DiagnosticsReadyEvent {
     pub metadata: EventMetadata,

     /// Proxy to the inspect data host.
     pub directory: Option<DirectoryProxy>,
 }

 /// A new incoming connection to `LogSink`.
 pub struct LogSinkRequestedEvent {
     pub metadata: EventMetadata,
     pub requests: LogSinkRequestStream,
 }

 impl LogSinkRequestedEvent {
     fn new(event: ValidatedEvent, metadata: EventMetadata) -> Result<Self, EventError> {
         let payload = event.event_result.ok_or(EventError::MissingField("event_result")).and_then(
             |result| match result {
                 fsys::EventResult::Payload(fsys::EventPayload::CapabilityRequested(payload)) => {
                     Ok(payload)
                 }
                 fsys::EventResult::Error(fsys::EventError {
                     description: Some(description),
                     ..
                 }) => Err(EventError::ReceivedError { description }),
                 result => Err(EventError::UnrecognizedResult { result }),
             },
         )?;

         let capability_name = payload.name.ok_or(EventError::MissingField("name"))?;
         if &capability_name != LogSinkMarker::NAME {
             Err(EventError::IncorrectName {
                 received: capability_name,
                 expected: LogSinkMarker::NAME,
             })?;
         }

         let capability = payload.capability.ok_or(EventError::MissingField("capability"))?;
         let requests = ServerEnd::<LogSinkMarker>::new(capability)
             .into_stream()
             .map_err(|source| EventError::InvalidServerEnd { source })?;

         Ok(Self { metadata, requests })
     }
 }

 impl std::fmt::Debug for LogSinkRequestedEvent {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("LogSinkRequestedEvent").field("metadata", &self.metadata).finish()
     }
 }

 pub type ComponentEventChannel = mpsc::Sender<ComponentEvent>;

 /// A stream of |ComponentEvent|s
 pub type ComponentEventStream = BoxStream<'static, ComponentEvent>;

 /// An event that occurred to a component.
 #[derive(Debug)]
 pub enum ComponentEvent {
     /// We observed the component starting.
     Start(StartEvent),

     /// We observed the component already started.
     Running(RunningEvent),

     /// We observed the component stopping.
     Stop(StopEvent),

     /// We observed the creation of a new `out` directory.
     DiagnosticsReady(DiagnosticsReadyEvent),

     /// We received a new connection to `LogSink`.
     LogSinkRequested(LogSinkRequestedEvent),
 }

 /// Data associated with a component.
 /// This data is stored by data collectors and passed by the collectors to processors.
 #[derive(Debug)]
 pub enum InspectData {
     /// Empty data, for testing.
     Empty,

     /// A VMO containing data associated with the event.
     Vmo(zx::Vmo),

     /// A file containing data associated with the event.
     ///
     /// Because we can't synchronously retrieve file contents like we can for VMOs, this holds
     /// the full file contents. Future changes should make streaming ingestion feasible.
     File(Vec<u8>),

     /// A connection to a Tree service and a handle to the root hierarchy VMO. This VMO is what a
     /// root.inspect file would contain and the result of calling Tree#GetContent. We hold to it
     /// so that we can use it when the component is removed, at which point calling the Tree
     /// service is not an option.
     Tree(TreeProxy, Option<zx::Vmo>),

     /// A connection to the deprecated Inspect service.
     DeprecatedFidl(InspectProxy),
 }

 impl TryFrom<Event> for ComponentEvent {
     type Error = EventError;

     fn try_from(event: Event) -> Result<ComponentEvent, Self::Error> {
         let event: ValidatedEvent = ValidatedEvent::try_from(event)?;

         let metadata = EventMetadata {
             identity: ComponentIdentity::from_identifier_and_url(
                 &ComponentIdentifier::parse_from_moniker(&event.header.moniker)?,
                 &event.header.component_url,
             ),
             timestamp: zx::Time::from_nanos(event.header.timestamp),
         };

         match event.header.event_type {
             fsys::EventType::Started => {
                 let start_event = StartEvent { metadata };
                 Ok(ComponentEvent::Start(start_event))
             }
             fsys::EventType::Stopped => {
                 let stop_event = StopEvent { metadata };
                 Ok(ComponentEvent::Stop(stop_event))
             }
             fsys::EventType::CapabilityReady | fsys::EventType::Running => {
                 construct_payload_holding_component_event(event, metadata)
             }
             fsys::EventType::CapabilityRequested => {
                 Ok(ComponentEvent::LogSinkRequested(LogSinkRequestedEvent::new(event, metadata)?))
             }
             _ => Err(EventError::InvalidEventType { ty: event.header.event_type }),
         }
     }
 }

 fn construct_payload_holding_component_event(
     event: ValidatedEvent,
     shared_data: EventMetadata,
 ) -> Result<ComponentEvent, EventError> {
     match event.event_result {
         Some(result) => {
             match result {
                 fsys::EventResult::Payload(fsys::EventPayload::CapabilityReady(
                     capability_ready,
                 )) => {
                     let name = capability_ready.name.ok_or(EventError::MissingField("name"))?;
                     if name == "diagnostics" {
                         match capability_ready.node {
                             Some(node) => {
                                 let diagnostics_ready_event = DiagnosticsReadyEvent {
                                     metadata: shared_data,
                                     directory: io_util::node_to_directory(node.into_proxy()?).ok(),
                                 };
                                 Ok(ComponentEvent::DiagnosticsReady(diagnostics_ready_event))
                             }
                             None => Err(EventError::MissingDiagnosticsDir),
                         }
                     } else {
                         Err(EventError::IncorrectName { received: name, expected: "diagnostics" })
                     }
                 }
                 fsys::EventResult::Payload(fsys::EventPayload::Running(payload)) => {
                     match payload.started_timestamp {
                         Some(timestamp) => {
                             let existing_data = RunningEvent {
                                 metadata: shared_data,
                                 component_start_time: zx::Time::from_nanos(timestamp),
                             };
                             Ok(ComponentEvent::Running(existing_data))
                         }
                         None => Err(EventError::MissingStartTimestamp),
                     }
                 }
                 fsys::EventResult::Error(fsys::EventError {
                     description: Some(description),
                     ..
                 }) => {
                     // TODO(fxbug.dev/53903): result.error carries information about errors that happened
                     // in component_manager. We should dump those in diagnostics.
                     Err(EventError::ReceivedError { description })
                 }
                 result => Err(EventError::UnrecognizedResult { result }),
             }
         }
         None => Err(EventError::MissingPayload { event }),
     }
 }

 impl PartialEq for ComponentEvent {
     fn eq(&self, other: &Self) -> bool {
         match (self, other) {
             (ComponentEvent::Start(a), ComponentEvent::Start(b)) => {
                 return a == b;
             }
             (ComponentEvent::Stop(a), ComponentEvent::Stop(b)) => {
                 return a == b;
             }
             (ComponentEvent::DiagnosticsReady(a), ComponentEvent::DiagnosticsReady(b)) => {
                 return a == b;
             }
             (ComponentEvent::Running(a), ComponentEvent::Running(b)) => {
                 return a == b;
             }
             // we can't check two LogSinkRequested events for equality because they have channels
             _ => false,
         }
     }
 }

 // Requires a custom partial_eq due to the presence of a directory proxy.
 // Two events with the same metadata and different directory proxies
 // will be considered the same.
 impl PartialEq for DiagnosticsReadyEvent {
     fn eq(&self, other: &Self) -> bool {
         self.metadata == other.metadata
     }
 }

 #[cfg(test)]
 mod tests {
     use super::*;
     use crate::logs::testing::create_log_sink_requested_event;
     use std::convert::TryInto;

     #[test]
     fn convert_v2_moniker_for_diagnostics() {
         let identifier = ComponentIdentifier::parse_from_moniker("./a:0").unwrap();
         assert_eq!(identifier.relative_moniker_for_selectors(), vec!["a"].into());
         assert_eq!(identifier.unique_key(), vec!["a", "0"].into());

         let identifier = ComponentIdentifier::parse_from_moniker("./a:0/b:1").unwrap();
         assert_eq!(identifier.relative_moniker_for_selectors(), vec!["a", "b"].into());
         assert_eq!(identifier.unique_key(), vec!["a", "0", "b", "1"].into());

         let identifier = ComponentIdentifier::parse_from_moniker("./a:0/coll:comp:1/b:0").unwrap();
         assert_eq!(identifier.relative_moniker_for_selectors(), vec!["a", "coll:comp", "b"].into());
         assert_eq!(identifier.unique_key(), vec!["a", "0", "coll:comp", "1", "b", "0"].into());

         let identifier = ComponentIdentifier::parse_from_moniker(".").unwrap();
         assert!(identifier.relative_moniker_for_selectors().is_empty());
         assert!(identifier.unique_key().is_empty());
     }

     #[fuchsia_async::run_singlethreaded(test)] // we need an executor for the fidl types
     async fn validate_logsink_requested_event() {
         let target_moniker = "./foo:0";
         let target_url = "http://foo.com".to_string();
         let (_log_sink_proxy, log_sink_server_end) =
             fidl::endpoints::create_proxy::<LogSinkMarker>().unwrap();
         let raw_event = create_log_sink_requested_event(
             target_moniker.to_owned(),
             target_url.clone(),
             log_sink_server_end.into_channel(),
         );
         let event = match raw_event.try_into().unwrap() {
             ComponentEvent::LogSinkRequested(e) => e,
             other => panic!("incorrect event type received: {:?}", other),
         };

         assert_eq!(
             event.metadata.identity,
             ComponentIdentity::from_identifier_and_url(
                 &ComponentIdentifier::parse_from_moniker(target_moniker).unwrap(),
                 target_url
             )
         );
     }
 }
	// Copyright 2020 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::ComponentIdentity,
	events::error::{EventError, MonikerError},
	},
	async_trait::async_trait,
	fidl::endpoints::{ServerEnd, ServiceMarker},
	fidl_fuchsia_inspect::TreeProxy,
	fidl_fuchsia_inspect_deprecated::InspectProxy,
	fidl_fuchsia_io::DirectoryProxy,
	fidl_fuchsia_logger::{LogSinkMarker, LogSinkRequestStream},
	fidl_fuchsia_sys2::{self as fsys, Event, EventHeader},
	fidl_fuchsia_sys_internal::SourceIdentity,
	fidl_table_validation::*,
	fuchsia_zircon as zx,
	futures::{channel::mpsc, stream::BoxStream},
	io_util,
	std::{convert::TryFrom, ops::Deref},
	};

	#[async_trait]
	pub trait EventSource: Sync + Send {
	async fn listen(&mut self, sender: mpsc::Sender<ComponentEvent>) -> Result<(), EventError>;
	}

	/// The capacity for bounded channels used by this implementation.
	pub static CHANNEL_CAPACITY: usize = 1024;

	#[derive(Clone, Debug, Eq, Hash, PartialEq)]
	pub struct Moniker(Vec<String>);

	impl Deref for Moniker {
	type Target = Vec<String>;
	fn deref(&self) -> &Self::Target {
	&self.0
	}
	}

	impl Into<Moniker> for Vec<&str> {
	fn into(self) -> Moniker {
	Moniker(self.into_iter().map(\|s\| s.to_string()).collect())
	}
	}

	impl Into<Moniker> for Vec<String> {
	fn into(self) -> Moniker {
	Moniker(self)
	}
	}

	#[derive(Clone, Debug, Eq, PartialEq)]
	pub struct UniqueKey(Vec<String>);

	impl Deref for UniqueKey {
	type Target = Vec<String>;
	fn deref(&self) -> &Self::Target {
	&self.0
	}
	}

	impl Into<UniqueKey> for Vec<&str> {
	fn into(self) -> UniqueKey {
	UniqueKey(self.into_iter().map(\|s\| s.to_string()).collect())
	}
	}

	/// Represents the ID of a component.
	#[derive(Debug, Clone, Eq, Hash, PartialEq)]
	pub enum ComponentIdentifier {
	Legacy {
	/// The realm path plus the name of the component.
	moniker: Moniker,

	/// The instance ID of the component.
	instance_id: String,
	},
	Moniker(Vec<MonikerSegment>),
	}

	impl ComponentIdentifier {
	/// Returns the relative moniker to be used for selectors.
	/// For legacy components (v1), this is the relative moniker with respect to the root realm.
	pub fn relative_moniker_for_selectors(&self) -> Moniker {
	match self {
	Self::Legacy { moniker, .. } => moniker.clone(),
	Self::Moniker(segments) => {
	if segments.is_empty() {
	Moniker(vec![])
	} else {
	Moniker(segments.iter().map(\|s\| s.to_string()).collect())
	}
	}
	}
	}

	pub fn unique_key(&self) -> UniqueKey {
	match self {
	Self::Legacy { instance_id, .. } => {
	let mut key = self.relative_moniker_for_selectors().0;
	key.push(instance_id.clone());
	UniqueKey(key)
	}
	Self::Moniker(segments) => {
	let mut key = vec![];
	for segment in segments {
	key.push(segment.to_string());
	key.push(segment.instance_id.clone());
	}
	UniqueKey(key)
	}
	}
	}

	pub fn parse_from_moniker(moniker: &str) -> Result<Self, MonikerError> {
	if moniker == "<component_manager>" {
	return Ok(ComponentIdentifier::Moniker(vec![MonikerSegment {
	collection: None,
	name: "<component_manager>".to_string(),
	instance_id: "0".to_string(),
	}]));
	}

	if moniker == "." {
	return Ok(ComponentIdentifier::Moniker(vec![]));
	}

	let without_root = moniker
	.strip_prefix("./")
	.ok_or_else(\|\| MonikerError::InvalidMonikerPrefix(moniker.to_string()))?;

	let mut segments = vec![];
	for raw_segment in without_root.split("/") {
	let mut parts = raw_segment.split(":");
	let segment = match (parts.next(), parts.next(), parts.next()) {
	// we have a collection, a component name, and an instance id
	(Some(c), Some(n), Some(i)) => MonikerSegment {
	collection: Some(c.to_string()),
	name: n.to_string(),
	instance_id: i.to_string(),
	},
	// we have a name and an instance id, no collection
	(Some(n), Some(i), None) => MonikerSegment {
	collection: None,
	name: n.to_string(),
	instance_id: i.to_string(),
	},
	_ => return Err(MonikerError::InvalidSegment(raw_segment.to_string())),
	};
	segments.push(segment);
	}

	Ok(ComponentIdentifier::Moniker(segments))
	}
	}

	impl std::fmt::Display for ComponentIdentifier {
	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
	match self {
	Self::Legacy { moniker, instance_id } => {
	for (i, segment) in moniker.iter().enumerate() {
	if i > 0 {
	write!(f, "/")?;
	}
	write!(f, "{}", segment)?;
	}
	write!(f, ":{}", instance_id)
	}
	Self::Moniker(segments) => {
	if segments.is_empty() {
	write!(f, ".")
	} else {
	for (i, segment) in segments.iter().enumerate() {
	if i > 0 {
	write!(f, "/")?;
	}
	write!(f, "{}", segment)?;
	}
	Ok(())
	}
	}
	}
	}
	}

	/// A single segment in the moniker of a component.
	#[derive(Debug, Clone, Eq, Hash, PartialEq)]
	pub struct MonikerSegment {
	/// The name of the component's collection, if any.
	pub collection: Option<String>,
	/// The name of the component.
	pub name: String,
	/// The instance of the component.
	pub instance_id: String,
	}

	impl std::fmt::Display for MonikerSegment {
	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
	if let Some(collection) = &self.collection {
	write!(f, "{}:", collection)?;
	}
	write!(f, "{}", self.name)
	}
	}

	#[derive(Debug, Clone, ValidFidlTable, PartialEq)]
	#[fidl_table_src(SourceIdentity)]
	pub struct ValidatedSourceIdentity {
	pub realm_path: Vec<String>,
	pub component_url: String,
	pub component_name: String,
	pub instance_id: String,
	}

	#[derive(Debug, ValidFidlTable)]
	#[fidl_table_src(EventHeader)]
	pub struct ValidatedEventHeader {
	pub event_type: fsys::EventType,
	pub component_url: String,
	pub moniker: String,
	pub timestamp: i64,
	}

	#[derive(Debug, ValidFidlTable)]
	#[fidl_table_src(Event)]
	pub struct ValidatedEvent {
	/// Information about the component for which this event was generated.
	pub header: ValidatedEventHeader,

	/// Optional payload for some event types
	#[fidl_field_type(optional)]
	pub event_result: Option<fsys::EventResult>,
	}

	/// The ID of a component as used in components V1.
	/// Represents the shared data associated with
	/// all component events.
	#[derive(Debug, PartialEq, Clone)]
	pub struct EventMetadata {
	pub identity: ComponentIdentity,

	pub timestamp: zx::Time,
	}

	impl EventMetadata {
	pub fn new(identity: ComponentIdentity) -> Self {
	Self { identity, timestamp: zx::Time::get_monotonic() }
	}
	}

	impl TryFrom<SourceIdentity> for EventMetadata {
	type Error = EventError;
	fn try_from(component: SourceIdentity) -> Result<Self, Self::Error> {
	Ok(EventMetadata {
	identity: ComponentIdentity::try_from(component)?,
	timestamp: zx::Time::get_monotonic(),
	})
	}
	}

	/// Represents the diagnostics data associated
	/// with a component being observed starting.
	#[derive(Debug, PartialEq)]
	pub struct StartEvent {
	pub metadata: EventMetadata,
	}

	/// Represents the diagnostics data associated
	/// with a component being observed running.
	#[derive(Debug, PartialEq)]
	pub struct RunningEvent {
	pub metadata: EventMetadata,

	pub component_start_time: zx::Time,
	}

	/// Represents the diagnostics data associated
	/// with a component being observed stopping.
	#[derive(Debug, PartialEq)]
	pub struct StopEvent {
	pub metadata: EventMetadata,
	}

	/// Represents the diagnostics data associated
	/// with a new Diagnostics Directory being
	/// made available.
	#[derive(Debug)]
	pub struct DiagnosticsReadyEvent {
	pub metadata: EventMetadata,

	/// Proxy to the inspect data host.
	pub directory: Option<DirectoryProxy>,
	}

	/// A new incoming connection to `LogSink`.
	pub struct LogSinkRequestedEvent {
	pub metadata: EventMetadata,
	pub requests: LogSinkRequestStream,
	}

	impl LogSinkRequestedEvent {
	fn new(event: ValidatedEvent, metadata: EventMetadata) -> Result<Self, EventError> {
	let payload = event.event_result.ok_or(EventError::MissingField("event_result")).and_then(
	\|result\| match result {
	fsys::EventResult::Payload(fsys::EventPayload::CapabilityRequested(payload)) => {
	Ok(payload)
	}
	fsys::EventResult::Error(fsys::EventError {
	description: Some(description),
	..
	}) => Err(EventError::ReceivedError { description }),
	result => Err(EventError::UnrecognizedResult { result }),
	},
	)?;

	let capability_name = payload.name.ok_or(EventError::MissingField("name"))?;
	if &capability_name != LogSinkMarker::NAME {
	Err(EventError::IncorrectName {
	received: capability_name,
	expected: LogSinkMarker::NAME,
	})?;
	}

	let capability = payload.capability.ok_or(EventError::MissingField("capability"))?;
	let requests = ServerEnd::<LogSinkMarker>::new(capability)
	.into_stream()
	.map_err(\|source\| EventError::InvalidServerEnd { source })?;

	Ok(Self { metadata, requests })
	}
	}

	impl std::fmt::Debug for LogSinkRequestedEvent {
	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
	f.debug_struct("LogSinkRequestedEvent").field("metadata", &self.metadata).finish()
	}
	}

	pub type ComponentEventChannel = mpsc::Sender<ComponentEvent>;

	/// A stream of \|ComponentEvent\|s
	pub type ComponentEventStream = BoxStream<'static, ComponentEvent>;

	/// An event that occurred to a component.
	#[derive(Debug)]
	pub enum ComponentEvent {
	/// We observed the component starting.
	Start(StartEvent),

	/// We observed the component already started.
	Running(RunningEvent),

	/// We observed the component stopping.
	Stop(StopEvent),

	/// We observed the creation of a new `out` directory.
	DiagnosticsReady(DiagnosticsReadyEvent),

	/// We received a new connection to `LogSink`.
	LogSinkRequested(LogSinkRequestedEvent),
	}

	/// Data associated with a component.
	/// This data is stored by data collectors and passed by the collectors to processors.
	#[derive(Debug)]
	pub enum InspectData {
	/// Empty data, for testing.
	Empty,

	/// A VMO containing data associated with the event.
	Vmo(zx::Vmo),

	/// A file containing data associated with the event.
	///
	/// Because we can't synchronously retrieve file contents like we can for VMOs, this holds
	/// the full file contents. Future changes should make streaming ingestion feasible.
	File(Vec<u8>),

	/// A connection to a Tree service and a handle to the root hierarchy VMO. This VMO is what a
	/// root.inspect file would contain and the result of calling Tree#GetContent. We hold to it
	/// so that we can use it when the component is removed, at which point calling the Tree
	/// service is not an option.
	Tree(TreeProxy, Option<zx::Vmo>),

	/// A connection to the deprecated Inspect service.
	DeprecatedFidl(InspectProxy),
	}

	impl TryFrom<Event> for ComponentEvent {
	type Error = EventError;

	fn try_from(event: Event) -> Result<ComponentEvent, Self::Error> {
	let event: ValidatedEvent = ValidatedEvent::try_from(event)?;

	let metadata = EventMetadata {
	identity: ComponentIdentity::from_identifier_and_url(
	&ComponentIdentifier::parse_from_moniker(&event.header.moniker)?,
	&event.header.component_url,
	),
	timestamp: zx::Time::from_nanos(event.header.timestamp),
	};

	match event.header.event_type {
	fsys::EventType::Started => {
	let start_event = StartEvent { metadata };
	Ok(ComponentEvent::Start(start_event))
	}
	fsys::EventType::Stopped => {
	let stop_event = StopEvent { metadata };
	Ok(ComponentEvent::Stop(stop_event))
	}
	fsys::EventType::CapabilityReady \| fsys::EventType::Running => {
	construct_payload_holding_component_event(event, metadata)
	}
	fsys::EventType::CapabilityRequested => {
	Ok(ComponentEvent::LogSinkRequested(LogSinkRequestedEvent::new(event, metadata)?))
	}
	_ => Err(EventError::InvalidEventType { ty: event.header.event_type }),
	}
	}
	}

	fn construct_payload_holding_component_event(
	event: ValidatedEvent,
	shared_data: EventMetadata,
	) -> Result<ComponentEvent, EventError> {
	match event.event_result {
	Some(result) => {
	match result {
	fsys::EventResult::Payload(fsys::EventPayload::CapabilityReady(
	capability_ready,
	)) => {
	let name = capability_ready.name.ok_or(EventError::MissingField("name"))?;
	if name == "diagnostics" {
	match capability_ready.node {
	Some(node) => {
	let diagnostics_ready_event = DiagnosticsReadyEvent {
	metadata: shared_data,
	directory: io_util::node_to_directory(node.into_proxy()?).ok(),
	};
	Ok(ComponentEvent::DiagnosticsReady(diagnostics_ready_event))
	}
	None => Err(EventError::MissingDiagnosticsDir),
	}
	} else {
	Err(EventError::IncorrectName { received: name, expected: "diagnostics" })
	}
	}
	fsys::EventResult::Payload(fsys::EventPayload::Running(payload)) => {
	match payload.started_timestamp {
	Some(timestamp) => {
	let existing_data = RunningEvent {
	metadata: shared_data,
	component_start_time: zx::Time::from_nanos(timestamp),
	};
	Ok(ComponentEvent::Running(existing_data))
	}
	None => Err(EventError::MissingStartTimestamp),
	}
	}
	fsys::EventResult::Error(fsys::EventError {
	description: Some(description),
	..
	}) => {
	// TODO(fxbug.dev/53903): result.error carries information about errors that happened
	// in component_manager. We should dump those in diagnostics.
	Err(EventError::ReceivedError { description })
	}
	result => Err(EventError::UnrecognizedResult { result }),
	}
	}
	None => Err(EventError::MissingPayload { event }),
	}
	}

	impl PartialEq for ComponentEvent {
	fn eq(&self, other: &Self) -> bool {
	match (self, other) {
	(ComponentEvent::Start(a), ComponentEvent::Start(b)) => {
	return a == b;
	}
	(ComponentEvent::Stop(a), ComponentEvent::Stop(b)) => {
	return a == b;
	}
	(ComponentEvent::DiagnosticsReady(a), ComponentEvent::DiagnosticsReady(b)) => {
	return a == b;
	}
	(ComponentEvent::Running(a), ComponentEvent::Running(b)) => {
	return a == b;
	}
	// we can't check two LogSinkRequested events for equality because they have channels
	_ => false,
	}
	}
	}

	// Requires a custom partial_eq due to the presence of a directory proxy.
	// Two events with the same metadata and different directory proxies
	// will be considered the same.
	impl PartialEq for DiagnosticsReadyEvent {
	fn eq(&self, other: &Self) -> bool {
	self.metadata == other.metadata
	}
	}

	#[cfg(test)]
	mod tests {
	use super::*;
	use crate::logs::testing::create_log_sink_requested_event;
	use std::convert::TryInto;

	#[test]
	fn convert_v2_moniker_for_diagnostics() {
	let identifier = ComponentIdentifier::parse_from_moniker("./a:0").unwrap();
	assert_eq!(identifier.relative_moniker_for_selectors(), vec!["a"].into());
	assert_eq!(identifier.unique_key(), vec!["a", "0"].into());

	let identifier = ComponentIdentifier::parse_from_moniker("./a:0/b:1").unwrap();
	assert_eq!(identifier.relative_moniker_for_selectors(), vec!["a", "b"].into());
	assert_eq!(identifier.unique_key(), vec!["a", "0", "b", "1"].into());

	let identifier = ComponentIdentifier::parse_from_moniker("./a:0/coll:comp:1/b:0").unwrap();
	assert_eq!(identifier.relative_moniker_for_selectors(), vec!["a", "coll:comp", "b"].into());
	assert_eq!(identifier.unique_key(), vec!["a", "0", "coll:comp", "1", "b", "0"].into());

	let identifier = ComponentIdentifier::parse_from_moniker(".").unwrap();
	assert!(identifier.relative_moniker_for_selectors().is_empty());
	assert!(identifier.unique_key().is_empty());
	}

	#[fuchsia_async::run_singlethreaded(test)] // we need an executor for the fidl types
	async fn validate_logsink_requested_event() {
	let target_moniker = "./foo:0";
	let target_url = "http://foo.com".to_string();
	let (_log_sink_proxy, log_sink_server_end) =
	fidl::endpoints::create_proxy::<LogSinkMarker>().unwrap();
	let raw_event = create_log_sink_requested_event(
	target_moniker.to_owned(),
	target_url.clone(),
	log_sink_server_end.into_channel(),
	);
	let event = match raw_event.try_into().unwrap() {
	ComponentEvent::LogSinkRequested(e) => e,
	other => panic!("incorrect event type received: {:?}", other),
	};

	assert_eq!(
	event.metadata.identity,
	ComponentIdentity::from_identifier_and_url(
	&ComponentIdentifier::parse_from_moniker(target_moniker).unwrap(),
	target_url
	)
	);
	}
	}