diff --git a/CHANGELOG.md b/CHANGELOG.md index 6e6ed37..537caf0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), ### Added +- `db.client.response.affected_rows` histogram – a custom metric (no OTel semconv equivalent) mirroring the existing `db.response.affected_rows` span attribute. Recorded on `execute()` calls; carries the same connection / annotation / error attribute set as the duration histogram so dashboards can slice mutation throughput by the same dimensions ([#33](https://github.com/chmodas/sqlx-otel/pull/33)). - `PoolBuilder::with_network_protocol_name` and `with_network_transport` builder methods, plus a per-backend `Database::DEFAULT_NETWORK_PROTOCOL_NAME` constant (Postgres → `"postgresql"`, MySQL → `"mysql"`, SQLite → `None`). `network.protocol.name`, `network.transport`, and `db.client.connection.pool.name` now surface on every span and per-operation metric data point so dashboards can slice query latency by the same dimensions OTel's database-spans semconv recommends ([#32](https://github.com/chmodas/sqlx-otel/pull/32)). ### Changed diff --git a/README.md b/README.md index dd5bd4c..6730d41 100644 --- a/README.md +++ b/README.md @@ -168,13 +168,16 @@ On error, the span status is set to `Error` and an `exception` event is added wi ### Operation metrics -| Instrument | Type | Unit | Description | -|------------------------------------|-----------|------|---------------------------------------| -| `db.client.operation.duration` | Histogram | `s` | Duration of each database operation | -| `db.client.response.returned_rows` | Histogram | | Number of rows returned per operation | +| Instrument | Type | Unit | Description | +|------------------------------------|-----------|------|-----------------------------------------------| +| `db.client.operation.duration` | Histogram | `s` | Duration of each database operation | +| `db.client.response.returned_rows` | Histogram | | Number of rows returned per `fetch*` call | +| `db.client.response.affected_rows` | Histogram | | Rows affected per `execute` call (custom) | These mirror the bounded portion of the span attribute set: connection-level attributes (`db.system.name`, `db.namespace`, `server.address`/`port`, `network.peer.address`/`port`, `network.protocol.name`, `network.transport`, `db.client.connection.pool.name` – wherever set), plus annotation-derived attributes (`db.operation.name`, `db.collection.name`, `db.query.summary`, `db.stored_procedure.name`) when present, plus error-path attributes (`error.type`, plus `db.response.status_code` for `sqlx::Error::Database`) on the error path. `db.query.text` is deliberately excluded for cardinality; `db.query.summary` is caller-controlled and inherits its cardinality cost from the span side. +`db.client.response.affected_rows` is not part of the OpenTelemetry semantic conventions – we ship it for the same reason as the matching span attribute: backends report a useful database-confirmed count that's worth slicing alongside duration. It is recorded only on `execute()` calls (where `QueryResult::rows_affected()` is meaningful) and is not recorded for `execute_many` (deprecated upstream). + ### Connection pool metrics | Instrument | Type | Unit | Description | diff --git a/src/executor.rs b/src/executor.rs index 80b5c23..5e70600 100644 --- a/src/executor.rs +++ b/src/executor.rs @@ -94,7 +94,7 @@ fn start_span(name: &str, span_attrs: Vec) -> (OtelContext, Instant) { /// attributes plus the four annotation-derived attributes when present, plus error-path attributes /// (`error.type`, `db.response.status_code`) appended later by `record_error`. The unbounded /// `db.query.text` attribute is deliberately excluded; `db.query.summary` is caller-controlled and -/// can be unbounded — that cardinality cost is inherited from the span side. +/// can be unbounded – that cardinality cost is inherited from the span side. fn begin_query_span( attrs: &ConnectionAttributes, sql: Option<&str>, @@ -187,16 +187,19 @@ fn record_affected_rows(cx: &OtelContext, rows: u64) { )); } -/// End the span and record metrics. +/// End the span and record metrics. `returned_rows` is `Some` for `fetch*` paths, +/// `affected_rows` is `Some` for `execute` paths; both are `None` for paths that report +/// neither (e.g. `prepare` / `describe` / `execute_many`'s streaming aggregate). fn finish( cx: &OtelContext, start: Instant, - rows: Option, + returned_rows: Option, + affected_rows: Option, metrics: &Metrics, attrs: &[KeyValue], ) { cx.span().end(); - metrics.record(start.elapsed(), rows, attrs); + metrics.record(start.elapsed(), returned_rows, affected_rows, attrs); } /// Await a future, record any error on the span, then finish. Used by `execute`, `prepare`, @@ -212,7 +215,7 @@ async fn execute_instrumented( if let Err(err) = &result { record_error(&cx, err, &mut metric_attrs); } - finish(&cx, start, None, &metrics, &metric_attrs); + finish(&cx, start, None, None, &metrics, &metric_attrs); result } @@ -297,6 +300,7 @@ impl InstrumentedStream { &self.cx, self.start, Some(self.rows), + None, &self.metrics, &self.metric_attrs, ); @@ -395,15 +399,18 @@ macro_rules! impl_executor { let fut = ($inner).execute(query); Box::pin(async move { let result = fut.await; - match &result { + let affected = match &result { Ok(qr) => { - record_affected_rows(&cx, DB::rows_affected(qr)); + let n = DB::rows_affected(qr); + record_affected_rows(&cx, n); + Some(n) } Err(err) => { record_error(&cx, err, &mut metric_attrs); + None } - } - finish(&cx, start, None, &state.metrics, &metric_attrs); + }; + finish(&cx, start, None, affected, &state.metrics, &metric_attrs); result }) } @@ -512,11 +519,11 @@ macro_rules! impl_executor { Ok(rows) => { let count = rows.len() as u64; record_rows(&cx, count); - finish(&cx, start, Some(count), &state.metrics, &metric_attrs); + finish(&cx, start, Some(count), None, &state.metrics, &metric_attrs); } Err(err) => { record_error(&cx, err, &mut metric_attrs); - finish(&cx, start, None, &state.metrics, &metric_attrs); + finish(&cx, start, None, None, &state.metrics, &metric_attrs); } } result @@ -545,11 +552,11 @@ macro_rules! impl_executor { match &result { Ok(_) => { record_rows(&cx, 1); - finish(&cx, start, Some(1), &state.metrics, &metric_attrs); + finish(&cx, start, Some(1), None, &state.metrics, &metric_attrs); } Err(err) => { record_error(&cx, err, &mut metric_attrs); - finish(&cx, start, None, &state.metrics, &metric_attrs); + finish(&cx, start, None, None, &state.metrics, &metric_attrs); } } result @@ -579,11 +586,11 @@ macro_rules! impl_executor { Ok(maybe_row) => { let count = u64::from(maybe_row.is_some()); record_rows(&cx, count); - finish(&cx, start, Some(count), &state.metrics, &metric_attrs); + finish(&cx, start, Some(count), None, &state.metrics, &metric_attrs); } Err(err) => { record_error(&cx, err, &mut metric_attrs); - finish(&cx, start, None, &state.metrics, &metric_attrs); + finish(&cx, start, None, None, &state.metrics, &metric_attrs); } } result @@ -1177,7 +1184,7 @@ mod tests { /// the appended key set is exactly `{"db.operation.name" iff op.is_some(), /// "db.collection.name" iff coll.is_some(), "db.query.summary" iff /// query_summary.is_some(), "db.stored_procedure.name" iff - /// stored_procedure.is_some()}` — and nothing else, in particular none of the + /// stored_procedure.is_some()}` – and nothing else, in particular none of the /// connection or query-text keys leak through. #[test] fn append_annotation_attrs_membership_invariant(ann in any_annotations()) { diff --git a/src/metrics.rs b/src/metrics.rs index 4c3e64e..ba038ba 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -13,6 +13,11 @@ use opentelemetry_semantic_conventions::metric; pub(crate) struct Metrics { duration: Histogram, returned_rows: Histogram, + /// Custom histogram (no `OTel` semconv equivalent) recording the database-confirmed + /// `rows_affected()` count for `execute()` operations. Mirrors the existing + /// `db.response.affected_rows` span attribute so dashboards can slice mutation + /// throughput by the same dimensions. + affected_rows: Histogram, } impl Metrics { @@ -28,18 +33,38 @@ impl Metrics { .f64_histogram(metric::DB_CLIENT_RESPONSE_RETURNED_ROWS) .with_description("Number of rows returned by database operations.") .build(); + let affected_rows = meter + .f64_histogram("db.client.response.affected_rows") + .with_description("Number of rows affected by database operations.") + .build(); Self { duration, returned_rows, + affected_rows, } } - /// Record a completed operation's duration and, optionally, the number of rows returned. - pub fn record(&self, elapsed: Duration, rows: Option, attributes: &[KeyValue]) { + /// Record a completed operation: always the duration histogram; `returned_rows` and + /// `affected_rows` histograms when their respective counts are `Some`. The two row- + /// count parameters are mutually exclusive in practice (a `fetch*` operation sets + /// `returned_rows`; an `execute` operation sets `affected_rows`), but the signature + /// allows both for forward compatibility with backends that report both for a single + /// operation. + pub fn record( + &self, + elapsed: Duration, + returned_rows: Option, + affected_rows: Option, + attributes: &[KeyValue], + ) { self.duration.record(elapsed.as_secs_f64(), attributes); - if let Some(count) = rows { + if let Some(count) = returned_rows { #[allow(clippy::cast_precision_loss)] self.returned_rows.record(count as f64, attributes); } + if let Some(count) = affected_rows { + #[allow(clippy::cast_precision_loss)] + self.affected_rows.record(count as f64, attributes); + } } } diff --git a/tests/common.rs b/tests/common.rs index 67dfce6..25b69f7 100644 --- a/tests/common.rs +++ b/tests/common.rs @@ -102,6 +102,65 @@ pub fn metric_attr( .map(|kv| kv.value.clone()) } +/// Find a single histogram data point on the named metric whose attribute set contains +/// the given expected key/value pairs. Returns `None` if the metric is absent or no data +/// point matches. Generalises [`find_duration_data_point_with`] to any histogram (e.g. +/// `db.client.response.affected_rows`). +pub fn find_histogram_data_point_with( + metrics: &[opentelemetry_sdk::metrics::data::ResourceMetrics], + metric_name: &str, + expected: &[(&str, &str)], +) -> Option> { + use opentelemetry_sdk::metrics::data::{AggregatedMetrics, MetricData}; + for rm in metrics { + for sm in rm.scope_metrics() { + for metric in sm.metrics() { + if metric.name() != metric_name { + continue; + } + if let AggregatedMetrics::F64(MetricData::Histogram(hist)) = metric.data() { + for dp in hist.data_points() { + let matches = expected.iter().all(|(k, v)| { + dp.attributes().any(|kv| { + kv.key.as_str() == *k + && matches!( + &kv.value, + opentelemetry::Value::String(s) if s.as_str() == *v + ) + }) + }); + if matches { + return Some(dp.clone()); + } + } + } + } + } + } + None +} + +/// Assert the `db.client.response.affected_rows` histogram has at least one data point +/// for the given backend `system`. The exact recorded value is already pinned on the +/// span via `db.response.affected_rows`; this helper checks that the metric mirror is +/// reaching the meter (the in-memory exporter aggregates cumulatively, so per-call value +/// assertions across `tel.reset()` boundaries are not robust). +pub fn assert_affected_rows_metric(tel: &TestTelemetry, system: &str) { + let metrics = tel.metrics(); + let dp = find_histogram_data_point_with( + &metrics, + "db.client.response.affected_rows", + &[("db.system.name", system)], + ) + .unwrap_or_else(|| { + panic!("no db.client.response.affected_rows data point found for system {system:?}") + }); + assert!( + dp.count() > 0, + "db.client.response.affected_rows data point has zero count", + ); +} + /// Locate the `db.client.operation.duration` histogram in a `ResourceMetrics` snapshot and /// return the first data point. /// @@ -135,7 +194,7 @@ pub fn find_duration_data_point( /// /// Used by [`assert_metric_data_point`] and the per-method macros to verify that /// instrumentation for a specific scenario landed on the histogram with the dimensions -/// the test asserts the *span* carries — i.e. metric/span attribute parity. +/// the test asserts the *span* carries – i.e. metric/span attribute parity. pub fn find_duration_data_point_with( metrics: &[opentelemetry_sdk::metrics::data::ResourceMetrics], expected: &[(&str, &str)], @@ -2277,7 +2336,7 @@ macro_rules! test_operation_duration_metric_carries_full_annotations { /// Targeted SQLSTATE assertion: on `sqlx::Error::Database`, the backend status code /// surfaces on the histogram as `db.response.status_code`. The expected code is backend- /// specific: `SQLite` extended result code `1` (`SQLITE_ERROR`), Postgres SQLSTATE -/// `42P01`, `MySQL` SQLSTATE `42S02` — each backend's `tests/{sqlite,postgres,mysql}.rs` +/// `42P01`, `MySQL` SQLSTATE `42S02` – each backend's `tests/{sqlite,postgres,mysql}.rs` /// passes the value it expects. Per-method `*_records_error` macros already assert the /// generic `error.type` propagation; this macro pins the SQLSTATE shape that varies per /// backend. @@ -3575,6 +3634,7 @@ macro_rules! test_execute_records_affected_rows { "inserting 3 rows should affect 3 rows" ); $crate::common::assert_metric_for_system(&tel, $dialect.system); + $crate::common::assert_affected_rows_metric(&tel, $dialect.system); tel.reset(); // --- Upsert (dialect-specific) --- @@ -3586,6 +3646,7 @@ macro_rules! test_execute_records_affected_rows { Some(opentelemetry::Value::I64($dialect.upsert_affected_rows)), "upsert affected_rows differs per backend" ); + $crate::common::assert_affected_rows_metric(&tel, $dialect.system); tel.reset(); // --- Update multiple rows (dialect-specific concat) --- @@ -3600,6 +3661,7 @@ macro_rules! test_execute_records_affected_rows { Some(opentelemetry::Value::I64(2)), "updating two rows should affect 2 rows" ); + $crate::common::assert_affected_rows_metric(&tel, $dialect.system); tel.reset(); // --- Delete multiple rows --- @@ -3614,6 +3676,7 @@ macro_rules! test_execute_records_affected_rows { Some(opentelemetry::Value::I64(3)), "deleting three rows should affect 3 rows" ); + $crate::common::assert_affected_rows_metric(&tel, $dialect.system); tel.reset(); // --- Delete with no matching rows --- @@ -3629,6 +3692,7 @@ macro_rules! test_execute_records_affected_rows { "deleting non-existent rows should affect 0 rows" ); $crate::common::assert_metric_for_system(&tel, $dialect.system); + $crate::common::assert_affected_rows_metric(&tel, $dialect.system); }}; }