diff --git a/Cargo.lock b/Cargo.lock index 45085c1b6..739daa995 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2466,10 +2466,14 @@ dependencies = [ "quent-attributes", "quent-io", "quent-model", + "quent-query-engine-analyzer", "quent-query-engine-model", + "quent-query-engine-ui", + "quent-simulator-analyzer", "quent-simulator-instrumentation", "quent-stdlib", "quent-time", + "quent-ui", "uuid", ] @@ -2631,6 +2635,7 @@ 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/ui/src/entities/mod.rs b/crates/ui/src/entities/mod.rs new file mode 100644 index 000000000..ccc169e80 --- /dev/null +++ b/crates/ui/src/entities/mod.rs @@ -0,0 +1,6 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Requests and responses for entity-list queries. +pub mod request; +pub mod response; diff --git a/crates/ui/src/entities/request.rs b/crates/ui/src/entities/request.rs new file mode 100644 index 000000000..4a0cea2b0 --- /dev/null +++ b/crates/ui/src/entities/request.rs @@ -0,0 +1,137 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +use std::collections::HashSet; + +use quent_analyzer::{AnalyzerResult, Model, resource::tree::ResourceTreeNode}; +use quent_time::{TimeError, TimeSec, TimeUnixNanoSec, span::SpanUnixNanoSec, to_nanosecs}; +use serde::{Deserialize, Serialize}; +use ts_rs::TS; +use uuid::Uuid; + +use crate::paginate::PageParams; + +/// Restricts returned entities to appear in a certain scope. +#[derive(TS, Debug, Clone, Serialize, Deserialize)] +pub enum EntityScope { + /// Only return entities that use this resource. + Resource { resource_id: Uuid }, + /// Only return entities that use any resource within this group. + ResourceGroup { + resource_group_id: Uuid, + resource_type_name: String, + }, +} + +impl EntityScope { + /// Resolve the scope to the resource IDs it covers: the single resource, or + /// the leaf resources of the group that have the requested type. + pub fn resolve(&self, model: &impl Model) -> AnalyzerResult> { + match self { + EntityScope::Resource { resource_id } => { + model.resource(*resource_id)?; + Ok([*resource_id].into_iter().collect()) + } + EntityScope::ResourceGroup { + resource_group_id, + resource_type_name, + } => { + let tree = ResourceTreeNode::try_new(model, *resource_group_id)?; + Ok(tree + .iter_leaf_ids() + .filter(|&id| { + model + .resource(id) + .is_ok_and(|r| r.type_name() == resource_type_name) + }) + .collect()) + } + } + } +} + +/// A time window, resolved against an epoch supplied at conversion. +#[derive(TS, Debug, Clone, Copy, Serialize, Deserialize)] +pub struct TimeWindow { + /// The start time of the window in seconds. + pub start: TimeSec, + /// The end time of the window in seconds. + pub end: TimeSec, +} + +impl TimeWindow { + /// Resolve to an absolute span by offsetting from `epoch`. + pub fn try_into_span(self, epoch: TimeUnixNanoSec) -> Result { + SpanUnixNanoSec::try_new( + epoch + to_nanosecs(self.start), + epoch + to_nanosecs(self.end), + ) + } +} + +/// Entity filters. +/// +/// Every field that is set must match, `None` fields do not filter. +#[derive(TS, Debug, Clone, Default, Serialize, Deserialize)] +pub struct EntityListFilter { + /// Restrict resulting entities to be in this scope. + pub scope: Option, + /// Restrict resulting entities to be of this type. + pub entity_type_name: Option, + /// Keep only entities with resource usages longer than this threshold. + /// + /// N.B. only Fsm-type entities can have usages. + pub min_usage_s: Option, +} + +/// The key entities are sorted by. +#[derive(TS, Debug, Clone, Copy, Serialize, Deserialize)] +pub enum EntitySortKey { + /// The longest single resource-usage span within the window. + UsageDuration, +} + +/// The direction to sort in. +#[derive(TS, Debug, Clone, Copy, Serialize, Deserialize)] +pub enum SortDir { + /// Sort in ascending order. + Asc, + /// Sort in descending order. + Desc, +} + +/// Sorting parameters. +#[derive(TS, Debug, Clone, Copy, Serialize, Deserialize)] +pub struct Sort { + /// The key to sort on. + pub key: EntitySortKey, + /// The direction to sort in. + pub dir: SortDir, +} + +/// A single entity-list query. +#[derive(TS, Debug, Clone, Serialize, Deserialize)] +pub struct EntityListEntry { + /// The window of time entities must fall in. + pub window: TimeWindow, + /// Filter parameters. + pub filter: EntityListFilter, + /// Sort parameters. + pub sort: Sort, + /// Pagination parameters. + /// + /// When this is not set, return the full list. Depending on the dataset + /// size and other parameters, this may result in a large volume of data and + /// should be used with care. + pub page: Option, + /// Per-query application-specific parameters. + pub application: EntryParams, +} + +/// Parameters for listing entities. +#[derive(TS, Debug, Clone, Serialize, Deserialize)] +pub struct EntityListRequest { + pub entry: EntityListEntry, + /// Global application parameters shared by the query. + pub app_params: GlobalParams, +} diff --git a/crates/ui/src/entities/response.rs b/crates/ui/src/entities/response.rs new file mode 100644 index 000000000..57d318a6e --- /dev/null +++ b/crates/ui/src/entities/response.rs @@ -0,0 +1,17 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +use serde::Serialize; +use ts_rs::TS; + +use crate::FiniteStateMachine; + +/// A ranked, paged list of entities. +#[derive(TS, Debug, Clone, Serialize)] +pub struct EntityListResponse { + // TODO(johanpel): generalize to other entity types, but only FSMs are + // represented today. + pub items: Vec, + /// The count of entities matching the filter before paging. + pub total: u32, +} diff --git a/crates/ui/src/lib.rs b/crates/ui/src/lib.rs index bb56e8bb1..3c2acebb0 100644 --- a/crates/ui/src/lib.rs +++ b/crates/ui/src/lib.rs @@ -1,12 +1,16 @@ // SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 +use std::collections::HashMap; + use quent_analyzer::{self as a, AnalyzerResult, Entity, Model, resource::tree::ResourceTreeNode}; use quent_time::{TimeSec, TimeUnixNanoSec, try_to_secs_relative}; use serde::Serialize; use ts_rs::TS; use uuid::Uuid; +pub mod entities; +pub mod paginate; pub mod quantity; pub mod timeline; @@ -249,4 +253,54 @@ impl FiniteStateMachine { .collect::, _>>()?, }) } + + /// Build from any application FSM via the [`FsmUsages`](a::fsm::FsmUsages) + /// interface. + /// + /// Usages are grouped onto their state's transition by state name, which is + /// unique within an FSM. + pub fn try_from_fsm<'a, F>( + fsm: &'a F, + epoch: TimeUnixNanoSec, + ) -> Result + where + F: a::fsm::FsmUsages<'a>, + { + use a::fsm::Transition; + use a::resource::Usage; + use quent_time::Timestamp; + + let mut usages_by_state: HashMap> = HashMap::new(); + for (state_name, usage) in fsm.usages_with_state_names() { + usages_by_state + .entry(state_name.to_owned()) + .or_default() + .push(FsmUsage { + resource: usage.resource_id(), + capacities: usage + .capacities() + .map(|c| (c.name.to_string(), c.value)) + .collect(), + }); + } + + // 0..=len covers every transition including the exit transition. + let transitions = (0..=fsm.len()) + .filter_map(|i| fsm.transition(i)) + .map(|t| { + Ok(FsmTransition { + name: t.name().to_owned(), + usages: usages_by_state.remove(t.name()).unwrap_or_default(), + timestamp: try_to_secs_relative(t.timestamp(), epoch)?, + }) + }) + .collect::, quent_time::TimeError>>()?; + + Ok(Self { + id: fsm.id(), + type_name: fsm.type_name().to_owned(), + instance_name: fsm.instance_name().to_owned(), + transitions, + }) + } } diff --git a/crates/ui/src/paginate.rs b/crates/ui/src/paginate.rs new file mode 100644 index 000000000..1e22ba0d0 --- /dev/null +++ b/crates/ui/src/paginate.rs @@ -0,0 +1,16 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Pagination parameters shared across list endpoints. + +use serde::{Deserialize, Serialize}; +use ts_rs::TS; + +/// Parameters for paginated lists. +#[derive(TS, Debug, Clone, Copy, Serialize, Deserialize)] +pub struct PageParams { + /// The maximum size of a page. + pub max: u32, + /// The zero-based index of the requested page. + pub page: u32, +} diff --git a/domains/query_engine/analyzer/src/entities.rs b/domains/query_engine/analyzer/src/entities.rs new file mode 100644 index 000000000..f351cba34 --- /dev/null +++ b/domains/query_engine/analyzer/src/entities.rs @@ -0,0 +1,141 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Generic entity-list query over any application FSM type. + +use std::collections::HashSet; + +use quent_analyzer::{ + AnalyzerResult, + fsm::{FsmUsages, collection::FsmCollection}, + resource::Usage, +}; +use quent_time::{TimeNanoSec, TimeUnixNanoSec, span::SpanUnixNanoSec, to_nanosecs}; +use quent_ui::{ + FiniteStateMachine, + entities::{ + request::{EntityListFilter, EntitySortKey, Sort, SortDir}, + response::EntityListResponse, + }, + paginate::PageParams, +}; +use uuid::Uuid; + +/// A single entity-list query with its scope already resolved to resource IDs. +pub struct ListQuery<'a> { + pub scope: Option<&'a HashSet>, + pub window: SpanUnixNanoSec, + pub filter: &'a EntityListFilter, + pub sort: Sort, + pub page: Option, + pub epoch: TimeUnixNanoSec, +} + +/// List the FSMs matching the scope, window, and filters, ranked and paged. +/// +/// `keep` is an extra application predicate for filters outside the generic +/// contract, e.g. by operator. +pub fn list_entities( + model: &M, + keep: P, + query: ListQuery<'_>, +) -> AnalyzerResult +where + M: FsmCollection, + M::Fsm: for<'a> FsmUsages<'a>, + P: Fn(&M::Fsm) -> bool, +{ + let ranked = model + .fsms() + .filter(|f| keep(f)) + .filter_map(|f| entry_matches(f, query.scope, query.window, query.filter).map(|m| (f, m))) + .collect(); + finalize(ranked, query.sort, query.page, query.epoch) +} + +/// The ranking metric if the FSM passes the filters, else `None`. +fn entry_matches<'a, F>( + fsm: &'a F, + scope: Option<&HashSet>, + window: SpanUnixNanoSec, + filter: &EntityListFilter, +) -> Option +where + F: FsmUsages<'a>, +{ + if filter + .entity_type_name + .as_deref() + .is_some_and(|name| fsm.type_name() != name) + { + return None; + } + let metric = usage_metric(fsm, scope, window)?; + let min_usage = filter.min_usage_s.map(to_nanosecs); + min_usage.is_none_or(|t| metric >= t).then_some(metric) +} + +/// Sort the scored candidates, slice the page, and convert to UI FSMs. +fn finalize<'a, F>( + mut ranked: Vec<(&'a F, TimeNanoSec)>, + sort: Sort, + page: Option, + epoch: TimeUnixNanoSec, +) -> AnalyzerResult +where + F: FsmUsages<'a>, +{ + ranked.sort_by(|(fa, ma), (fb, mb)| { + let by_key = match sort.key { + EntitySortKey::UsageDuration => ma.cmp(mb), + }; + let by_key = match sort.dir { + SortDir::Asc => by_key, + SortDir::Desc => by_key.reverse(), + }; + by_key.then_with(|| fa.id().cmp(&fb.id())) + }); + + let total = ranked.len() as u32; + + let page_iter: Box> = match page { + Some(p) => Box::new( + ranked + .into_iter() + // Saturate so an out-of-range page skips everything (empty page) + // instead of overflowing usize on a 32-bit target. + .skip(p.page.saturating_mul(p.max) as usize) + .take(p.max as usize), + ), + None => Box::new(ranked.into_iter()), + }; + + let items = page_iter + .map(|(f, _)| FiniteStateMachine::try_from_fsm(f, epoch)) + .collect::, _>>()?; + + Ok(EntityListResponse { items, total }) +} + +/// The longest single usage span within the window on a scope resource, or any +/// resource when `scope` is `None`. +fn usage_metric<'a, F>( + fsm: &'a F, + scope: Option<&HashSet>, + window: SpanUnixNanoSec, +) -> Option +where + F: FsmUsages<'a>, +{ + let longest = fsm + .usages_with_state_names() + .filter(|(_, u)| scope.is_none_or(|s| s.contains(&u.resource_id()))) + .filter_map(|(_, u)| u.span().intersection(&window)) + .map(|s| s.duration()) + .max(); + + match scope { + Some(_) => longest, + None => Some(longest.unwrap_or(0)), + } +} diff --git a/domains/query_engine/analyzer/src/lib.rs b/domains/query_engine/analyzer/src/lib.rs index b39ca2d1a..1520b96f2 100644 --- a/domains/query_engine/analyzer/src/lib.rs +++ b/domains/query_engine/analyzer/src/lib.rs @@ -46,6 +46,7 @@ pub mod model; pub mod view; // UI related mods +pub mod entities; pub mod ui; pub trait QueryEngineModel: Model { diff --git a/domains/query_engine/analyzer/src/ui.rs b/domains/query_engine/analyzer/src/ui.rs index 64c515a25..0e28316bd 100644 --- a/domains/query_engine/analyzer/src/ui.rs +++ b/domains/query_engine/analyzer/src/ui.rs @@ -8,11 +8,14 @@ use quent_analyzer::AnalyzerResult; use quent_events::Event; use quent_model::io::ImporterResult; use quent_query_engine_ui as ui; -use quent_ui::timeline::{ - request::{BulkChunkedTimelineRequest, BulkTimelineRequest, SingleTimelineRequest}, - response::{ - BulkChunkedTimelinesResponse, BulkTimelinesResponse, BulkTimelinesResponseEntry, - SingleTimelineResponse, +use quent_ui::{ + entities::{request::EntityListRequest, response::EntityListResponse}, + timeline::{ + request::{BulkChunkedTimelineRequest, BulkTimelineRequest, SingleTimelineRequest}, + response::{ + BulkChunkedTimelinesResponse, BulkTimelinesResponse, BulkTimelinesResponseEntry, + SingleTimelineResponse, + }, }, }; use uuid::Uuid; @@ -61,6 +64,13 @@ pub trait UiAnalyzer { request: SingleTimelineRequest, ) -> AnalyzerResult; + /// List the entities matching a scope, window, and filter, ranked by the + /// requested sort key and sliced to the requested page. + fn list_entities( + &self, + request: EntityListRequest, + ) -> AnalyzerResult; + /// Return a set of resource timelines in bulk. fn bulk_resource_timeline( &self, diff --git a/domains/query_engine/server/src/timeline_cache.rs b/domains/query_engine/server/src/timeline_cache.rs index 1b66ad577..85f37ee44 100644 --- a/domains/query_engine/server/src/timeline_cache.rs +++ b/domains/query_engine/server/src/timeline_cache.rs @@ -950,6 +950,13 @@ mod tests { &self.model } + fn list_entities( + &self, + _request: quent_ui::entities::request::EntityListRequest, + ) -> AnalyzerResult { + unimplemented!("not needed by timeline cache tests") + } + fn single_resource_timeline( &self, request: SingleTimelineRequest, diff --git a/domains/query_engine/server/src/ui.rs b/domains/query_engine/server/src/ui.rs index 70126636a..119155537 100644 --- a/domains/query_engine/server/src/ui.rs +++ b/domains/query_engine/server/src/ui.rs @@ -10,6 +10,7 @@ use axum::{ use quent_analyzer::AnalyzerResult; use quent_query_engine_analyzer::{QueryEngineModel, query_group::QueryGroup, ui::UiAnalyzer}; use quent_query_engine_ui as ui; +use quent_ui::entities::{request::EntityListRequest, response::EntityListResponse}; use quent_ui::timeline::{ request::{BulkTimelineRequest, SingleTimelineRequest}, response::{BulkTimelinesResponse, SingleTimelineResponse}, @@ -261,6 +262,32 @@ where )) } +/// List the entities of a resource or resource group, ranked and paged. +#[cfg_attr(feature = "swagger", utoipa::path( + post, + path = "/api/engines/{engine_id}/entities", + tag = "entities", + params( + ("engine_id" = Uuid, Path, description = "The engine ID") + ), + request_body = Object, + responses( + (status = 200, description = "Ranked, paged list of entities", body = Object) + ) +))] +#[tracing::instrument(skip_all, err)] +async fn entities( + State(state): State>, + Path(engine_id): Path, + Json(request): Json>, +) -> ServerResult> +where + A: UiAnalyzer + Send + Sync + 'static, +{ + let analyzer = state.analyzers.get(engine_id).await?; + Ok(Json(analyzer.list_entities(request)?)) +} + #[cfg(feature = "swagger")] #[derive(utoipa::OpenApi)] #[openapi( @@ -272,10 +299,12 @@ where query, single_timeline, bulk_timelines, + entities, ), tags( (name = "engines", description = "Engine, query group, and query management"), (name = "timelines", description = "Resource timeline data"), + (name = "entities", description = "Entity list queries"), ) )] pub(crate) struct ApiDoc; @@ -296,5 +325,6 @@ where .route("/{engine_id}/query/{query_id}", get(query)) .route("/{engine_id}/timeline/single", post(single_timeline)) .route("/{engine_id}/timeline/bulk", post(bulk_timelines)) + .route("/{engine_id}/entities", post(entities)) .with_state(state) } diff --git a/domains/query_engine/tests/fixed/Cargo.toml b/domains/query_engine/tests/fixed/Cargo.toml index 3ae6a02f6..a677a9c96 100644 --- a/domains/query_engine/tests/fixed/Cargo.toml +++ b/domains/query_engine/tests/fixed/Cargo.toml @@ -9,8 +9,14 @@ clap = { version = "4.5.57", features = ["derive", "env"] } quent-time = { path = "../../../../crates/time", features = ["__test-clock-override"] } quent-model = { path = "../../../../crates/model" } quent-attributes = { path = "../../../../crates/attributes" } -quent-io = { path = "../../../../crates/io", features = ["clap"] } +quent-io = { path = "../../../../crates/io", features = ["clap", "callback"] } quent-query-engine-model = { path = "../../model" } quent-simulator-instrumentation = { path = "../../../../examples/simulator/instrumentation" } quent-stdlib = { path = "../../../../crates/stdlib" } uuid = { workspace = true } + +[dev-dependencies] +quent-simulator-analyzer = { path = "../../../../examples/simulator/analyzer" } +quent-query-engine-analyzer = { path = "../../analyzer" } +quent-query-engine-ui = { path = "../../ui" } +quent-ui = { path = "../../../../crates/ui" } diff --git a/domains/query_engine/tests/fixed/tests/list_entities.rs b/domains/query_engine/tests/fixed/tests/list_entities.rs new file mode 100644 index 000000000..5aa1deb7d --- /dev/null +++ b/domains/query_engine/tests/fixed/tests/list_entities.rs @@ -0,0 +1,229 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Functional test of `UiAnalyzer::list_entities` over the fixed scenario. +//! +//! Captures the fixed 7-second telemetry in memory via a callback exporter, +//! builds a `SimulatorUiAnalyzer` from it, and asserts the entity-list query +//! against the scenario's known tasks and their resource usage. +//! +//! Ground truth: every task holds its memory for the `computing→exit` span +//! (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_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_ui::entities::request::{ + EntityListEntry, EntityListFilter, EntityListRequest, EntityScope, EntitySortKey, Sort, + SortDir, TimeWindow, +}; +use quent_ui::entities::response::EntityListResponse; +use quent_ui::paginate::PageParams; +use std::sync::{Arc, Mutex}; +use uuid::Uuid; + +// Tasks using MEMORY_W0, in ascending UUID order (the tiebreaker). +const MEMORY_W0_TASKS: [Uuid; 8] = [ + fixed::TASK_0, + fixed::TASK_1, + fixed::TASK_4, + fixed::TASK_5, + fixed::TASK_8, + fixed::TASK_9, + fixed::TASK_10, + fixed::TASK_11, +]; + +// Tasks using MEMORY_W1, in ascending UUID order. +const MEMORY_W1_TASKS: [Uuid; 4] = [fixed::TASK_2, fixed::TASK_3, fixed::TASK_6, fixed::TASK_7]; + +// All 12 tasks ranked by longest usage, descending. TASK_6 and TASK_7 sort +// last: their `computing` is cut short by a `sending` transition, so their +// longest single usage span is 0.5s (the send) versus 0.75s for every other +// task. The remaining ten tie at 0.75s and fall back to ascending UUID order. +const ALL_TASKS_RANKED: [Uuid; 12] = [ + fixed::TASK_0, + fixed::TASK_1, + fixed::TASK_2, + fixed::TASK_3, + fixed::TASK_4, + fixed::TASK_5, + fixed::TASK_8, + fixed::TASK_9, + fixed::TASK_10, + fixed::TASK_11, + fixed::TASK_6, + fixed::TASK_7, +]; + +/// Emit the fixed scenario into memory via a callback exporter and build an +/// analyzer from the captured events. +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), + )))) + .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())); + SimulatorUiAnalyzer::try_new(fixed::ENGINE, events.into_iter()).unwrap() +} + +/// An entity-list entry over the whole query window, ranked by usage duration. +fn entry( + scope: Option, + min_usage_s: Option, + page: Option, + operator_id: Option, +) -> EntityListEntry { + EntityListEntry { + window: TimeWindow { + start: 0.0, + end: 7.0, + }, + filter: EntityListFilter { + scope, + entity_type_name: None, + min_usage_s, + }, + sort: Sort { + key: EntitySortKey::UsageDuration, + dir: SortDir::Desc, + }, + page, + application: OperatorFilter { operator_id }, + } +} + +fn request_scoped( + scope: Option, + min_usage_s: Option, + page: Option, +) -> EntityListRequest { + EntityListRequest { + entry: entry(scope, min_usage_s, page, None), + app_params: QueryFilter { + query_id: fixed::QUERY, + }, + } +} + +/// A request scoped to a single resource. +fn request( + resource_id: Uuid, + min_usage_s: Option, + page: Option, +) -> EntityListRequest { + request_scoped( + Some(EntityScope::Resource { resource_id }), + min_usage_s, + page, + ) +} + +fn ids(resp: &EntityListResponse) -> Vec { + resp.items.iter().map(|fsm| fsm.id).collect() +} + +#[test] +fn lists_all_tasks_on_a_resource_ranked_by_uuid_tiebreak() { + let analyzer = fixed_analyzer(); + let resp = analyzer + .list_entities(request(fixed::MEMORY_W0, None, None)) + .unwrap(); + + assert_eq!(resp.total, 8); + assert_eq!(ids(&resp), MEMORY_W0_TASKS); +} + +#[test] +fn no_scope_lists_every_entity() { + let analyzer = fixed_analyzer(); + let resp = analyzer + .list_entities(request_scoped(None, None, None)) + .unwrap(); + + // Every task is ranked regardless of which resource it used. + assert_eq!(resp.total, 12); + assert_eq!(ids(&resp), ALL_TASKS_RANKED); +} + +#[test] +fn scope_restricts_to_the_resources_tasks() { + let analyzer = fixed_analyzer(); + let resp = analyzer + .list_entities(request(fixed::MEMORY_W1, None, None)) + .unwrap(); + + assert_eq!(resp.total, 4); + assert_eq!(ids(&resp), MEMORY_W1_TASKS); +} + +#[test] +fn min_usage_filter_includes_or_excludes_by_threshold() { + let analyzer = fixed_analyzer(); + + // Each task's memory usage is 0.75s; a lower threshold keeps all. + let kept = analyzer + .list_entities(request(fixed::MEMORY_W0, Some(0.5), None)) + .unwrap(); + assert_eq!(kept.total, 8); + + // A threshold above 0.75s drops every task. + let dropped = analyzer + .list_entities(request(fixed::MEMORY_W0, Some(1.0), None)) + .unwrap(); + assert_eq!(dropped.total, 0); + assert!(dropped.items.is_empty()); +} + +#[test] +fn pagination_slices_the_ranked_set_with_stable_total() { + let analyzer = fixed_analyzer(); + + let page0 = analyzer + .list_entities(request( + fixed::MEMORY_W0, + None, + Some(PageParams { max: 3, page: 0 }), + )) + .unwrap(); + assert_eq!(page0.total, 8); + assert_eq!(ids(&page0), MEMORY_W0_TASKS[0..3]); + + let page2 = analyzer + .list_entities(request( + fixed::MEMORY_W0, + None, + Some(PageParams { max: 3, page: 2 }), + )) + .unwrap(); + assert_eq!(page2.total, 8); + assert_eq!(ids(&page2), MEMORY_W0_TASKS[6..8]); +} + +#[test] +fn operator_filter_restricts_to_an_operators_tasks() { + let analyzer = fixed_analyzer(); + + // ScanFilter_W0 runs exactly TASK_0 and TASK_1. + let request = EntityListRequest { + entry: entry(None, None, None, Some(fixed::PHYS_SCAN_FILTER_W0)), + app_params: QueryFilter { + query_id: fixed::QUERY, + }, + }; + let resp = analyzer.list_entities(request).unwrap(); + + assert_eq!(resp.total, 2); + assert_eq!(ids(&resp), [fixed::TASK_0, fixed::TASK_1]); +} diff --git a/examples/simulator/analyzer/src/lib.rs b/examples/simulator/analyzer/src/lib.rs index fff817744..f9e142607 100644 --- a/examples/simulator/analyzer/src/lib.rs +++ b/examples/simulator/analyzer/src/lib.rs @@ -3,7 +3,10 @@ use quent_events::Event; pub use quent_query_engine_analyzer::QueryEngineModel; -use quent_query_engine_analyzer::ui::{QuentViewer, UiAnalyzer, ViewerEventStream}; +use quent_query_engine_analyzer::{ + entities, + ui::{QuentViewer, UiAnalyzer, ViewerEventStream}, +}; use quent_query_engine_ui::{OperatorFilter, QueryBundle, QueryEntities, QueryFilter}; use quent_ui::{ FiniteStateMachine, ResourceGroupNode, ResourceTree, convert_resource_tree, @@ -256,6 +259,52 @@ impl UiAnalyzer for SimulatorUiAnalyzer { &self.model } + fn list_entities( + &self, + request: quent_ui::entities::request::EntityListRequest, + ) -> AnalyzerResult { + let query_id = request.app_params.query_id; + let epoch = self.query_engine_model().query_epoch(query_id)?; + let entry = request.entry; + let window = entry.window.try_into_span(epoch)?; + let scope = entry + .filter + .scope + .as_ref() + .map(|s| s.resolve(&self.model)) + .transpose()?; + let operator_filter = entry.application.operator_id; + + // Restrict candidates to the requested query: a task belongs to a query + // iff its operator is one of that query's operators. Without this, tasks + // from a different query sharing a resource and overlapping the window + // would leak in. + let query_operators: HashSet = self + .model + .query_view(query_id)? + .operators() + .map(|op| op.id()) + .collect(); + + entities::list_entities( + &self.model, + |task| { + task.operator_id().is_some_and(|op| { + query_operators.contains(&op) + && operator_filter.is_none_or(|filter| op == filter) + }) + }, + entities::ListQuery { + scope: scope.as_ref(), + window, + filter: &entry.filter, + sort: entry.sort, + page: entry.page, + epoch, + }, + ) + } + // TODO(johanpel): consider re-using the bulk request API with a single entry for requests like this. fn single_resource_timeline( &self, diff --git a/examples/simulator/instrumentation/Cargo.toml b/examples/simulator/instrumentation/Cargo.toml index 2defcbd1e..d4e480e90 100644 --- a/examples/simulator/instrumentation/Cargo.toml +++ b/examples/simulator/instrumentation/Cargo.toml @@ -9,6 +9,7 @@ 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 112404873..1f0ba051a 100644 --- a/examples/simulator/instrumentation/src/lib.rs +++ b/examples/simulator/instrumentation/src/lib.rs @@ -40,3 +40,5 @@ 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 new file mode 100644 index 000000000..5bddba794 --- /dev/null +++ b/examples/simulator/instrumentation/src/test_utils.rs @@ -0,0 +1,63 @@ +// 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/build.rs b/examples/simulator/server/build.rs index a209df7f6..b8c5df33b 100644 --- a/examples/simulator/server/build.rs +++ b/examples/simulator/server/build.rs @@ -3,6 +3,7 @@ use quent_query_engine_ui::{OperatorFilter, QueryBundle, QueryFilter}; use quent_simulator_ui::EntityRef; +use quent_ui::entities::{request::EntityListRequest, response::EntityListResponse}; use quent_ui::timeline::{ request::{BulkTimelineRequest, SingleTimelineRequest}, response::{BulkTimelinesResponse, SingleTimelineResponse}, @@ -22,5 +23,8 @@ fn main() -> Result<(), Box> { as TS>::export_all(&cfg)?; ::export_all(&cfg)?; + as TS>::export_all(&cfg)?; + ::export_all(&cfg)?; + Ok(()) } diff --git a/examples/simulator/server/ts-bindings/EntityListEntry.ts b/examples/simulator/server/ts-bindings/EntityListEntry.ts new file mode 100644 index 000000000..0dace70f3 --- /dev/null +++ b/examples/simulator/server/ts-bindings/EntityListEntry.ts @@ -0,0 +1,34 @@ +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. +import type { EntityListFilter } from "./EntityListFilter"; +import type { PageParams } from "./PageParams"; +import type { Sort } from "./Sort"; +import type { TimeWindow } from "./TimeWindow"; + +/** + * A single entity-list query. + */ +export type EntityListEntry = { +/** + * The window of time entities must fall in. + */ +window: TimeWindow, +/** + * Filter parameters. + */ +filter: EntityListFilter, +/** + * Sort parameters. + */ +sort: Sort, +/** + * Pagination parameters. + * + * When this is not set, return the full list. Depending on the dataset + * size and other parameters, this may result in a large volume of data and + * should be used with care. + */ +page: PageParams | null, +/** + * Per-query application-specific parameters. + */ +application: EntryParams, }; diff --git a/examples/simulator/server/ts-bindings/EntityListFilter.ts b/examples/simulator/server/ts-bindings/EntityListFilter.ts new file mode 100644 index 000000000..e8c0a03c3 --- /dev/null +++ b/examples/simulator/server/ts-bindings/EntityListFilter.ts @@ -0,0 +1,23 @@ +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. +import type { EntityScope } from "./EntityScope"; + +/** + * Entity filters. + * + * Every field that is set must match, `None` fields do not filter. + */ +export type EntityListFilter = { +/** + * Restrict resulting entities to be in this scope. + */ +scope: EntityScope | null, +/** + * Restrict resulting entities to be of this type. + */ +entity_type_name: string | null, +/** + * Keep only entities with resource usages longer than this threshold. + * + * N.B. only Fsm-type entities can have usages. + */ +min_usage_s: number | null, }; diff --git a/examples/simulator/server/ts-bindings/EntityListRequest.ts b/examples/simulator/server/ts-bindings/EntityListRequest.ts new file mode 100644 index 000000000..0a2991fd0 --- /dev/null +++ b/examples/simulator/server/ts-bindings/EntityListRequest.ts @@ -0,0 +1,11 @@ +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. +import type { EntityListEntry } from "./EntityListEntry"; + +/** + * Parameters for listing entities. + */ +export type EntityListRequest = { entry: EntityListEntry, +/** + * Global application parameters shared by the query. + */ +app_params: GlobalParams, }; diff --git a/examples/simulator/server/ts-bindings/EntityListResponse.ts b/examples/simulator/server/ts-bindings/EntityListResponse.ts new file mode 100644 index 000000000..3c24d2572 --- /dev/null +++ b/examples/simulator/server/ts-bindings/EntityListResponse.ts @@ -0,0 +1,11 @@ +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. +import type { FiniteStateMachine } from "./FiniteStateMachine"; + +/** + * A ranked, paged list of entities. + */ +export type EntityListResponse = { items: Array, +/** + * The count of entities matching the filter before paging. + */ +total: number, }; diff --git a/examples/simulator/server/ts-bindings/EntityScope.ts b/examples/simulator/server/ts-bindings/EntityScope.ts new file mode 100644 index 000000000..af3df67a1 --- /dev/null +++ b/examples/simulator/server/ts-bindings/EntityScope.ts @@ -0,0 +1,6 @@ +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. + +/** + * Restricts returned entities to appear in a certain scope. + */ +export type EntityScope = { "Resource": { resource_id: string, } } | { "ResourceGroup": { resource_group_id: string, resource_type_name: string, } }; diff --git a/examples/simulator/server/ts-bindings/EntitySortKey.ts b/examples/simulator/server/ts-bindings/EntitySortKey.ts new file mode 100644 index 000000000..73113cd19 --- /dev/null +++ b/examples/simulator/server/ts-bindings/EntitySortKey.ts @@ -0,0 +1,6 @@ +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. + +/** + * The key entities are sorted by. + */ +export type EntitySortKey = "UsageDuration"; diff --git a/examples/simulator/server/ts-bindings/PageParams.ts b/examples/simulator/server/ts-bindings/PageParams.ts new file mode 100644 index 000000000..a225ca519 --- /dev/null +++ b/examples/simulator/server/ts-bindings/PageParams.ts @@ -0,0 +1,14 @@ +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. + +/** + * Parameters for paginated lists. + */ +export type PageParams = { +/** + * The maximum size of a page. + */ +max: number, +/** + * The zero-based index of the requested page. + */ +page: number, }; diff --git a/examples/simulator/server/ts-bindings/Sort.ts b/examples/simulator/server/ts-bindings/Sort.ts new file mode 100644 index 000000000..e7930af86 --- /dev/null +++ b/examples/simulator/server/ts-bindings/Sort.ts @@ -0,0 +1,16 @@ +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. +import type { EntitySortKey } from "./EntitySortKey"; +import type { SortDir } from "./SortDir"; + +/** + * Sorting parameters. + */ +export type Sort = { +/** + * The key to sort on. + */ +key: EntitySortKey, +/** + * The direction to sort in. + */ +dir: SortDir, }; diff --git a/examples/simulator/server/ts-bindings/SortDir.ts b/examples/simulator/server/ts-bindings/SortDir.ts new file mode 100644 index 000000000..feccd3b5b --- /dev/null +++ b/examples/simulator/server/ts-bindings/SortDir.ts @@ -0,0 +1,6 @@ +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. + +/** + * The direction to sort in. + */ +export type SortDir = "Asc" | "Desc"; diff --git a/examples/simulator/server/ts-bindings/TimeWindow.ts b/examples/simulator/server/ts-bindings/TimeWindow.ts new file mode 100644 index 000000000..c96fdcad2 --- /dev/null +++ b/examples/simulator/server/ts-bindings/TimeWindow.ts @@ -0,0 +1,14 @@ +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. + +/** + * A time window, resolved against an epoch supplied at conversion. + */ +export type TimeWindow = { +/** + * The start time of the window in seconds. + */ +start: number, +/** + * The end time of the window in seconds. + */ +end: number, };