From 81eca6ab165fb8dd8c299318cddee769827dfd2b Mon Sep 17 00:00:00 2001 From: Johan Peltenburg Date: Wed, 5 Aug 2026 07:33:49 +0200 Subject: [PATCH 1/2] feat(instrumentation): support typed callback providers Signed-off-by: Johan Peltenburg --- Cargo.lock | 4 +- crates/codegen/src/cxx_bridge.rs | 2 +- crates/codegen/src/pyo3_bridge.rs | 2 +- .../instrumentation-build/example/Cargo.lock | 3 +- crates/instrumentation-build/example/build.rs | 11 +- .../instrumentation-build/example/src/main.rs | 19 +-- crates/instrumentation-build/src/any_event.rs | 148 ------------------ crates/instrumentation-build/src/lib.rs | 98 +++--------- .../src/runtime/context.rs | 67 ++++---- .../instrumentation-build/src/runtime/mod.rs | 29 +++- crates/instrumentation/Cargo.toml | 8 +- crates/instrumentation/benches/event_emit.rs | 2 +- crates/instrumentation/src/context.rs | 2 +- crates/instrumentation/src/lib.rs | 14 +- crates/instrumentation/src/model.rs | 48 +++--- crates/instrumentation/src/noop.rs | 65 ++++++++ crates/instrumentation/src/sidecar.rs | 22 +++ .../tests/collector_roundtrip.rs | 2 +- .../instrumentation/tests/runtime_flavors.rs | 2 +- crates/io/Cargo.toml | 2 - crates/io/callback/src/lib.rs | 115 +++++++------- crates/io/src/lib.rs | 30 +--- crates/io/types/src/lib.rs | 2 +- crates/model-macros/src/lib.rs | 2 +- crates/model-macros/src/model_macro.rs | 72 ++++----- crates/model/Cargo.toml | 2 +- crates/model/src/lib.rs | 4 +- domains/query_engine/tests/fixed/Cargo.toml | 2 +- domains/query_engine/tests/fixed/src/main.rs | 5 +- .../tests/fixed/tests/data_flow.rs | 12 +- .../tests/fixed/tests/list_entities.rs | 12 +- examples/readme/src/main.rs | 2 +- examples/simulator/application/src/main.rs | 5 +- examples/simulator/instrumentation/Cargo.toml | 1 - examples/simulator/instrumentation/src/lib.rs | 2 - .../instrumentation/src/test_utils.rs | 63 -------- examples/simulator/server/src/main.rs | 3 +- integrations/nvtx/example/src/lib.rs | 6 +- integrations/nvtx/example/src/main.rs | 8 +- integrations/nvtx/example/tests/capture.rs | 8 +- integrations/nvtx/example/tests/thread_id.rs | 8 +- 41 files changed, 364 insertions(+), 550 deletions(-) delete mode 100644 crates/instrumentation-build/src/any_event.rs create mode 100644 crates/instrumentation/src/noop.rs delete mode 100644 examples/simulator/instrumentation/src/test_utils.rs diff --git a/Cargo.lock b/Cargo.lock index b066c1b93..c763bbd0d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2690,6 +2690,7 @@ dependencies = [ name = "quent-instrumentation" version = "0.1.0" dependencies = [ + "async-trait", "criterion", "http", "pprof", @@ -2699,6 +2700,7 @@ dependencies = [ "quent-dynamic-attributes", "quent-events", "quent-io", + "quent-io-callback", "serde", "tempfile", "thiserror", @@ -2733,7 +2735,6 @@ dependencies = [ "clap", "http", "quent-events", - "quent-io-callback", "quent-io-collector", "quent-io-msgpack", "quent-io-ndjson", @@ -3119,7 +3120,6 @@ dependencies = [ name = "quent-simulator-instrumentation" version = "0.1.0" dependencies = [ - "quent-io-callback", "quent-model", "quent-query-engine-model", "quent-stdlib", diff --git a/crates/codegen/src/cxx_bridge.rs b/crates/codegen/src/cxx_bridge.rs index 4247a41d8..11a664647 100644 --- a/crates/codegen/src/cxx_bridge.rs +++ b/crates/codegen/src/cxx_bridge.rs @@ -583,7 +583,7 @@ fn emit_context_bridge( ); inner.block_on(async { let (#(#build_fields,)*) = #q::tokio::try_join!( - #(inner.observer::<#build_event_tys>(options.clone()),)* + #(inner.observer::<#build_event_tys>(&options),)* ) .map_err(|e| e.to_string())?; Ok::<_, String>((#(#build_wraps(#build_fields),)*)) diff --git a/crates/codegen/src/pyo3_bridge.rs b/crates/codegen/src/pyo3_bridge.rs index 9d745ff8c..d1b8c62ab 100644 --- a/crates/codegen/src/pyo3_bridge.rs +++ b/crates/codegen/src/pyo3_bridge.rs @@ -766,7 +766,7 @@ fn emit_context( ); inner.block_on(async { let (#(#build_fields,)*) = #q::tokio::try_join!( - #(inner.observer::<#build_event_tys>(options.clone()),)* + #(inner.observer::<#build_event_tys>(&options),)* ) .map_err(|err| pyo3::exceptions::PyRuntimeError::new_err(err.to_string()))?; Ok::<_, pyo3::PyErr>((#(#build_wraps(#build_fields),)*)) diff --git a/crates/instrumentation-build/example/Cargo.lock b/crates/instrumentation-build/example/Cargo.lock index b5a49eff9..59f0a3180 100644 --- a/crates/instrumentation-build/example/Cargo.lock +++ b/crates/instrumentation-build/example/Cargo.lock @@ -374,10 +374,12 @@ dependencies = [ name = "quent-instrumentation" version = "0.1.0" dependencies = [ + "async-trait", "quent-build-info", "quent-dynamic-attributes", "quent-events", "quent-io", + "quent-io-callback", "thiserror", "tokio", "tokio-util", @@ -415,7 +417,6 @@ version = "0.1.0" dependencies = [ "async-trait", "quent-events", - "quent-io-callback", "quent-io-types", "uuid", ] diff --git a/crates/instrumentation-build/example/build.rs b/crates/instrumentation-build/example/build.rs index 8495b3a3c..a149d4127 100644 --- a/crates/instrumentation-build/example/build.rs +++ b/crates/instrumentation-build/example/build.rs @@ -21,13 +21,10 @@ fn main() -> Result<(), Box> { // Schema -> generated Rust instrumentation source. let opts = Options { - // To just print the events in this example, we'll be using the callback - // exporter. This exporter takes a type-erased event, so in order to - // simplify downcasting back to a statically-typed event, this features - // enables the generation of the "AnyEvent" helper type (see main.rs). - // This is typically left false when using "real" exporters. - any_event: true, - ..Default::default() + // Generate `DemoEvent`, which lets one typed callback receive events + // from every entity in the model. + umbrella_event: true, + ..Options::default() }; let GenerateInfo { path, warnings } = generate(&parsed.schema, &opts)?; diff --git a/crates/instrumentation-build/example/src/main.rs b/crates/instrumentation-build/example/src/main.rs index e4753e493..724ff71ba 100644 --- a/crates/instrumentation-build/example/src/main.rs +++ b/crates/instrumentation-build/example/src/main.rs @@ -1,9 +1,9 @@ // SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 -use quent_instrumentation::{EventCallback, ExporterOptions}; +use quent_instrumentation::EventCallback; -use crate::demo::{Connection, Context, Demo, Handle, Observer, Query, Server, Uuid}; +use crate::demo::{Connection, Context, Demo, DemoEvent, Handle, Observer, Query, Server, Uuid}; #[allow(unused)] mod demo { @@ -11,8 +11,9 @@ mod demo { } fn main() -> Result<(), Box> { - // The context owns the exporter and exposes one observer per entity type. - let context: Context = Context::try_new(Some(debug_printing_exporter()))?; + // The context builds one exporter pipeline per entity event type and + // exposes the corresponding typed observers. + let context: Context = Context::try_new(println_exporter())?; // `observer.handle()` creates a fresh entity instance to events emit for. let mut server = context.observer::().handle(); @@ -71,11 +72,7 @@ fn main() -> Result<(), Box> { Ok(()) } -/// Return an exporter that debug-prints each emitted event's payload. -fn debug_printing_exporter() -> ExporterOptions { - ExporterOptions::Callback(EventCallback::new(|recorded| { - if let Some(event) = demo::AnyEvent::from_any(recorded.event.as_ref()) { - println!("{event:?}"); - } - })) +/// Return a callback that debug-prints each emitted event. +fn println_exporter() -> EventCallback { + EventCallback::new(|event| println!("{event:?}")) } diff --git a/crates/instrumentation-build/src/any_event.rs b/crates/instrumentation-build/src/any_event.rs deleted file mode 100644 index 54b800645..000000000 --- a/crates/instrumentation-build/src/any_event.rs +++ /dev/null @@ -1,148 +0,0 @@ -// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. -// SPDX-License-Identifier: Apache-2.0 - -//! Generation of `AnyEvent`: a decoder from a type-erased event to the concrete -//! `Event` for whichever entity produced it. - -use convert_case::Case; -use proc_macro2::TokenStream; -use quote::quote; -use syn::Ident; - -use crate::common::{ - derive_attr, module_ident, path_name_pascal, raw_ident, relative_type_path, to_case, -}; -use crate::namespace::Namespace; -use crate::{GenerateError, Options}; - -/// Generate `AnyEvent` and its `from_any` decoder. -/// -/// # Errors -/// -/// Returns [`GenerateError`] if a derive entry is not a parseable Rust path. -pub(crate) fn generate_any_event( - namespace: &Namespace<'_>, - opts: &Options, -) -> Result { - let variants: Vec<(Ident, TokenStream)> = namespace - .entities() - .iter() - .map(|entity| { - let variant = raw_ident(path_name_pascal(entity.path())); - let event = relative_type_path(entity.path(), namespace.path(), "Event"); - (variant, event) - }) - .collect(); - let children: Vec<(Ident, Ident)> = namespace - .children_with_entities() - .map(|child| { - let segment = child - .path() - .last() - .expect("child namespaces extend their parent"); - ( - raw_ident(to_case(segment, Case::Pascal)), - module_ident(segment), - ) - }) - .collect(); - if variants.is_empty() && children.is_empty() { - return Ok(quote! {}); - } - - let derives = derive_attr(opts.event_derives, opts.debug, opts.serde, false)?; - let runtime = opts.event_runtime(); - let decls = variants.iter().map(|(variant, event)| { - quote! { #variant(&'a #runtime::Event<#event>) } - }); - let child_decls = children.iter().map(|(variant, module)| { - quote! { #variant(#module::AnyEvent<'a>) } - }); - let direct_arms = variants.iter().map(|(variant, event)| { - quote! { - if let Some(event) = any.downcast_ref::<#runtime::Event<#event>>() { - return Some(Self::#variant(event)); - } - } - }); - let child_arms = children.iter().map(|(variant, module)| { - quote! { - if let Some(event) = #module::AnyEvent::from_any(any) { - return Some(Self::#variant(event)); - } - } - }); - - Ok(quote! { - #derives - pub enum AnyEvent<'a> { - #(#decls,)* - #(#child_decls,)* - } - impl<'a> AnyEvent<'a> { - pub fn from_any(any: &'a dyn ::core::any::Any) -> Option { - #(#direct_arms)* - #(#child_arms)* - None - } - } - }) -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::common::pretty; - use quent_schema::builder::SchemaBuilder; - use quent_schema::test_utils::{entity, event, ident}; - - #[test] - fn emits_a_variant_and_arm_per_entity() { - let schema = SchemaBuilder::new(ident("Demo")) - .with_entity(entity("Query", [event("submitted", [])])) - .with_entity(entity("Server", [event("booted", [])])) - .build() - .unwrap(); - let opts = Options::default(); - let namespaces = Namespace::root(&schema); - let expected = quote! { - #[derive(Debug)] - pub enum AnyEvent<'a> { - Query(&'a ::quent_instrumentation::Event), - Server(&'a ::quent_instrumentation::Event) - } - - impl<'a> AnyEvent<'a> { - pub fn from_any(any: &'a dyn ::core::any::Any) -> Option { - if let Some(event) = - any.downcast_ref::<::quent_instrumentation::Event>() - { - return Some(Self::Query(event)); - } - if let Some(event) = - any.downcast_ref::<::quent_instrumentation::Event>() - { - return Some(Self::Server(event)); - } - None - } - } - }; - assert_eq!( - pretty(generate_any_event(&namespaces, &opts).unwrap()), - pretty(expected) - ); - } - - #[test] - fn emits_nothing_without_entities() { - let schema = SchemaBuilder::new(ident("Demo")).build().unwrap(); - let namespaces = Namespace::root(&schema); - - assert!( - generate_any_event(&namespaces, &Options::default()) - .unwrap() - .is_empty() - ); - } -} diff --git a/crates/instrumentation-build/src/lib.rs b/crates/instrumentation-build/src/lib.rs index cd3b48e25..15e7254db 100644 --- a/crates/instrumentation-build/src/lib.rs +++ b/crates/instrumentation-build/src/lib.rs @@ -44,7 +44,6 @@ //! must also depend on `serde` with its derive feature and enable the matching //! runtime crate's `serde` feature. -mod any_event; mod common; mod data_type; mod events; @@ -70,7 +69,6 @@ pub struct Options { /// Derive `serde::Serialize` and `serde::Deserialize` on generated event /// and record types. /// - /// `AnyEvent` is borrowed, so it derives only `serde::Serialize`. pub serde: bool, /// Derives applied to every generated event payload enum. @@ -90,13 +88,6 @@ pub struct Options { /// `None`. pub file_name: Option, - /// Emit root and namespace-local `AnyEvent` enums that decode type-erased - /// events. Each enum carries [`Self::debug`], compatible serde derives, and - /// [`Self::event_derives`]. - /// - /// No aggregate is emitted for a namespace without events. - pub any_event: bool, - /// Emit model-wide umbrella event enums and implement the umbrella /// capability for the generated model. /// @@ -117,7 +108,6 @@ impl Default for Options { record_derives: Default::default(), out_dir: PathBuf::from(std::env::var("OUT_DIR").unwrap_or_default()), file_name: None, - any_event: false, umbrella_event: false, analyzer_package: None, } @@ -215,21 +205,15 @@ pub fn generate_str(schema: &Schema, opts: &Options) -> Result(quote! { #reexports #entity_types #types #observable - #any_event }) .map_err(GenerateError::InvalidGeneratedCode)?; Ok(prettyplease::unparse(&file)) @@ -239,7 +223,6 @@ fn generate_namespace( schema: &Schema, opts: &Options, namespace: &namespace::Namespace<'_>, - include_any_event: bool, ) -> Result { let records = namespace .records() @@ -274,7 +257,7 @@ fn generate_namespace( .last() .expect("child namespaces extend their parent"); let module = common::module_ident(segment); - let contents = generate_namespace(schema, opts, child, true)?; + let contents = generate_namespace(schema, opts, child)?; Ok::<_, GenerateError>(quote! { pub mod #module { #contents @@ -282,11 +265,6 @@ fn generate_namespace( }) }) .collect::, _>>()?; - let any_event = if include_any_event && opts.any_event && namespace.has_entities() { - any_event::generate_any_event(namespace, opts)? - } else { - quote! {} - }; let observer_storage = if opts.instrumentation { runtime::observer_storage(schema, namespace)? } else { @@ -301,7 +279,6 @@ fn generate_namespace( #(#children)* #model #observer_storage - #any_event }) } @@ -315,6 +292,28 @@ mod path_tests { use quent_schema::test_utils::{entity, event, field, path, record, record_type}; use quent_schema::{Annotations, DataType}; + #[test] + fn generates_event_only_umbrella_without_instrumentation() { + let schema = SchemaBuilder::try_new("Demo") + .unwrap() + .with_entity(entity("Query", [event("created", [])])) + .build() + .unwrap(); + let opts = Options { + instrumentation: false, + umbrella_event: true, + ..Options::default() + }; + + let source = generate_str(&schema, &opts).unwrap(); + + assert!(source.contains("impl ::quent_events::ModelEvents for Demo")); + assert!(source.contains("pub enum DemoEvent")); + assert!(!source.contains("quent_instrumentation")); + assert!(!source.contains("pub struct Handle")); + assert!(!source.contains("Observers")); + } + #[test] fn built_in_derive_path_spellings_are_deduplicated() { let schema = SchemaBuilder::try_new("Demo") @@ -486,53 +485,4 @@ mod path_tests { source.contains("any: ::quent_instrumentation::EntityRef") ); } - - #[test] - fn generates_any_event_per_entity_namespace() { - let schema = SchemaBuilder::try_new("Demo") - .unwrap() - .with_entity(entity("Root", [event("created", [])])) - .with_entity(entity("Foo::Query", [event("created", [])])) - .with_entity(entity("Foo::Nested::Task", [event("created", [])])) - .build() - .unwrap(); - let opts = Options { - any_event: true, - ..Options::default() - }; - - let source = generate_str(&schema, &opts).unwrap(); - assert!(source.contains("Root(&'a ::quent_instrumentation::Event)")); - assert!(source.contains("Foo(foo::AnyEvent<'a>)")); - assert!(source.contains("Query(&'a ::quent_instrumentation::Event)")); - assert!(source.contains("Nested(nested::AnyEvent<'a>)")); - assert!(source.contains("Task(&'a ::quent_instrumentation::Event)")); - assert!(source.contains("foo::AnyEvent::from_any(any)")); - assert!(source.contains("nested::AnyEvent::from_any(any)")); - assert!( - source.rfind("pub enum AnyEvent") - > source.rfind("impl ::quent_instrumentation::InstrumentedModel") - ); - } - - #[test] - fn generates_any_event_without_instrumentation() { - let schema = SchemaBuilder::try_new("Demo") - .unwrap() - .with_entity(entity("Query", [event("created", [])])) - .build() - .unwrap(); - let opts = Options { - instrumentation: false, - any_event: true, - ..Options::default() - }; - - let source = generate_str(&schema, &opts).unwrap(); - assert!(source.contains("Query(&'a ::quent_events::Event)")); - assert!(!source.contains("quent_instrumentation")); - assert!(!source.contains("pub struct Handle")); - assert!(!source.contains("Observers")); - assert!(!source.contains("impl ::quent_instrumentation::Model")); - } } diff --git a/crates/instrumentation-build/src/runtime/context.rs b/crates/instrumentation-build/src/runtime/context.rs index e5237e7c8..17396a5ca 100644 --- a/crates/instrumentation-build/src/runtime/context.rs +++ b/crates/instrumentation-build/src/runtime/context.rs @@ -92,13 +92,22 @@ pub(super) fn observer_storage( pub(super) fn schema_model(schema: &Schema, namespaces: &Namespace<'_>) -> TokenStream { let model = model_ident(schema); let observers = observers_ident(schema, namespaces); - let active_observers = observer_storage_initializer(schema, namespaces, true); - let noop_observers = observer_storage_initializer(schema, namespaces, false); - let options_binding = if schema.entities().next().is_some() { - raw_ident("options".to_owned()) + let observers_initializer = observer_storage_initializer(schema, namespaces); + let provider_binding = if schema.entities().next().is_some() { + raw_ident("provider".to_owned()) } else { - raw_ident("_options".to_owned()) + raw_ident("_provider".to_owned()) }; + let provider_event_types = schema + .entities() + .map(|entity| relative_type_path(entity.path(), &[], "Event")) + .collect::>(); + let provider_bounds = (!provider_event_types.is_empty()).then(|| { + quote! { + where + #(P: ::quent_instrumentation::ExporterProvider<#provider_event_types>,)* + } + }); let observer_impls = schema .entities() .map(|entity| observer_storage_impl(schema, entity)); @@ -108,53 +117,39 @@ pub(super) fn schema_model(schema: &Schema, namespaces: &Namespace<'_>) -> Token impl ::quent_instrumentation::InstrumentedModel for #model { type Observers = #observers; + } + impl

::quent_instrumentation::ObserverBuilder

for #model + #provider_bounds + { fn build_observers( context: &::quent_instrumentation::ContextInner, - exporter: ::core::option::Option<&::quent_instrumentation::ExporterOptions>, + #provider_binding: &P, ) -> ::core::result::Result< Self::Observers, ::std::boxed::Box, > { - match exporter { - ::core::option::Option::Some(#options_binding) => { - context.block_on(async { - ::core::result::Result::< - _, - ::std::boxed::Box, - >::Ok(#active_observers) - }) - } - ::core::option::Option::None => { - ::core::result::Result::Ok(#noop_observers) - } - } + context.block_on(async { + ::core::result::Result::< + _, + ::std::boxed::Box, + >::Ok(#observers_initializer) + }) } - } } } -fn observer_storage_initializer( - schema: &Schema, - namespace: &Namespace<'_>, - active: bool, -) -> TokenStream { +fn observer_storage_initializer(schema: &Schema, namespace: &Namespace<'_>) -> TokenStream { let storage = observers_path(schema, namespace); let entity_fields = namespace.entities().iter().map(|entity| { let field = entity_observer_field(entity); let entity_ty = relative_type_path(entity.path(), &[], ""); let event_ty = relative_type_path(entity.path(), &[], "Event"); - let observer = if active { - quote! { - context - .observer::<#event_ty>(::core::clone::Clone::clone(options)) - .await? - } - } else { - quote! { - ::quent_instrumentation::ObserverInner::<#event_ty>::noop() - } + let observer = quote! { + context + .observer::<#event_ty>(provider) + .await? }; quote! { #field: ::quent_instrumentation::Observer::<#entity_ty>::new( @@ -168,7 +163,7 @@ fn observer_storage_initializer( .last() .expect("child namespaces extend their parent"); let field = namespace_observers_field(segment); - let value = observer_storage_initializer(schema, child, active); + let value = observer_storage_initializer(schema, child); quote! { #field: #value } }); quote! { diff --git a/crates/instrumentation-build/src/runtime/mod.rs b/crates/instrumentation-build/src/runtime/mod.rs index 63e2315b6..f182b3b11 100644 --- a/crates/instrumentation-build/src/runtime/mod.rs +++ b/crates/instrumentation-build/src/runtime/mod.rs @@ -84,7 +84,8 @@ pub(crate) fn entity_types(schema: &Schema) -> TokenStream { pub(crate) fn reexports() -> TokenStream { quote! { pub use ::quent_instrumentation::{ - AnyEntity, Context, DynamicAttributes, EntityRef, Event, HandleError, Observer, Uuid, + AnyEntity, Context, DynamicAttributes, EntityRef, Event, HandleError, Noop, Observer, + Uuid, }; } } @@ -125,7 +126,7 @@ mod tests { use quent_schema::Cardinality; use quent_schema::DataType; use quent_schema::builder::{EntityBuilder, EventBuilder, SchemaBuilder}; - use quent_schema::test_utils::{field, ident}; + use quent_schema::test_utils::{entity, event, field, ident}; #[test] fn generate_assembles_event_impl_observer_handle_and_context() { @@ -157,5 +158,29 @@ mod tests { assert!(src.contains("type Event = ConnectionEvent")); assert!(src.contains("impl Handle")); assert!(src.contains("pub struct Demo")); + assert!(src.contains("impl

::quent_instrumentation::ObserverBuilder

for Demo")); + assert!(src.contains("P: ::quent_instrumentation::ExporterProvider")); + } + + #[test] + fn generates_provider_observers_for_nested_namespaces() { + let schema = SchemaBuilder::try_new("Demo") + .unwrap() + .with_entity(entity("Foo::Query", [event("created", [])])) + .with_entity(entity("Foo::Nested::Task", [event("created", [])])) + .build() + .unwrap(); + let namespaces = crate::namespace::Namespace::root(&schema); + + let src = pretty(generate_model(&schema, &namespaces)); + + assert!(src.contains("P: ::quent_instrumentation::ExporterProvider")); + assert!( + src.contains("P: ::quent_instrumentation::ExporterProvider") + ); + assert!(src.contains("context.observer::(provider)")); + assert!(src.contains("context.observer::(provider)")); + assert!(src.contains("foo::FooObservers")); + assert!(src.contains("foo::nested::NestedObservers")); } } diff --git a/crates/instrumentation/Cargo.toml b/crates/instrumentation/Cargo.toml index 7ea55218b..c67a79452 100644 --- a/crates/instrumentation/Cargo.toml +++ b/crates/instrumentation/Cargo.toml @@ -12,18 +12,20 @@ serde = [ "quent-dynamic-attributes/serde", "uuid/serde", ] -# Exporter backends, forwarded to the matching `quent-io` feature. -io-callback = ["quent-io/callback"] +# Optional I/O integrations. +io-callback = ["dep:quent-io-callback"] io-ndjson = ["quent-io/ndjson", "serde"] io-msgpack = ["quent-io/msgpack", "serde"] io-postcard = ["quent-io/postcard", "serde"] io-collector = ["quent-io/collector", "serde"] [dependencies] +async-trait.workspace = true quent-dynamic-attributes = { path = "../dynamic-attributes" } quent-build-info = { path = "../build-info" } quent-events = { path = "../events" } quent-io = { path = "../io", default-features = false } +quent-io-callback = { path = "../io/callback", optional = true } serde = { workspace = true, features = ["derive"], optional = true } thiserror.workspace = true tokio = { workspace = true, features = ["rt-multi-thread", "sync", "macros"]} @@ -38,7 +40,7 @@ pprof = { version = "0.15", features = ["flamegraph"] } quent-collector = { path = "../collector/server" } quent-collector-proto = { path = "../collector/proto" } quent-events = { path = "../events", features = ["serde"] } -quent-io = { path = "../io", features = ["callback", "ndjson"] } +quent-io = { path = "../io", features = ["ndjson"] } serde = { workspace = true, features = ["derive"] } tempfile = "3" tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } diff --git a/crates/instrumentation/benches/event_emit.rs b/crates/instrumentation/benches/event_emit.rs index 00d2a1a4d..1ec46be8b 100644 --- a/crates/instrumentation/benches/event_emit.rs +++ b/crates/instrumentation/benches/event_emit.rs @@ -89,7 +89,7 @@ fn build_observer( return Ok((ContextInner::noop(id), ObserverInner::noop())); }; let ctx = ContextInner::try_new(id)?; - let observer = ctx.block_on(async { ctx.observer::(options).await })?; + let observer = ctx.block_on(async { ctx.observer::(&options).await })?; Ok((ctx, observer)) } diff --git a/crates/instrumentation/src/context.rs b/crates/instrumentation/src/context.rs index 143bdbc03..18db990fd 100644 --- a/crates/instrumentation/src/context.rs +++ b/crates/instrumentation/src/context.rs @@ -138,7 +138,7 @@ impl ContextInner { /// context builds no exporter. pub async fn observer( &self, - provider: impl ExporterProvider, + provider: &impl ExporterProvider, ) -> Result, Box> where T: Send + EntityEvent + 'static, diff --git a/crates/instrumentation/src/lib.rs b/crates/instrumentation/src/lib.rs index e8d5d7975..725b6ab07 100644 --- a/crates/instrumentation/src/lib.rs +++ b/crates/instrumentation/src/lib.rs @@ -11,15 +11,17 @@ mod context; mod entity; mod handle; mod model; +mod noop; mod observer; mod sidecar; pub use context::ContextInner; pub use entity::{InstrumentedEntity, Observer}; pub use handle::{HandleError, HandleInner}; -pub use model::{Context, InstrumentedModel, ObserverProvider}; +pub use model::{Context, InstrumentedModel, ObserverBuilder, ObserverProvider}; +pub use noop::Noop; pub use observer::{EventSender, ObserverInner}; -pub use sidecar::write_sidecar; +pub use sidecar::{ContextExporter, write_sidecar}; // Re-export everything the generated instrumentation code references, so a // consumer needs only the `quent-instrumentation` dependency, selecting an @@ -29,12 +31,12 @@ pub use quent_dynamic_attributes::DynamicAttributes; #[doc(hidden)] pub use quent_events as events; pub use quent_events::{AnyEntity, EntityEvent, EntityRef, Event, Model, ModelEvents}; -pub use quent_io::ExporterOptions; +pub use quent_io::{ExporterOptions, ExporterProvider}; pub use uuid::Uuid; -/// A caller-supplied event sink, selected via the `io-callback` feature. +/// A caller-supplied typed event sink, selected via the `io-callback` feature. #[cfg(feature = "io-callback")] -pub use quent_io::EventCallback; +pub use quent_io_callback::EventCallback; #[cfg(test)] mod tests { @@ -81,7 +83,7 @@ mod tests { { let observer = ctx - .block_on(async { ctx.observer::(options).await }) + .block_on(async { ctx.observer::(&options).await }) .unwrap(); observer.send(Event::new_now(Uuid::now_v7(), TestEvent)); // Drop the observer to drain and flush before asserting. diff --git a/crates/instrumentation/src/model.rs b/crates/instrumentation/src/model.rs index d502fdccd..fcf7990eb 100644 --- a/crates/instrumentation/src/model.rs +++ b/crates/instrumentation/src/model.rs @@ -3,7 +3,7 @@ //! Instrumentation models and their contexts. -use crate::{ContextInner, ExporterOptions, InstrumentedEntity, Observer, Uuid, write_sidecar}; +use crate::{ContextExporter, ContextInner, InstrumentedEntity, Observer, Uuid}; /// Provides typed access to an entity observer in a generated model. /// @@ -22,20 +22,23 @@ pub trait InstrumentedModel { /// Hidden because callers access observers through [`Context::observer`]. #[doc(hidden)] type Observers; +} - /// Builds the observers for this model. - /// - /// `exporter` is `None` for a no-op context. - /// - /// Hidden because [`Context`] invokes it during construction. +/// Builds a model's observers from an exporter provider. +/// +/// Generated implementations require `P` to provide an exporter for every +/// entity event type in the model. +#[doc(hidden)] +pub trait ObserverBuilder

: InstrumentedModel { + /// Builds every observer from `provider`. /// /// # Errors /// - /// Returns an error when an observer or its exporter cannot be constructed. + /// Returns an error when an observer or exporter cannot be constructed. #[doc(hidden)] fn build_observers( context: &ContextInner, - exporter: Option<&ExporterOptions>, + provider: &P, ) -> Result>; } @@ -47,32 +50,27 @@ pub struct Context { impl Context { /// Creates a context and builds every entity's exporter pipeline. - /// - /// Passing `None` creates a no-op context that discards events. - pub fn try_new(exporter: Option) -> Result> + pub fn try_new

(provider: P) -> Result> where - M: crate::build_info::ModelSource, + M: crate::build_info::ModelSource + ObserverBuilder

, + P: ContextExporter, { - Self::try_with_id(Uuid::now_v7(), exporter) + Self::try_with_id(Uuid::now_v7(), provider) } /// Creates a context with the supplied ID. - pub fn try_with_id( - id: Uuid, - exporter: Option, - ) -> Result> + pub fn try_with_id

(id: Uuid, provider: P) -> Result> where - M: crate::build_info::ModelSource, + M: crate::build_info::ModelSource + ObserverBuilder

, + P: ContextExporter, { - let inner = if exporter.is_some() { - ContextInner::try_new(id)? - } else { + let inner = if provider.is_noop() { ContextInner::noop(id) + } else { + ContextInner::try_new(id)? }; - if let Some(options) = &exporter { - write_sidecar(options, id, M::model_info()); - } - let observers = M::build_observers(&inner, exporter.as_ref())?; + provider.prepare_context(id, M::model_info()); + let observers = M::build_observers(&inner, &provider)?; Ok(Self { observers, inner }) } diff --git a/crates/instrumentation/src/noop.rs b/crates/instrumentation/src/noop.rs new file mode 100644 index 000000000..32a1262e0 --- /dev/null +++ b/crates/instrumentation/src/noop.rs @@ -0,0 +1,65 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! No-op event export. + +use quent_build_info::ModelInfo; +use quent_events::Event; +use quent_io::{Exporter, ExporterProvider, ExporterResult}; +use uuid::Uuid; + +use crate::ContextExporter; + +/// An exporter provider that discards every event. +#[derive(Clone, Copy, Debug, Default)] +pub struct Noop; + +#[async_trait::async_trait] +impl Exporter for Noop +where + T: Send + 'static, +{ + async fn push(&mut self, _event: Event) -> ExporterResult<()> { + Ok(()) + } + + async fn shutdown(self: Box) -> ExporterResult<()> { + Ok(()) + } +} + +#[async_trait::async_trait] +impl ExporterProvider for Noop +where + T: Send + 'static, +{ + async fn create_exporter(&self, _context_id: Uuid) -> ExporterResult>> { + Ok(Box::new(*self)) + } +} + +impl ContextExporter for Noop { + fn is_noop(&self) -> bool { + true + } + + fn prepare_context(&self, _context_id: Uuid, _model: ModelInfo) {} +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn provides_an_exporter_for_any_event_type() { + let mut exporter: Box> = + Noop.create_exporter(Uuid::nil()).await.unwrap(); + + exporter + .push(Event::new(Uuid::nil(), 0, "ignored".to_owned())) + .await + .unwrap(); + exporter.shutdown().await.unwrap(); + assert!(Noop.is_noop()); + } +} diff --git a/crates/instrumentation/src/sidecar.rs b/crates/instrumentation/src/sidecar.rs index 56e839bd8..ed08f3929 100644 --- a/crates/instrumentation/src/sidecar.rs +++ b/crates/instrumentation/src/sidecar.rs @@ -8,6 +8,28 @@ use quent_io::ExporterOptions; use tracing::warn; use uuid::Uuid; +/// Prepares context-wide output for an exporter provider. +pub trait ContextExporter { + /// Returns whether this provider disables exporter pipelines. + fn is_noop(&self) -> bool { + false + } + + /// Prepares output for `model` under `context_id`. + fn prepare_context(&self, context_id: Uuid, model: ModelInfo); +} + +impl ContextExporter for ExporterOptions { + fn prepare_context(&self, context_id: Uuid, model: ModelInfo) { + write_sidecar(self, context_id, model); + } +} + +#[cfg(feature = "io-callback")] +impl ContextExporter for quent_io_callback::EventCallback { + fn prepare_context(&self, _context_id: Uuid, _model: ModelInfo) {} +} + /// Write the model provenance sidecar file into the filesystem exporter /// directory. /// diff --git a/crates/instrumentation/tests/collector_roundtrip.rs b/crates/instrumentation/tests/collector_roundtrip.rs index 59ce7be51..0f46cf80d 100644 --- a/crates/instrumentation/tests/collector_roundtrip.rs +++ b/crates/instrumentation/tests/collector_roundtrip.rs @@ -85,7 +85,7 @@ fn collector_client_flushes_all_events_on_drop() { let options = ExporterOptions::Collector(CollectorExporterOptions::new(address)); { let observer = ctx - .block_on(async { ctx.observer::(options).await }) + .block_on(async { ctx.observer::(&options).await }) .unwrap(); for _ in 0..EVENTS { observer.emit(Uuid::now_v7(), TestEvent); diff --git a/crates/instrumentation/tests/runtime_flavors.rs b/crates/instrumentation/tests/runtime_flavors.rs index 69181bea8..e89cb2d32 100644 --- a/crates/instrumentation/tests/runtime_flavors.rs +++ b/crates/instrumentation/tests/runtime_flavors.rs @@ -34,7 +34,7 @@ fn active(root: &Path) -> (ContextInner, ExporterOptions, Uuid) { /// Build an observer through the one bridge: the context builds the exporter /// from the options (bound to its id) and hosts it on its runtime. fn build(ctx: &ContextInner, exporter_opts: &ExporterOptions) -> ObserverInner { - ctx.block_on(async { ctx.observer::(exporter_opts.clone()).await }) + ctx.block_on(async { ctx.observer::(exporter_opts).await }) .unwrap() } diff --git a/crates/io/Cargo.toml b/crates/io/Cargo.toml index d78639cad..33c0b279e 100644 --- a/crates/io/Cargo.toml +++ b/crates/io/Cargo.toml @@ -11,7 +11,6 @@ msgpack = ["dep:quent-io-msgpack", "dep:serde"] postcard = ["dep:quent-io-postcard", "dep:serde"] collector = ["dep:quent-io-collector", "dep:http", "dep:serde"] clap = ["dep:clap", "dep:http"] -callback = ["dep:quent-io-callback"] [dependencies] async-trait.workspace = true @@ -19,7 +18,6 @@ clap = { version = "4.5.57", features = ["derive", "env"], optional = true } http = { workspace = true, optional = true } quent-events = { path = "../events" } quent-io-types = { path = "types" } -quent-io-callback = { path = "callback", optional = true } quent-io-collector = { path = "collector", optional = true } quent-io-msgpack = { path = "msgpack", optional = true } quent-io-ndjson = { path = "ndjson", optional = true } diff --git a/crates/io/callback/src/lib.rs b/crates/io/callback/src/lib.rs index 101201416..0d8846321 100644 --- a/crates/io/callback/src/lib.rs +++ b/crates/io/callback/src/lib.rs @@ -4,57 +4,51 @@ //! An exporter that hands each event to a caller-supplied callback. Intended //! for tests that collect emitted events in memory. -use std::any::Any; use std::sync::Arc; -use quent_events::{EntityEvent, Event}; +use quent_events::Event; use quent_io_types::{Exporter, ExporterProvider, ExporterResult}; -/// One exported event, type-erased so a single callback can receive events of -/// any entity type. `event` is a boxed `Event` for the entity named by -/// `entity`; downcast it with `event.downcast::>()`. -pub struct RecordedEvent { - pub entity: &'static str, - pub event: Box, -} - /// A thread-safe callback invoked once per exported event. -#[derive(Clone)] -pub struct EventCallback(Arc); +pub struct EventCallback(Arc) + Send + Sync>); + +impl Clone for EventCallback { + fn clone(&self) -> Self { + Self(Arc::clone(&self.0)) + } +} -impl EventCallback { - pub fn new(callback: impl Fn(RecordedEvent) + Send + Sync + 'static) -> Self { +impl EventCallback { + pub fn new(callback: impl Fn(Event) + Send + Sync + 'static) -> Self { Self(Arc::new(callback)) } } -impl std::fmt::Debug for EventCallback { +impl std::fmt::Debug for EventCallback { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("EventCallback").finish_non_exhaustive() } } -/// Type-erases each pushed event and forwards it to an [`EventCallback`]. -pub struct CallbackExporter { - callback: EventCallback, +/// Converts each pushed event and forwards it to an [`EventCallback`]. +pub struct CallbackExporter { + callback: EventCallback, } -impl CallbackExporter { - pub fn new(callback: EventCallback) -> Self { +impl CallbackExporter { + pub fn new(callback: EventCallback) -> Self { Self { callback } } } #[async_trait::async_trait] -impl Exporter for CallbackExporter +impl Exporter for CallbackExporter where - T: Send + EntityEvent + 'static, + S: Into + Send + 'static, + T: Send + 'static, { - async fn push(&mut self, event: Event) -> ExporterResult<()> { - (self.callback.0)(RecordedEvent { - entity: T::NAME, - event: Box::new(event), - }); + async fn push(&mut self, event: Event) -> ExporterResult<()> { + (self.callback.0)(Event::new(event.id, event.timestamp, event.data.into())); Ok(()) } @@ -64,15 +58,16 @@ where } #[async_trait::async_trait] -impl ExporterProvider for EventCallback +impl ExporterProvider for EventCallback where - T: Send + EntityEvent + 'static, + S: Into + Send + 'static, + T: Send + 'static, { async fn create_exporter( &self, _context_id: uuid::Uuid, - ) -> ExporterResult>> { - Ok(Box::new(CallbackExporter::new(self.clone()))) + ) -> ExporterResult>> { + Ok(Box::new(CallbackExporter::::new(self.clone()))) } } @@ -84,32 +79,41 @@ mod tests { use super::*; + enum ModelEvent { + Alpha(Alpha), + Beta(Beta), + } + struct Alpha { a: u32, } - impl EntityEvent for Alpha { - const NAME: &'static str = "alpha"; + impl From for ModelEvent { + fn from(value: Alpha) -> Self { + Self::Alpha(value) + } } struct Beta { b: String, } - impl EntityEvent for Beta { - const NAME: &'static str = "beta"; + impl From for ModelEvent { + fn from(value: Beta) -> Self { + Self::Beta(value) + } } - // One shared callback receives events from exporters of two distinct entity - // types; each erased event must downcast back to its concrete `Event`. #[tokio::test] - async fn forwards_multiple_event_types_erased() { - let recorded = Arc::new(Mutex::new(Vec::::new())); + async fn forwards_multiple_event_types_as_model_events() { + let recorded = Arc::new(Mutex::new(Vec::>::new())); let callback = { let recorded = recorded.clone(); EventCallback::new(move |rec| recorded.lock().unwrap().push(rec)) }; - let mut alpha: Box> = Box::new(CallbackExporter::new(callback.clone())); - let mut beta: Box> = Box::new(CallbackExporter::new(callback.clone())); + let mut alpha: Box> = + callback.create_exporter(Uuid::from_u128(10)).await.unwrap(); + let mut beta: Box> = + callback.create_exporter(Uuid::from_u128(10)).await.unwrap(); let (id0, id1, id2) = (Uuid::from_u128(1), Uuid::from_u128(2), Uuid::from_u128(3)); alpha @@ -127,22 +131,17 @@ mod tests { let recorded = recorded.lock().unwrap(); assert_eq!(recorded.len(), 3); - // Entity names are preserved in emission order. - assert_eq!( - recorded.iter().map(|r| r.entity).collect::>(), - [Alpha::NAME, Beta::NAME, Alpha::NAME], - ); - - let a0 = recorded[0].event.downcast_ref::>().unwrap(); - assert_eq!((a0.id, a0.timestamp, a0.data.a), (id0, 10, 7)); - - let b1 = recorded[1].event.downcast_ref::>().unwrap(); - assert_eq!((b1.id, b1.timestamp, b1.data.b.as_str()), (id1, 20, "x")); - - let a2 = recorded[2].event.downcast_ref::>().unwrap(); - assert_eq!((a2.id, a2.data.a), (id2, 9)); - - // A mismatched concrete type does not downcast. - assert!(recorded[1].event.downcast_ref::>().is_none()); + assert_eq!((recorded[0].id, recorded[0].timestamp), (id0, 10)); + assert!(matches!( + recorded[0].data, + ModelEvent::Alpha(Alpha { a: 7 }) + )); + assert_eq!((recorded[1].id, recorded[1].timestamp), (id1, 20)); + assert!(matches!(&recorded[1].data, ModelEvent::Beta(Beta { b }) if b == "x")); + assert_eq!(recorded[2].id, id2); + assert!(matches!( + recorded[2].data, + ModelEvent::Alpha(Alpha { a: 9 }) + )); } } diff --git a/crates/io/src/lib.rs b/crates/io/src/lib.rs index 72a04388a..d73f6aaa8 100644 --- a/crates/io/src/lib.rs +++ b/crates/io/src/lib.rs @@ -7,16 +7,6 @@ use quent_events::EntityEvent; use uuid::Uuid; -// Error out compilation if no exporter is selected at all. -#[cfg(not(any( - feature = "ndjson", - feature = "msgpack", - feature = "postcard", - feature = "collector", - feature = "callback" -)))] -compile_error!("at least one exporter feature must be enabled"); - // Re-exports. pub use quent_io_types::{ Exporter, ExporterProvider, ExporterResult, ImporterError, ImporterProvider, ImporterResult, @@ -27,8 +17,6 @@ pub use quent_io_types::{ pub use crate::filesystem::{ Format as FileSystemFormat, exporter::Options as FileSystemExporterOptions, }; -#[cfg(feature = "callback")] -pub use quent_io_callback::EventCallback; #[cfg(feature = "collector")] pub use quent_io_collector::{CollectorAddressError, Options as CollectorExporterOptions}; @@ -38,16 +26,13 @@ pub mod clap; #[cfg(filesystem)] pub mod filesystem; -/// Where events go: local files (filesystem), a collector service, or a -/// caller-supplied callback (e.g. an in-memory collector for tests). +/// Where events go: local files or a collector service. #[derive(Debug, Clone)] pub enum ExporterOptions { #[cfg(filesystem)] FileSystem(FileSystemExporterOptions), #[cfg(feature = "collector")] Collector(CollectorExporterOptions), - #[cfg(feature = "callback")] - Callback(EventCallback), } impl ExporterOptions { @@ -60,8 +45,8 @@ impl ExporterOptions { ExporterOptions::FileSystem(options) => Some(options.dir(context_id)), #[cfg(feature = "collector")] ExporterOptions::Collector(_) => None, - #[cfg(feature = "callback")] - ExporterOptions::Callback(_) => None, + #[cfg(not(any(filesystem, feature = "collector")))] + _ => None, } } } @@ -78,8 +63,6 @@ where ExporterOptions::FileSystem(options) => options.create_exporter(context_id).await, #[cfg(feature = "collector")] ExporterOptions::Collector(options) => options.create_exporter(context_id).await, - #[cfg(feature = "callback")] - ExporterOptions::Callback(callback) => callback.create_exporter(context_id).await, } } } @@ -90,11 +73,8 @@ impl ExporterProvider for ExporterOptions where T: Send + EntityEvent + 'static, { - async fn create_exporter(&self, context_id: Uuid) -> ExporterResult>> { - match self { - #[cfg(feature = "callback")] - ExporterOptions::Callback(callback) => callback.create_exporter(context_id).await, - } + async fn create_exporter(&self, _context_id: Uuid) -> ExporterResult>> { + unreachable!("ExporterOptions has no enabled variants") } } diff --git a/crates/io/types/src/lib.rs b/crates/io/types/src/lib.rs index 03a9709cf..9399bcbe4 100644 --- a/crates/io/types/src/lib.rs +++ b/crates/io/types/src/lib.rs @@ -52,7 +52,7 @@ pub trait Exporter: Send { /// context whose events it exports). Backends that do not scope by context, such /// as a callback, ignore it. #[async_trait::async_trait] -pub trait ExporterProvider { +pub trait ExporterProvider: Send + Sync { async fn create_exporter(&self, context_id: Uuid) -> ExporterResult>>; } diff --git a/crates/model-macros/src/lib.rs b/crates/model-macros/src/lib.rs index 0dbaa5004..ce1cad888 100644 --- a/crates/model-macros/src/lib.rs +++ b/crates/model-macros/src/lib.rs @@ -90,7 +90,7 @@ pub fn model(input: TokenStream) -> TokenStream { /// This generates `AppContext`, the entry point for instrumenting your /// application. To start emitting events: /// -/// 1. Create a context: `let ctx = AppContext::try_new(Some(exporter_options))?;` +/// 1. Create a context: `let ctx = AppContext::try_new(exporter_options)?;` /// 2. Get an observer: `let obs = ctx.cluster_observer();` /// 3. Declare the root entity with the context id: `obs.cluster(ctx.id(), "my-cluster");` /// diff --git a/crates/model-macros/src/model_macro.rs b/crates/model-macros/src/model_macro.rs index a9f7fad27..d72897fac 100644 --- a/crates/model-macros/src/model_macro.rs +++ b/crates/model-macros/src/model_macro.rs @@ -261,8 +261,7 @@ pub fn expand(input: TokenStream) -> syn::Result { current-thread runtime).\n\ \n\ # Arguments\n\ - * `exporter` — optional exporter configuration (e.g., ndjson, msgpack). \ - Pass `None` for a no-op context that discards events." + * `provider` — exporter provider used for every entity event type." ); let doc_import = format!( @@ -292,48 +291,37 @@ pub fn expand(input: TokenStream) -> syn::Result { None => quote! {}, }; - // The options-driven constructors. In a serde build the exporter provider - // impl bounds each event type by `Serialize`; in a callback-only build it - // does not, so these are `Serialize`-free there. let constructor_api = quote! { impl #context_type { #[doc = #doc_try_new] - pub fn try_new( - exporter: Option, - ) -> Result> { - Self::try_with_id(quent_model::uuid::Uuid::now_v7(), exporter) + pub fn try_new

(provider: P) -> Result> + where + #(P: quent_model::io::ExporterProvider<#event_types>,)* + P: quent_model::ContextExporter, + { + Self::try_with_id(quent_model::uuid::Uuid::now_v7(), provider) } /// Build a context that adopts an existing `id` instead of /// generating one — e.g. the collector reproducing a remote /// source's output under that source's id. Same blocking and /// runtime restriction as [`Self::try_new`]. - pub fn try_with_id( + pub fn try_with_id

( id: quent_model::uuid::Uuid, - exporter: Option, - ) -> Result> { - match exporter { - None => Ok(Self::noop(id)), - Some(options) => { - quent_model::write_sidecar( - &options, - id, - <#name as quent_model::events::Model>::model_info(), - ); - Self::build(id, options) - } - } + provider: P, + ) -> Result> + where + #(P: quent_model::io::ExporterProvider<#event_types>,)* + P: quent_model::ContextExporter, + { + quent_model::ContextExporter::prepare_context( + &provider, + id, + <#name as quent_model::events::Model>::model_info(), + ); + Self::build(id, provider) } - /// A no-op context adopting `id`: every observer discards its events. - fn noop(id: quent_model::uuid::Uuid) -> Self { - Self { - #(#observer_fields: #observer_types::new( - quent_model::Observer::<#event_types>::noop(), - ),)* - _inner: quent_model::ContextInner::noop(id), - } - } } }; @@ -484,17 +472,25 @@ pub fn expand(input: TokenStream) -> syn::Result { impl #context_type { // The single sync/async bridge: on an active context, build // every entity's observer (each constructing its exporter from - // the options, bound to the context id) concurrently on the + // the provider, bound to the context id) concurrently on the // runtime, block until all complete, and assemble. - fn build( + fn build

( id: quent_model::uuid::Uuid, - options: quent_model::io::ExporterOptions, - ) -> Result> { - let inner = quent_model::ContextInner::try_new(id)?; + provider: P, + ) -> Result> + where + #(P: quent_model::io::ExporterProvider<#event_types>,)* + P: quent_model::ContextExporter, + { + let inner = if quent_model::ContextExporter::is_noop(&provider) { + quent_model::ContextInner::noop(id) + } else { + quent_model::ContextInner::try_new(id)? + }; let ( #(#observer_fields,)* ) = inner.block_on(async { let ( #(#observer_fields,)* ) = quent_model::tokio::try_join!( #( - inner.observer::<#event_types>(options.clone()), + inner.observer::<#event_types>(&provider), )* )?; Ok::<_, Box>(( #(#observer_fields,)* )) diff --git a/crates/model/Cargo.toml b/crates/model/Cargo.toml index 4951db1ad..64bb703e5 100644 --- a/crates/model/Cargo.toml +++ b/crates/model/Cargo.toml @@ -24,7 +24,7 @@ quent-build-info = { path = "../build-info" } quent-model-macros = { path = "../model-macros" } quent-events = { path = "../events" } quent-io = { path = "../io" } -quent-instrumentation = { path = "../instrumentation" } +quent-instrumentation = { path = "../instrumentation", features = ["io-callback"] } quent-time = { path = "../time" } tokio = { workspace = true, features = ["macros"] } diff --git a/crates/model/src/lib.rs b/crates/model/src/lib.rs index d09c7ac13..c95ea1a8a 100644 --- a/crates/model/src/lib.rs +++ b/crates/model/src/lib.rs @@ -85,7 +85,9 @@ pub use quent_dynamic_attributes as attributes; #[doc(hidden)] pub use quent_events as events; pub use quent_events::{EntityEvent, Event}; -pub use quent_instrumentation::{ContextInner, ObserverInner as Observer, write_sidecar}; +pub use quent_instrumentation::{ + ContextExporter, ContextInner, EventCallback, Noop, ObserverInner as Observer, write_sidecar, +}; pub use quent_io as io; pub use quent_time::timestamp; #[cfg(feature = "serde")] diff --git a/domains/query_engine/tests/fixed/Cargo.toml b/domains/query_engine/tests/fixed/Cargo.toml index 0736e9d35..ab3b782e5 100644 --- a/domains/query_engine/tests/fixed/Cargo.toml +++ b/domains/query_engine/tests/fixed/Cargo.toml @@ -9,7 +9,7 @@ clap = { version = "4.5.57", features = ["derive", "env"] } quent-time = { path = "../../../../crates/time", features = ["__test-clock-override"] } quent-model = { path = "../../../../crates/model" } quent-dynamic-attributes = { path = "../../../../crates/dynamic-attributes" } -quent-io = { path = "../../../../crates/io", features = ["clap", "callback"] } +quent-io = { path = "../../../../crates/io", features = ["clap"] } quent-query-engine-model = { path = "../../model" } quent-simulator-instrumentation = { path = "../../../../examples/simulator/instrumentation" } quent-stdlib = { path = "../../../../crates/stdlib" } diff --git a/domains/query_engine/tests/fixed/src/main.rs b/domains/query_engine/tests/fixed/src/main.rs index d3759cb4b..6e520cbd2 100644 --- a/domains/query_engine/tests/fixed/src/main.rs +++ b/domains/query_engine/tests/fixed/src/main.rs @@ -19,7 +19,10 @@ struct Args { fn main() -> Result<(), Box> { let args = Args::parse(); - let ctx = SimulatorContext::try_new(args.exporter.into_options())?; + let ctx = match args.exporter.into_options() { + Some(provider) => SimulatorContext::try_new(provider)?, + None => SimulatorContext::try_new(quent_model::Noop)?, + }; emit(&ctx); Ok(()) } diff --git a/domains/query_engine/tests/fixed/tests/data_flow.rs b/domains/query_engine/tests/fixed/tests/data_flow.rs index 6ff1ad55d..fd547b4ef 100644 --- a/domains/query_engine/tests/fixed/tests/data_flow.rs +++ b/domains/query_engine/tests/fixed/tests/data_flow.rs @@ -10,13 +10,13 @@ //! `sending` state from +500ms. Computing holds 256 bytes of the worker's //! "memory" resource; allocating and sending hold no memory. -use quent_io::{EventCallback, ExporterOptions}; +use quent_model::EventCallback; use quent_query_engine_analyzer::ui::UiAnalyzer; use quent_query_engine_fixed as fixed; use quent_query_engine_ui::DataFlowTimelineBinned; use quent_query_engine_ui::QueryFilter; use quent_simulator_analyzer::SimulatorUiAnalyzer; -use quent_simulator_instrumentation::{SimulatorContext, test_utils::events_from_recorded}; +use quent_simulator_instrumentation::SimulatorContext; use quent_ui::timeline::{categorical::CategoricalTimelineRequest, request::TimelineConfig}; use std::sync::{Arc, Mutex}; @@ -26,15 +26,15 @@ fn fixed_analyzer() -> SimulatorUiAnalyzer { let recorded = Arc::new(Mutex::new(Vec::new())); { let captured = Arc::clone(&recorded); - let ctx = SimulatorContext::try_new(Some(ExporterOptions::Callback(EventCallback::new( - move |event| captured.lock().unwrap().push(event), - )))) + let ctx = SimulatorContext::try_new(EventCallback::new(move |event| { + captured.lock().unwrap().push(event); + })) .unwrap(); fixed::emit(&ctx); // ctx dropped here, flushing all events to the callback. } - let events = events_from_recorded(std::mem::take(&mut *recorded.lock().unwrap())); + let events = std::mem::take(&mut *recorded.lock().unwrap()); SimulatorUiAnalyzer::try_new(fixed::ENGINE, events.into_iter()).unwrap() } diff --git a/domains/query_engine/tests/fixed/tests/list_entities.rs b/domains/query_engine/tests/fixed/tests/list_entities.rs index b773036f2..0bb9d85ba 100644 --- a/domains/query_engine/tests/fixed/tests/list_entities.rs +++ b/domains/query_engine/tests/fixed/tests/list_entities.rs @@ -11,12 +11,12 @@ //! (0.75s). `MEMORY_W0` is used by 8 tasks, `MEMORY_W1` by 4. Because the spans //! are equal, results are ordered by the UUID tiebreaker. -use quent_io::{EventCallback, ExporterOptions}; +use quent_model::EventCallback; use quent_query_engine_analyzer::ui::UiAnalyzer; use quent_query_engine_fixed as fixed; use quent_query_engine_ui::{OperatorFilter, QueryFilter}; use quent_simulator_analyzer::SimulatorUiAnalyzer; -use quent_simulator_instrumentation::{SimulatorContext, test_utils::events_from_recorded}; +use quent_simulator_instrumentation::SimulatorContext; use quent_ui::entities::request::{ EntityListEntry, EntityListFilter, EntityListRequest, EntityScope, EntitySortKey, Sort, SortDir, TimeWindow, @@ -66,15 +66,15 @@ fn fixed_analyzer() -> SimulatorUiAnalyzer { let recorded = Arc::new(Mutex::new(Vec::new())); { let captured = Arc::clone(&recorded); - let ctx = SimulatorContext::try_new(Some(ExporterOptions::Callback(EventCallback::new( - move |event| captured.lock().unwrap().push(event), - )))) + let ctx = SimulatorContext::try_new(EventCallback::new(move |event| { + captured.lock().unwrap().push(event); + })) .unwrap(); fixed::emit(&ctx); // ctx dropped here, flushing all events to the callback. } - let events = events_from_recorded(std::mem::take(&mut *recorded.lock().unwrap())); + let events = std::mem::take(&mut *recorded.lock().unwrap()); SimulatorUiAnalyzer::try_new(fixed::ENGINE, events.into_iter()).unwrap() } diff --git a/examples/readme/src/main.rs b/examples/readme/src/main.rs index cd7c61660..a2b224bc1 100644 --- a/examples/readme/src/main.rs +++ b/examples/readme/src/main.rs @@ -12,7 +12,7 @@ fn main() -> Result<(), Box> { root.clone(), ), ); - let context = AppContext::try_new(Some(exporter))?; + let context = AppContext::try_new(exporter)?; // The context generates its own id and writes events under `root//`. // Reuse it as the root resource group id. diff --git a/examples/simulator/application/src/main.rs b/examples/simulator/application/src/main.rs index e84c81c3d..b67b7f05a 100644 --- a/examples/simulator/application/src/main.rs +++ b/examples/simulator/application/src/main.rs @@ -1129,7 +1129,10 @@ fn main() -> Result<(), Box> { let mut engine = Engine::new(); - let context = SimulatorContext::try_new(args.exporter.into_options())?; + let context = match args.exporter.into_options() { + Some(provider) => SimulatorContext::try_new(provider)?, + None => SimulatorContext::try_new(quent_model::Noop)?, + }; engine.spawn(&context, args.num_workers, args.num_threads); diff --git a/examples/simulator/instrumentation/Cargo.toml b/examples/simulator/instrumentation/Cargo.toml index d4e480e90..2defcbd1e 100644 --- a/examples/simulator/instrumentation/Cargo.toml +++ b/examples/simulator/instrumentation/Cargo.toml @@ -9,7 +9,6 @@ collector = ["quent-model/collector"] [dependencies] quent-model = { path = "../../../crates/model" } -quent-io-callback = { path = "../../../crates/io/callback" } quent-query-engine-model = { path = "../../../domains/query_engine/model" } quent-stdlib = { path = "../../../crates/stdlib" } serde.workspace = true diff --git a/examples/simulator/instrumentation/src/lib.rs b/examples/simulator/instrumentation/src/lib.rs index 90d0d6a81..25b71834c 100644 --- a/examples/simulator/instrumentation/src/lib.rs +++ b/examples/simulator/instrumentation/src/lib.rs @@ -40,5 +40,3 @@ model! { } instrumentation!(Simulator); - -pub mod test_utils; diff --git a/examples/simulator/instrumentation/src/test_utils.rs b/examples/simulator/instrumentation/src/test_utils.rs deleted file mode 100644 index b988c1ee2..000000000 --- a/examples/simulator/instrumentation/src/test_utils.rs +++ /dev/null @@ -1,63 +0,0 @@ -// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. -// SPDX-License-Identifier: Apache-2.0 - -//! In-memory test helpers for the `Simulator` model. - -use quent_io_callback::RecordedEvent; -use quent_model::Event; - -use crate::{ - NetworkEvent, SimulatorEvent, ThreadPoolEvent, channel, engine, memory, operator, plan, port, - processor, query, query_group, task, worker, -}; - -/// Reconstruct the `Simulator` event stream from events captured in memory by a -/// callback exporter, e.g. for feeding an analyzer in tests. -/// -/// Events whose entity is not part of the model are skipped. -pub fn events_from_recorded( - recorded: impl IntoIterator, -) -> Vec> { - // Try to downcast the type-erased event to each concrete `Event`; the - // matching one lifts into the umbrella `SimulatorEvent`. `downcast` hands the - // box back on a miss, so attempts thread through it. - macro_rules! rebuild { - ($($ty:ty),+ $(,)?) => { - |rec: RecordedEvent| { - let mut any = rec.event; - $( - any = match any.downcast::>() { - Ok(event) => { - return Some(Event::new( - event.id, - event.timestamp, - SimulatorEvent::from(event.data), - )); - } - Err(any) => any, - }; - )+ - let _ = any; - None - } - }; - } - recorded - .into_iter() - .filter_map(rebuild!( - engine::EngineEvent, - worker::WorkerEvent, - query_group::QueryGroupEvent, - query::QueryEvent, - plan::PlanEvent, - operator::OperatorEvent, - port::PortEvent, - task::TaskEvent, - ThreadPoolEvent, - NetworkEvent, - memory::MemoryEvent, - processor::ProcessorEvent, - channel::ChannelEvent, - )) - .collect() -} diff --git a/examples/simulator/server/src/main.rs b/examples/simulator/server/src/main.rs index 1545f4c51..050c8a3a5 100644 --- a/examples/simulator/server/src/main.rs +++ b/examples/simulator/server/src/main.rs @@ -100,8 +100,7 @@ async fn main() -> Result<(), Box> { let collector = async { collector_service::(move |id| { - SimulatorContext::try_with_id(id, Some(exporter_kind.clone())) - .map_err(|e| e.to_string()) + SimulatorContext::try_with_id(id, exporter_kind.clone()).map_err(|e| e.to_string()) })? .serve(collector_addr) .await diff --git a/integrations/nvtx/example/src/lib.rs b/integrations/nvtx/example/src/lib.rs index c74aeac60..efff3a728 100644 --- a/integrations/nvtx/example/src/lib.rs +++ b/integrations/nvtx/example/src/lib.rs @@ -27,7 +27,7 @@ use uuid::Uuid; /// drops the pipeline to flush. pub fn run_capture( session: Uuid, - exporter: EventCallback, + exporter: EventCallback, ) -> Result<(), Box> { run_capture_n_threads(1, session, exporter) } @@ -40,11 +40,11 @@ pub fn run_capture( pub fn run_capture_n_threads( n: usize, session: Uuid, - exporter: EventCallback, + exporter: EventCallback, ) -> Result<(), Box> { let context = ContextInner::try_new(session)?; let pipeline = - context.block_on(async { context.observer::(exporter).await })?; + context.block_on(async { context.observer::(&exporter).await })?; // Forward each captured event into the pipeline, before the first NVTX call. let sender = pipeline.sender(); diff --git a/integrations/nvtx/example/src/main.rs b/integrations/nvtx/example/src/main.rs index 9e2b45687..4844f39f9 100644 --- a/integrations/nvtx/example/src/main.rs +++ b/integrations/nvtx/example/src/main.rs @@ -8,15 +8,13 @@ //! ``` use nvtx_bridge::NvtxEventEntity; -use quent_instrumentation::{Event, EventCallback}; +use quent_instrumentation::EventCallback; use uuid::Uuid; fn main() -> Result<(), Box> { // The app's exporter: debug-print each captured NVTX event. - let printer = EventCallback::new(|recorded| { - if let Some(event) = recorded.event.downcast_ref::>() { - println!("[{} @ {}] {:?}", event.id, event.timestamp, event.data.0); - } + let printer = EventCallback::::new(|event| { + println!("[{} @ {}] {:?}", event.id, event.timestamp, event.data.0); }); nvtx_example::run_capture(Uuid::now_v7(), printer) diff --git a/integrations/nvtx/example/tests/capture.rs b/integrations/nvtx/example/tests/capture.rs index 6ca56ebff..f5af6e0b5 100644 --- a/integrations/nvtx/example/tests/capture.rs +++ b/integrations/nvtx/example/tests/capture.rs @@ -12,7 +12,7 @@ use std::sync::{Arc, Mutex}; use nvtx_bridge::NvtxEventEntity; use nvtx_events::NvtxEvent; -use quent_instrumentation::{Event, EventCallback}; +use quent_instrumentation::EventCallback; use uuid::Uuid; /// The variant name of an [`NvtxEvent`], for coverage assertions. @@ -39,10 +39,8 @@ fn captures_core_nvtx_kinds() { let collected: Arc>> = Arc::new(Mutex::new(Vec::new())); let sink = { let collected = Arc::clone(&collected); - EventCallback::new(move |recorded| { - if let Some(event) = recorded.event.downcast_ref::>() { - collected.lock().unwrap().push(event.data.0.clone()); - } + EventCallback::::new(move |event| { + collected.lock().unwrap().push(event.data.0.clone()); }) }; diff --git a/integrations/nvtx/example/tests/thread_id.rs b/integrations/nvtx/example/tests/thread_id.rs index 519f02d01..24a8c5436 100644 --- a/integrations/nvtx/example/tests/thread_id.rs +++ b/integrations/nvtx/example/tests/thread_id.rs @@ -19,7 +19,7 @@ use std::sync::{Arc, Mutex}; use nvtx_bridge::NvtxEventEntity; use nvtx_events::NvtxEvent; -use quent_instrumentation::{Event, EventCallback}; +use quent_instrumentation::EventCallback; use uuid::Uuid; const N_THREADS: usize = 4; @@ -29,10 +29,8 @@ fn pushpop_four_threads_get_distinct_ids() { let collected: Arc>> = Arc::new(Mutex::new(Vec::new())); let sink = { let collected = Arc::clone(&collected); - EventCallback::new(move |recorded| { - if let Some(event) = recorded.event.downcast_ref::>() { - collected.lock().unwrap().push(event.data.0.clone()); - } + EventCallback::::new(move |event| { + collected.lock().unwrap().push(event.data.0.clone()); }) }; From 484292e7e016f4e4395626cf272b2822f5741883 Mon Sep 17 00:00:00 2001 From: Johan Peltenburg Date: Wed, 5 Aug 2026 07:41:56 +0200 Subject: [PATCH 2/2] Remove superfluous test --- crates/instrumentation/src/noop.rs | 18 ------------------ 1 file changed, 18 deletions(-) diff --git a/crates/instrumentation/src/noop.rs b/crates/instrumentation/src/noop.rs index 32a1262e0..e6ad67f52 100644 --- a/crates/instrumentation/src/noop.rs +++ b/crates/instrumentation/src/noop.rs @@ -45,21 +45,3 @@ impl ContextExporter for Noop { fn prepare_context(&self, _context_id: Uuid, _model: ModelInfo) {} } - -#[cfg(test)] -mod tests { - use super::*; - - #[tokio::test] - async fn provides_an_exporter_for_any_event_type() { - let mut exporter: Box> = - Noop.create_exporter(Uuid::nil()).await.unwrap(); - - exporter - .push(Event::new(Uuid::nil(), 0, "ignored".to_owned())) - .await - .unwrap(); - exporter.shutdown().await.unwrap(); - assert!(Noop.is_noop()); - } -}