diff --git a/Cargo.lock b/Cargo.lock index 6c9768e9299..651657dee04 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7051,6 +7051,8 @@ dependencies = [ "duckdb", "error-stack", "hash-integrations-definition", + "opentelemetry", + "opentelemetry_sdk", "pretty_assertions", "proptest", "serde", diff --git a/libs/@local/integrations/engine/Cargo.toml b/libs/@local/integrations/engine/Cargo.toml index b9402070afb..3139ca8fc4f 100644 --- a/libs/@local/integrations/engine/Cargo.toml +++ b/libs/@local/integrations/engine/Cargo.toml @@ -20,6 +20,7 @@ duckdb = { workspace = true, public = true, features = ["bundled", "json"] } # Private third-party dependencies derive_more = { workspace = true, features = ["display", "error"] } +opentelemetry = { workspace = true, features = ["metrics"] } serde_json = { workspace = true } simple-mermaid = { workspace = true } tokio = { workspace = true, features = ["sync"] } @@ -27,6 +28,7 @@ tracing = { workspace = true, features = ["attributes"] } uuid = { workspace = true, features = ["v5"] } [dev-dependencies] +opentelemetry_sdk = { workspace = true, features = ["metrics", "testing"] } pretty_assertions = { workspace = true } proptest = { workspace = true, features = ["attr-macro", "std"] } serde = { workspace = true, features = ["derive"] } diff --git a/libs/@local/integrations/engine/README.md b/libs/@local/integrations/engine/README.md index 29487d7f163..c1fc028d9d0 100644 --- a/libs/@local/integrations/engine/README.md +++ b/libs/@local/integrations/engine/README.md @@ -5,3 +5,17 @@ Runs an integration's pipelines in DuckDB. Each run gets a `Workspace`, a DuckDB database file in which the run's queries execute. A query is a single `SELECT`, and cannot open files, load extensions or change settings. `derive_entity_id` derives stable entity IDs from an integration namespace, entity type and source record ID using UUID v5. + +`Progress` counts an operation's observed work and emits structured tracing events at most every ten seconds as work advances. Call `report` for a final observation, including short operations. Reports carry the operation name, observed count, optional expected total, and elapsed seconds in the tracing span where tracking started. `snapshot` reads the counters for other reporting backends. + +Workspace operations export OpenTelemetry metrics through the application's global meter provider. Initialize it before opening workspaces, for example through `hash_telemetry::TelemetryRegistry::init_global`. + +| Metric | Observed value | +| --------------------------------------- | -------------------------------------------------------------------------------------------------- | +| `integrations.workspace.requests` | Request counts by operation and outcome: `success`, `failure`, `cancelled`, or `skipped`. | +| `integrations.workspace.duration` | Request execution duration in seconds, excluding queue time. | +| `integrations.workspace.rows` | Rows read by queries or materialized into tables, including work followed by a disk-limit failure. | +| `integrations.workspace.disk.usage` | Workspace bytes measured by existing disk-limit checks. | +| `integrations.workspace.snapshot.bytes` | Bytes copied to successfully published snapshots. | + +The operation labels are `open`, `query`, `create_table`, `snapshot`, and `close`. `cancelled` means a caller dropped an executing request; `skipped` means it dropped the request before execution. Metrics aggregate across workspaces. diff --git a/libs/@local/integrations/engine/src/lib.rs b/libs/@local/integrations/engine/src/lib.rs index 5f00a9aafd6..5ef3ebbdccb 100644 --- a/libs/@local/integrations/engine/src/lib.rs +++ b/libs/@local/integrations/engine/src/lib.rs @@ -17,6 +17,7 @@ extern crate alloc; mod byte_size; mod entity_id; +mod progress; mod sql; mod variant_name; mod workspace; @@ -24,6 +25,7 @@ mod workspace; pub use self::{ byte_size::ByteSize, entity_id::derive_entity_id, + progress::{Progress, ProgressSnapshot}, workspace::{ CloseError, DataErrorType, DiskUsageError, OpenError, QueryErrorType, Rows, Setting, SnapshotError, StatementError, Workspace, WorkspaceConfig, diff --git a/libs/@local/integrations/engine/src/progress/mod.rs b/libs/@local/integrations/engine/src/progress/mod.rs new file mode 100644 index 00000000000..125419f2b4f --- /dev/null +++ b/libs/@local/integrations/engine/src/progress/mod.rs @@ -0,0 +1,100 @@ +#[cfg(test)] +mod tests; + +use core::{ + sync::atomic::{AtomicU64, Ordering}, + time::Duration, +}; +use std::time::Instant; + +use tracing::Span; + +const REPORT_INTERVAL: Duration = Duration::from_secs(10); + +/// Tracks an operation's observed work and reports it through structured tracing events. +/// +/// Share it between tasks to accumulate completed work. Periodic reports are emitted by +/// [`advance`](Self::advance); call [`report`](Self::report) for a final observation. Events +/// belong to the tracing span in which the operation started. +#[derive(Debug)] +pub struct Progress { + operation: Box, + total: Option, + completed: AtomicU64, + started: Instant, + last_report_ms: AtomicU64, + span: Span, +} + +/// The observed work and elapsed time of an operation at the time of a report. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ProgressSnapshot { + /// The number of work items observed so far. + pub completed: u64, + /// The expected number of work items, if known when the operation started. + pub total: Option, + /// The time since the operation started. + pub elapsed: Duration, +} + +impl Progress { + /// Starts tracking an operation, with an optional expected work count. + #[must_use] + pub fn start(operation: impl Into>, total: Option) -> Self { + Self { + operation: operation.into(), + total, + completed: AtomicU64::new(0), + started: Instant::now(), + last_report_ms: AtomicU64::new(0), + span: Span::current(), + } + } + + /// Adds observed work, saturating at the counter's maximum, and reports at most every 10 + /// seconds. + pub fn advance(&self, count: u64) { + self.completed + .update(Ordering::Relaxed, Ordering::Relaxed, |completed| { + completed.saturating_add(count) + }); + if self.claim_report(self.started.elapsed()) { + self.report(); + } + } + + /// Reads the observed work count and elapsed time. + #[must_use] + pub fn snapshot(&self) -> ProgressSnapshot { + ProgressSnapshot { + completed: self.completed.load(Ordering::Relaxed), + total: self.total, + elapsed: self.started.elapsed(), + } + } + + /// Emits the observed progress immediately, including operations shorter than the report + /// interval. + pub fn report(&self) { + let snapshot = self.snapshot(); + tracing::info!( + parent: &self.span, + operation = %self.operation, + completed = snapshot.completed, + total = snapshot.total, + elapsed_seconds = snapshot.elapsed.as_secs_f64(), + "integration progress" + ); + } + + /// Reserves a periodic report when its interval has elapsed. + fn claim_report(&self, elapsed: Duration) -> bool { + let elapsed_ms = u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX); + self.last_report_ms + .try_update(Ordering::Relaxed, Ordering::Relaxed, |last| { + (Duration::from_millis(elapsed_ms.saturating_sub(last)) >= REPORT_INTERVAL) + .then_some(elapsed_ms) + }) + .is_ok() + } +} diff --git a/libs/@local/integrations/engine/src/progress/tests.rs b/libs/@local/integrations/engine/src/progress/tests.rs new file mode 100644 index 00000000000..ce3177a49eb --- /dev/null +++ b/libs/@local/integrations/engine/src/progress/tests.rs @@ -0,0 +1,102 @@ +use core::time::Duration; +use std::thread; + +use pretty_assertions::assert_eq; + +use super::Progress; + +#[test] +fn progress_concurrent_updates() { + let progress = Progress::start("apply", Some(800)); + thread::scope(|scope| { + for _ in 0..8 { + scope.spawn(|| { + for _ in 0..100 { + progress.advance(1); + } + }); + } + }); + + let snapshot = progress.snapshot(); + assert_eq!( + snapshot.completed, 800, + "all observed work should be counted" + ); + assert_eq!( + snapshot.total, + Some(800), + "the expected total should remain fixed" + ); +} + +#[test] +fn progress_exceeds_expected_total() { + let progress = Progress::start("apply", Some(2)); + progress.advance(3); + assert_eq!( + progress.snapshot().completed, + 3, + "observed work should not be capped by an estimate" + ); + + let progress = Progress::start("scan", None); + progress.advance(u64::MAX); + progress.advance(1); + let snapshot = progress.snapshot(); + assert_eq!( + snapshot.completed, + u64::MAX, + "the counter should saturate instead of wrapping" + ); + assert_eq!( + snapshot.total, None, + "an unknown total should remain unknown" + ); +} + +#[test] +fn progress_report_interval() { + let progress = Progress::start("apply", None); + assert!( + !progress.claim_report(Duration::from_millis(9_999)), + "an early report should be suppressed" + ); + assert!( + progress.claim_report(Duration::from_secs(10)), + "the first report should be due after 10 seconds" + ); + assert!( + !progress.claim_report(Duration::from_secs(10)), + "only one task should claim each report" + ); + assert!( + !progress.claim_report(Duration::from_secs(9)), + "an older observation should not move the interval backwards" + ); + assert!( + progress.claim_report(Duration::from_secs(20)), + "the next interval should allow another report" + ); +} + +#[test] +fn progress_concurrent_reports() { + let progress = Progress::start("apply", None); + let reports = thread::scope(|scope| { + let workers: Vec<_> = core::iter::repeat_with(|| { + scope.spawn(|| progress.claim_report(Duration::from_secs(10))) + }) + .take(8) + .collect(); + workers + .into_iter() + .map(|worker| worker.join().expect("reporting thread should finish")) + .filter(|claimed| *claimed) + .count() + }); + assert_eq!( + reports, 1, + "concurrent updates should reserve only one periodic report" + ); +} diff --git a/libs/@local/integrations/engine/src/workspace/connection.rs b/libs/@local/integrations/engine/src/workspace/connection.rs index d49a5b6825a..a01ac532087 100644 --- a/libs/@local/integrations/engine/src/workspace/connection.rs +++ b/libs/@local/integrations/engine/src/workspace/connection.rs @@ -4,6 +4,7 @@ use std::{ fs::{self, File, OpenOptions, TryLockError}, io, panic, path::{Path, PathBuf}, + time::Instant, }; use duckdb::{Config, Statement, params_from_iter, types::Value}; @@ -19,6 +20,7 @@ use crate::{ StatementError, files::WorkspaceFiles, interrupt::{Interrupt, InterruptGuard, RequestId}, + metrics::Metrics, reply::Reply, select::Select, }, @@ -40,6 +42,7 @@ pub(super) struct Connection { files: WorkspaceFiles, disk_limit: Option, variables: Variables, + metrics: Metrics, /// An exclusive lock on the workspace's lock file, held until the database is closed. lock: File, } @@ -99,6 +102,7 @@ impl Connection { })?; used = used.saturating_add(size); } + self.metrics.disk_usage(used); Ok(ByteSize::from_bytes(used)) } @@ -134,7 +138,11 @@ impl Connection { } /// Opens the database at `path` with the limits in `config` and its settings locked. - pub(super) fn open(path: PathBuf, config: WorkspaceConfig) -> Result> { + fn open( + path: PathBuf, + config: WorkspaceConfig, + metrics: Metrics, + ) -> Result> { let files = WorkspaceFiles::new(path); let lock = Self::lock(&files)?; @@ -184,6 +192,7 @@ impl Connection { files, disk_limit: config.disk_limit, variables: config.variables, + metrics, lock, }; match opened.disk_usage().change_context(OpenError::DiskUsage)? { @@ -272,6 +281,8 @@ impl Connection { .into_iter() .map(String::into_boxed_str) .collect(); + self.metrics + .rows("query", u64::try_from(rows.len()).unwrap_or(u64::MAX)); self.check_disk_limit()?; Ok(Rows::new(columns, rows)) } @@ -289,9 +300,10 @@ impl Connection { )) .map_err(StatementError::from_duckdb)?; let values = self.parameter_values(&statement)?; - statement - .execute(params_from_iter(values)) + let rows = statement + .query_row(params_from_iter(values), |row| row.get(0)) .map_err(StatementError::from_duckdb)?; + self.metrics.rows("create_table", rows); self.check_disk_limit() } @@ -299,7 +311,7 @@ impl Connection { /// /// An existing file at `partial` is removed first, so the copy cannot write through a link to /// another file, such as the database file itself. - fn write_copy(&self, partial: &Path, target: &Path) -> io::Result<()> { + fn write_copy(&self, partial: &Path, target: &Path) -> io::Result { match fs::remove_file(partial) { Ok(()) => {} Err(error) if error.kind() == io::ErrorKind::NotFound => {} @@ -309,7 +321,7 @@ impl Connection { .write(true) .create_new(true) .open(partial)?; - io::copy(&mut File::open(self.files.database())?, &mut copy)?; + let bytes = io::copy(&mut File::open(self.files.database())?, &mut copy)?; copy.sync_all()?; drop(copy); fs::rename(partial, target)?; @@ -318,7 +330,8 @@ impl Connection { .parent() .filter(|parent| !parent.as_os_str().is_empty()) .unwrap_or_else(|| Path::new(".")); - File::open(directory)?.sync_all() + File::open(directory)?.sync_all()?; + Ok(bytes) } #[tracing::instrument(level = "debug", skip(self))] @@ -336,7 +349,7 @@ impl Connection { .execute_batch("CHECKPOINT") .change_context(SnapshotError::Checkpoint)?; - self.write_copy(&partial, target).map_err(|error| { + let bytes = self.write_copy(&partial, target).map_err(|error| { if let Err(removal) = fs::remove_file(&partial) && removal.kind() != io::ErrorKind::NotFound { @@ -349,7 +362,9 @@ impl Connection { Report::new(error).change_context(SnapshotError::Write { target: target.to_owned(), }) - }) + })?; + self.metrics.snapshot(bytes); + Ok(()) } /// Checkpoints the database, then closes it whether or not the checkpoint succeeded. @@ -357,6 +372,7 @@ impl Connection { /// DuckDB also checkpoints when it closes the database, but it cannot report a failure from /// there. fn close(self) -> Result<(), Report> { + let started = Instant::now(); let checkpoint = self .duckdb .execute_batch("CHECKPOINT") @@ -364,6 +380,15 @@ impl Connection { drop(self.interrupt); drop(self.duckdb); drop(self.lock); + self.metrics.request( + "close", + if checkpoint.is_ok() { + "success" + } else { + "failure" + }, + started.elapsed(), + ); checkpoint } @@ -378,18 +403,31 @@ impl Connection { &self, id: RequestId, reply: Reply, + operation: &'static str, request: impl FnOnce(&Self) -> Result>, ) where C: Error + Send + Sync + 'static, { + let started = Instant::now(); let interrupt = self.interrupt(); interrupt.start(id); if reply.is_closed() { interrupt.finish(); + self.metrics + .request(operation, "skipped", started.elapsed()); return; } let result = request(self); - if !interrupt.finish() { + let cancelled = interrupt.finish(); + let outcome = if cancelled { + "cancelled" + } else if result.is_ok() { + "success" + } else { + "failure" + }; + self.metrics.request(operation, outcome, started.elapsed()); + if !cancelled { reply.send(result); } } @@ -401,15 +439,17 @@ impl Connection { while let Some((id, command)) = commands.blocking_recv() { match command { Command::Query { sql, reply } => { - self.answer(id, reply, |connection| connection.query(&sql)); + self.answer(id, reply, "query", |connection| connection.query(&sql)); } Command::CreateTable { table, sql, reply } => { - self.answer(id, reply, |connection| { + self.answer(id, reply, "create_table", |connection| { connection.create_table(&table, &sql) }); } Command::Snapshot { target, reply } => { - self.answer(id, reply, |connection| connection.snapshot(&target)); + self.answer(id, reply, "snapshot", |connection| { + connection.snapshot(&target) + }); } Command::Close { reply } => { reply.send(self.close()); @@ -434,7 +474,15 @@ impl Connection { ready: Reply, OpenError>, commands: &mut mpsc::Receiver<(RequestId, Command)>, ) { - let connection = match Self::open(path, config) { + let metrics = Metrics::new(); + let started = Instant::now(); + let opened = Self::open(path, config, metrics.clone()); + metrics.request( + "open", + if opened.is_ok() { "success" } else { "failure" }, + started.elapsed(), + ); + let connection = match opened { Ok(connection) => connection, Err(report) => { ready.send(Err(report)); diff --git a/libs/@local/integrations/engine/src/workspace/metrics.rs b/libs/@local/integrations/engine/src/workspace/metrics.rs new file mode 100644 index 00000000000..93b041fae61 --- /dev/null +++ b/libs/@local/integrations/engine/src/workspace/metrics.rs @@ -0,0 +1,76 @@ +use core::time::Duration; + +use opentelemetry::{ + KeyValue, global, + metrics::{Counter, Histogram}, +}; + +/// Records the outcomes observed by workspace threads through the process's meter provider. +#[derive(Clone)] +pub(super) struct Metrics { + requests: Counter, + duration: Histogram, + rows: Counter, + disk_usage: Histogram, + snapshot_bytes: Counter, +} + +impl Metrics { + pub(super) fn new() -> Self { + let meter = global::meter("hash-integrations-engine"); + Self { + requests: meter + .u64_counter("integrations.workspace.requests") + .with_description("Workspace requests by operation and outcome") + .build(), + duration: meter + .f64_histogram("integrations.workspace.duration") + .with_description("Workspace operation execution time") + .with_unit("s") + .build(), + rows: meter + .u64_counter("integrations.workspace.rows") + .with_description("Rows read by queries or materialized into tables") + .build(), + disk_usage: meter + .u64_histogram("integrations.workspace.disk.usage") + .with_description("Workspace file sizes observed during disk-limit checks") + .with_unit("By") + .build(), + snapshot_bytes: meter + .u64_counter("integrations.workspace.snapshot.bytes") + .with_description("Bytes written to successfully published workspace snapshots") + .with_unit("By") + .build(), + } + } + + /// Records a request's outcome and execution time, excluding time in the queue. + pub(super) fn request( + &self, + operation: &'static str, + outcome: &'static str, + elapsed: Duration, + ) { + let attributes = [ + KeyValue::new("operation", operation), + KeyValue::new("outcome", outcome), + ]; + self.requests.add(1, &attributes); + self.duration.record(elapsed.as_secs_f64(), &attributes); + } + + /// Counts rows already read or materialized, including work followed by a disk-limit failure. + pub(super) fn rows(&self, operation: &'static str, count: u64) { + self.rows + .add(count, &[KeyValue::new("operation", operation)]); + } + + pub(super) fn disk_usage(&self, bytes: u64) { + self.disk_usage.record(bytes, &[]); + } + + pub(super) fn snapshot(&self, bytes: u64) { + self.snapshot_bytes.add(bytes, &[]); + } +} diff --git a/libs/@local/integrations/engine/src/workspace/mod.rs b/libs/@local/integrations/engine/src/workspace/mod.rs index 641528b7cf1..6f394d72f12 100644 --- a/libs/@local/integrations/engine/src/workspace/mod.rs +++ b/libs/@local/integrations/engine/src/workspace/mod.rs @@ -2,6 +2,7 @@ mod connection; mod error; mod files; mod interrupt; +mod metrics; mod reply; mod rows; mod select; diff --git a/libs/@local/integrations/engine/tests/workspace_metrics.rs b/libs/@local/integrations/engine/tests/workspace_metrics.rs new file mode 100644 index 00000000000..91b0041f479 --- /dev/null +++ b/libs/@local/integrations/engine/tests/workspace_metrics.rs @@ -0,0 +1,204 @@ +//! Workspace outcomes exported through the application's OpenTelemetry meter provider. + +use core::{future::poll_fn, pin::pin, task::Poll}; +use std::fs; + +use hash_integrations_engine::{ + ByteSize, OpenError, SnapshotError, StatementError, Workspace, WorkspaceConfig, +}; +use opentelemetry::{KeyValue, global}; +use opentelemetry_sdk::metrics::{ + InMemoryMetricExporter, SdkMeterProvider, + data::{ + AggregatedMetrics, HistogramDataPoint, Metric, MetricData, ResourceMetrics, ScopeMetrics, + SumDataPoint, + }, +}; +use pretty_assertions::assert_eq; +use tempfile::TempDir; + +fn metric<'a>(resources: &'a [ResourceMetrics], name: &str) -> &'a Metric { + resources + .iter() + .flat_map(ResourceMetrics::scope_metrics) + .flat_map(ScopeMetrics::metrics) + .find(|metric| metric.name() == name) + .expect("the workspace should export the metric") +} + +fn counter(resources: &[ResourceMetrics], name: &str, attributes: &[KeyValue]) -> u64 { + let AggregatedMetrics::U64(MetricData::Sum(sum)) = metric(resources, name).data() else { + panic!("the metric should be an unsigned counter"); + }; + sum.data_points() + .filter(|point| { + attributes + .iter() + .all(|expected| point.attributes().any(|actual| actual == expected)) + }) + .map(SumDataPoint::value) + .sum() +} + +async fn exercise_workspace() -> u64 { + let directory = TempDir::new().expect("the temporary directory should be created"); + let path = directory.path().join("workspace.duckdb"); + let workspace = Workspace::open( + path.clone(), + WorkspaceConfig { + disk_limit: Some(ByteSize::from_bytes(64 * 1024 * 1024)), + ..WorkspaceConfig::default() + }, + ) + .await + .expect("the workspace should open"); + let report = Workspace::open(path, WorkspaceConfig::default()) + .await + .expect_err("the workspace lock should prevent a second open"); + assert!( + matches!(report.current_context(), OpenError::Locked { .. }), + "the second open should fail on the lock" + ); + workspace + .create_table("items".into(), "SELECT * FROM range(3)".into()) + .await + .expect("the table should be created"); + workspace + .query("SELECT * FROM items LIMIT 2".into()) + .await + .expect("the query should return two rows"); + let report = workspace + .query("SELECT * FROM missing_table".into()) + .await + .expect_err("the missing table should fail the query"); + assert!( + matches!( + report.current_context(), + StatementError::InvalidQuery { .. } + ), + "the missing table should produce a query error" + ); + + let snapshot_path = directory.path().join("snapshot.duckdb"); + workspace + .snapshot(snapshot_path.clone()) + .await + .expect("the snapshot should be published"); + let report = workspace + .snapshot(directory.path().join("missing/snapshot.duckdb")) + .await + .expect_err("the absent directory should fail the snapshot"); + assert!( + matches!(report.current_context(), SnapshotError::Write { .. }), + "the absent directory should produce a write error" + ); + + { + let mut query = pin!(workspace.query("SELECT sum(i) FROM range(10000000000) t(i)".into())); + assert!( + poll_fn(|context| Poll::Ready(query.as_mut().poll(context).is_pending())).await, + "the long query should still be pending" + ); + } + workspace.close().await.expect("the workspace should close"); + fs::metadata(snapshot_path) + .expect("the snapshot should exist") + .len() +} + +#[tokio::test] +async fn workspace_observed_outcomes() { + let exporter = InMemoryMetricExporter::default(); + let provider = SdkMeterProvider::builder() + .with_periodic_exporter(exporter.clone()) + .build(); + global::set_meter_provider(provider.clone()); + + let snapshot_bytes = exercise_workspace().await; + provider.force_flush().expect("the metrics should flush"); + let recorded = exporter + .get_finished_metrics() + .expect("the metrics should be exported"); + + for (operation, outcome, expected) in [ + ("open", "success", 1), + ("open", "failure", 1), + ("create_table", "success", 1), + ("query", "success", 1), + ("query", "failure", 1), + ("snapshot", "success", 1), + ("snapshot", "failure", 1), + ("close", "success", 1), + ] { + assert_eq!( + counter( + &recorded, + "integrations.workspace.requests", + &[ + KeyValue::new("operation", operation), + KeyValue::new("outcome", outcome), + ] + ), + expected, + "each observed {operation} {outcome} should be counted once" + ); + } + let cancelled = ["cancelled", "skipped"] + .into_iter() + .map(|outcome| { + counter( + &recorded, + "integrations.workspace.requests", + &[ + KeyValue::new("operation", "query"), + KeyValue::new("outcome", outcome), + ], + ) + }) + .sum::(); + assert_eq!( + cancelled, 1, + "the dropped query should be cancelled or skipped" + ); + for (operation, expected) in [("query", 2), ("create_table", 3)] { + assert_eq!( + counter( + &recorded, + "integrations.workspace.rows", + &[KeyValue::new("operation", operation)] + ), + expected, + "observed {operation} rows should be counted" + ); + } + assert_eq!( + counter(&recorded, "integrations.workspace.snapshot.bytes", &[]), + snapshot_bytes, + "only successfully published snapshot bytes should be counted" + ); + let AggregatedMetrics::U64(MetricData::Histogram(disk)) = + metric(&recorded, "integrations.workspace.disk.usage").data() + else { + panic!("disk usage should be recorded as a histogram"); + }; + assert!( + disk.data_points().any(|point| point.sum() > 0), + "disk checks should report the observed file sizes" + ); + let AggregatedMetrics::F64(MetricData::Histogram(duration)) = + metric(&recorded, "integrations.workspace.duration").data() + else { + panic!("request duration should be recorded as a histogram"); + }; + assert_eq!( + duration + .data_points() + .map(HistogramDataPoint::count) + .sum::(), + 9, + "each observed request should have an execution duration" + ); + provider + .shutdown() + .expect("the meter provider should shut down"); +}