From df763be77e2520d48a9ad70fd4e783f508e62347 Mon Sep 17 00:00:00 2001 From: Borislav Borisov Date: Mon, 4 May 2026 15:06:40 +0100 Subject: [PATCH] fix: Align OpenTelemetry attribute coverage with database-spans semconv MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The `db.client.operation.duration` histogram previously carried only connection-level attributes (`db.system.name`, `db.namespace`). Annotation-derived attributes (`db.operation.name`, `db.collection.name`, `db.query.summary`, `db.stored_procedure.name`) and error-path attributes (`error.type`, plus `db.response.status_code` for `sqlx::Error::Database`) are now appended to `metric_attrs` in lock-step with the span attribute set, so dashboards can slice DB latency by operation verb, target collection, or error class – the dimensions OTel's database-spans semconv recommends. `InstrumentedStream` gains a single-error latch so streams that yield multiple `Err` items before terminating cannot append duplicate `error.type` / `db.response.status_code` keys. Three additional attributes from the OTel database-spans recommended set now surface on every span and per-operation metric: `db.client.connection.pool.name` (previously only on the `db.client.connection.count` gauge; now on every span and the rest of the `db.client.connection.*` family via the shared connection-attribute set), `network.protocol.name` (per-backend default via the new `Database::DEFAULT_NETWORK_PROTOCOL_NAME` constant; overridable via `PoolBuilder::with_network_protocol_name`), and `network.transport` (set explicitly via `PoolBuilder::with_network_transport`; the wrapper does not infer transport from the connect string). --- CHANGELOG.md | 11 +- README.md | 67 ++--- src/annotations.rs | 3 + src/attributes.rs | 68 ++++- src/database.rs | 10 + src/executor.rs | 247 +++++++++++++++--- src/pool.rs | 33 ++- tests/common.rs | 627 ++++++++++++++++++++++++++++++++++++++++++--- tests/mysql.rs | 45 +++- tests/postgres.rs | 45 +++- tests/sqlite.rs | 49 +++- 11 files changed, 1090 insertions(+), 115 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d12f05c..6e6ed37 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,9 +6,18 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), ## [Unreleased] +### Added + +- `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 + +- `db.client.connection.pool.name` previously appeared only on the `db.client.connection.count` gauge; it now also propagates to spans, the `db.client.operation.duration` / `db.client.response.returned_rows` histograms, and the rest of the `db.client.connection.*` family via the shared connection-attribute set. The `count` gauge's attribute set is unchanged ([#32](https://github.com/chmodas/sqlx-otel/pull/32)). + ### Fixed -- Query-side `with_annotations` / `with_operation` now compile inside `Send`-required async contexts (axum handlers, `tokio::spawn`, `tower::Service`-bounded futures). The `AnnotatedQuery::fetch_*` / `execute` forwarders were rewritten from `async fn` into `fn -> impl Future + Send + 'e`, so the HRTB carried by the internal `IntoAnnotatedExecutor` impls no longer leaks into auto-trait inference of an opaque coroutine. Span output is byte-identical to the executor-side surface; metrics emission is unchanged by construction because both surfaces continue to funnel through the same `Annotated<'_, Pool>` `Executor` impl ([#NN](https://github.com/chmodas/sqlx-otel/pull/31)). +- Query-side `with_annotations` / `with_operation` now compile inside `Send`-required async contexts (axum handlers, `tokio::spawn`, `tower::Service`-bounded futures). The `AnnotatedQuery::fetch_*` / `execute` forwarders were rewritten from `async fn` into `fn -> impl Future + Send + 'e`, so the HRTB carried by the internal `IntoAnnotatedExecutor` impls no longer leaks into auto-trait inference of an opaque coroutine. Span output is byte-identical to the executor-side surface; metrics emission is unchanged by construction because both surfaces continue to funnel through the same `Annotated<'_, Pool>` `Executor` impl ([#31](https://github.com/chmodas/sqlx-otel/pull/31)). +- `db.client.operation.duration` histogram now carries annotation-derived attributes (`db.operation.name`, `db.collection.name`, `db.query.summary`, `db.stored_procedure.name`) and error-path attributes (`error.type`, plus `db.response.status_code` for `sqlx::Error::Database`). Previously only connection-level attributes (`db.system.name`, `db.namespace`) reached the histogram, so dashboards could not slice DB latency by operation verb, target collection, or error class. The `InstrumentedStream` poll loop also gains a single-error latch so streams that yield multiple `Err` items before terminating cannot append duplicate `error.type` / `db.response.status_code` keys ([#32](https://github.com/chmodas/sqlx-otel/pull/32)). ## [0.2.0] – 2026-04-28 diff --git a/README.md b/README.md index 389c2fb..dd5bd4c 100644 --- a/README.md +++ b/README.md @@ -13,9 +13,9 @@ The wrapper talks to the [`opentelemetry`](https://docs.rs/opentelemetry) API di ## Highlights -- **Spans on every operation.** Every `sqlx::Executor` method emits a `SpanKind::Client` span carrying `db.system.name`, `db.namespace`, `server.address`/`port`, `db.query.text`, returned/affected row counts, and SQLSTATE / `error.type` on failure. -- **Operation and pool metrics.** Histograms for query duration and rows returned; histograms, counters, and gauges for pool waits, hold times, timeouts, and connection state. -- **Caller-supplied annotations.** The library does not parse SQL. Per-query attributes (`db.operation.name`, `db.collection.name`, `db.query.summary`) come from a small annotation API. +- **Spans on every operation.** Every `sqlx::Executor` method emits a `SpanKind::Client` span carrying the OTel database-spans recommended set: `db.system.name`, `db.namespace`, `server.address`/`port`, `network.peer.address`/`port`, `network.protocol.name`, `network.transport`, `db.client.connection.pool.name`, `db.query.text`, returned/affected row counts, and SQLSTATE / `error.type` on failure. +- **Span/metric attribute parity.** The `db.client.operation.duration` histogram carries the same bounded attribute set as spans – every annotation, every error-path attribute, every connection-level attribute – so dashboards can slice query latency by operation verb, target collection, error class, or pool name. `db.query.text` is the only span attribute deliberately excluded from metrics for cardinality. +- **Caller-supplied annotations.** The library does not parse SQL. Per-query attributes (`db.operation.name`, `db.collection.name`, `db.query.summary`, `db.stored_procedure.name`) come from a small annotation API. - **Three backends, one API.** Postgres, SQLite, and MySQL behind feature flags; the wrapper API is identical across all three. - **Drop-in.** `&Pool` implements `sqlx::Executor`, so existing call sites keep working unchanged. @@ -75,13 +75,17 @@ let pool = PoolBuilder::from(raw_pool) .with_port(5432) // override server.port .with_network_peer_address("10.0.0.5") // network.peer.address (not auto-extracted) .with_network_peer_port(5432) // network.peer.port (not auto-extracted) + .with_network_protocol_name("postgresql") // override network.protocol.name (defaulted per backend) + .with_network_transport("tcp") // network.transport (not inferred from connect string) .with_query_text_mode(QueryTextMode::Off) - .with_pool_name("my-service-db") // required for connection pool gauges + .with_pool_name("my-service-db") // db.client.connection.pool.name + connection-count polling .with_pool_metrics_interval(Duration::from_secs(5)) .build(); ``` -`with_pool_name` is the switch that activates `db.client.connection.count` polling – the gauge is silent until both a pool name and a runtime feature are present. +`with_pool_name` populates `db.client.connection.pool.name` on every span and per-operation metric, and is the switch that activates `db.client.connection.count` polling – the gauge is silent until both a pool name and a runtime feature are present. + +`network.protocol.name` defaults to the backend's wire protocol (`"postgresql"` for Postgres, `"mysql"` for MySQL, absent for SQLite); override only when the connection is tunnelled through a different application-layer protocol. `network.transport` is not inferred from the connect string – callers who want this attribute on spans / metrics must declare it explicitly so the value reflects the deployment configuration rather than a guess. ## Per-query annotations @@ -137,25 +141,28 @@ See [`QueryAnnotations`](https://docs.rs/sqlx-otel/latest/sqlx_otel/struct.Query Set on every `Executor` method (`execute`, `fetch`, `fetch_all`, `fetch_one`, `fetch_optional`, `fetch_many`, `execute_many`, `prepare`, `prepare_with`, `describe`): -| Attribute | Source | Condition | -|-----------------------------|-------------------------------------------------|-----------------------------| -| `db.system.name` | Backend (`"postgresql"`, `"sqlite"`, `"mysql"`) | Always | -| `db.namespace` | Database name, extracted from connect options | When available | -| `server.address` | Hostname, extracted from connect options | When available | -| `server.port` | Port, extracted from connect options | When available | -| `network.peer.address` | Resolved IP address | When set via builder | -| `network.peer.port` | Resolved port | When set via builder | -| `db.query.text` | The SQL query string | Unless `QueryTextMode::Off` | -| `db.operation.name` | Database operation (e.g. `SELECT`) | When annotated | -| `db.collection.name` | Target table or collection | When annotated | -| `db.query.summary` | Low-cardinality query summary | When annotated | -| `db.stored_procedure.name` | Stored procedure name | When annotated | -| `db.response.returned_rows` | Row count | On `fetch*` methods | -| `db.response.affected_rows` | Rows affected (`rows_affected()`) | On `execute` | -| `db.response.status_code` | SQLSTATE error code | On database errors | -| `error.type` | Error variant name | On any error | - -`db.response.affected_rows` is not part of the OpenTelemetry semantic conventions but we find it useful so have included it. It is a custom attribute that reports the database-confirmed count from `QueryResult::rows_affected()`, carrying the same connection-level attributes as `db.response.returned_rows`. It is not recorded for `execute_many`, which is [considered deprecated by the SQLx team](https://github.com/launchbadge/sqlx/issues/3108). +| Attribute | Source | Condition | +|----------------------------------|-------------------------------------------------------------|-----------------------------| +| `db.system.name` | Backend (`"postgresql"`, `"sqlite"`, `"mysql"`) | Always | +| `db.namespace` | Database name, extracted from connect options | When available | +| `server.address` | Hostname, extracted from connect options | When available | +| `server.port` | Port, extracted from connect options | When available | +| `network.peer.address` | Resolved IP address | When set via builder | +| `network.peer.port` | Resolved port | When set via builder | +| `network.protocol.name` | Wire protocol; defaults per backend, overridable on builder | When applicable | +| `network.transport` | OSI L4 transport (`"tcp"`, `"unix"`, `"pipe"`, `"inproc"`) | When set via builder | +| `db.client.connection.pool.name` | Pool identifier set via `with_pool_name` | When set via builder | +| `db.query.text` | The SQL query string | Unless `QueryTextMode::Off` | +| `db.operation.name` | Database operation (e.g. `SELECT`) | When annotated | +| `db.collection.name` | Target table or collection | When annotated | +| `db.query.summary` | Low-cardinality query summary | When annotated | +| `db.stored_procedure.name` | Stored procedure name | When annotated | +| `db.response.returned_rows` | Row count | On `fetch*` methods | +| `db.response.affected_rows` | Rows affected (`rows_affected()`) | On `execute` | +| `db.response.status_code` | SQLSTATE error code | On database errors | +| `error.type` | Error variant name | On any error | + +`db.response.affected_rows` is not part of the OpenTelemetry semantic conventions, but we find it useful so have included it. It is a custom attribute that reports the database-confirmed count from `QueryResult::rows_affected()`, carrying the same connection-level attributes as `db.response.returned_rows`. It is not recorded for `execute_many`, which is [considered deprecated by the SQLx team](https://github.com/launchbadge/sqlx/issues/3108). On error, the span status is set to `Error` and an `exception` event is added with `exception.type` and `exception.message` attributes. @@ -166,7 +173,7 @@ On error, the span status is set to `Error` and an `exception` event is added wi | `db.client.operation.duration` | Histogram | `s` | Duration of each database operation | | `db.client.response.returned_rows` | Histogram | | Number of rows returned per operation | -These carry the connection-level attributes (`db.system.name`, `db.namespace`, `server.address`, `server.port`). +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. ### Connection pool metrics @@ -185,11 +192,11 @@ The first four are recorded inline on every `acquire()` / connection drop – no ## Backend support -| Backend | `db.namespace` | `server.address` / `server.port` | Notes | -|------------|--------------------|----------------------------------|------------------------------------| -| `postgres` | Database name | Yes (from connect URL) | – | -| `mysql` | Database name | Yes (from connect URL) | – | -| `sqlite` | File path or `:memory:` URI | n/a | No host/port; file or in-memory. | +| Backend | `db.namespace` | `server.address` / `server.port` | Default `network.protocol.name` | Notes | +|------------|-----------------------------|----------------------------------|-------------------------------------|----------------------------------| +| `postgres` | Database name | Yes (from connect URL) | `"postgresql"` | – | +| `mysql` | Database name | Yes (from connect URL) | `"mysql"` | – | +| `sqlite` | File path or `:memory:` URI | n/a | absent (embedded; no wire protocol) | No host/port; file or in-memory. | `db.response.returned_rows`, `db.response.affected_rows`, `db.response.status_code`, and `error.type` are recorded uniformly across all three backends. diff --git a/src/annotations.rs b/src/annotations.rs index 36915fe..fe1ac78 100644 --- a/src/annotations.rs +++ b/src/annotations.rs @@ -276,6 +276,9 @@ mod tests { namespace: None, network_peer_address: None, network_peer_port: None, + network_protocol_name: None, + network_transport: None, + pool_name: None, query_text_mode: QueryTextMode::Off, }), metrics: Arc::new(Metrics::new()), diff --git a/src/attributes.rs b/src/attributes.rs index 7d41368..292eeb8 100644 --- a/src/attributes.rs +++ b/src/attributes.rs @@ -57,6 +57,18 @@ pub(crate) struct ConnectionAttributes { pub network_peer_address: Option, /// `network.peer.port` – the resolved port, user-provided. pub network_peer_port: Option, + /// `network.protocol.name` – the OSI L7 wire protocol (e.g. `"postgresql"`, `"mysql"`). + /// `None` for embedded backends that do not speak a wire protocol (e.g. `SQLite`). + pub network_protocol_name: Option, + /// `network.transport` – the OSI L4 transport (`"tcp"`, `"udp"`, `"pipe"`, `"unix"`, + /// `"inproc"`). User-provided via [`PoolBuilder::with_network_transport`]( + /// crate::PoolBuilder::with_network_transport); the wrapper does not infer it from the + /// connect string. + pub network_transport: Option, + /// `db.client.connection.pool.name` – user-provided pool identifier shared with the + /// `db.client.connection.*` metric family. Surfaces on every span and per-operation + /// metric so dashboards can correlate query latency with pool-level signals. + pub pool_name: Option, /// Controls `db.query.text` capture. pub query_text_mode: QueryTextMode, } @@ -65,7 +77,7 @@ impl ConnectionAttributes { /// Produce the base `KeyValue` set for span and metric attribute lists. Only includes /// attributes that have a value – optional fields are omitted when `None`. pub fn base_key_values(&self) -> Vec { - let mut attrs = Vec::with_capacity(6); + let mut attrs = Vec::with_capacity(9); attrs.push(KeyValue::new(attribute::DB_SYSTEM_NAME, self.system)); if let Some(ref host) = self.host { attrs.push(KeyValue::new(attribute::SERVER_ADDRESS, host.clone())); @@ -82,6 +94,24 @@ impl ConnectionAttributes { if let Some(port) = self.network_peer_port { attrs.push(KeyValue::new(attribute::NETWORK_PEER_PORT, i64::from(port))); } + if let Some(ref proto) = self.network_protocol_name { + attrs.push(KeyValue::new( + attribute::NETWORK_PROTOCOL_NAME, + proto.clone(), + )); + } + if let Some(ref transport) = self.network_transport { + attrs.push(KeyValue::new( + attribute::NETWORK_TRANSPORT, + transport.clone(), + )); + } + if let Some(ref name) = self.pool_name { + attrs.push(KeyValue::new( + attribute::DB_CLIENT_CONNECTION_POOL_NAME, + name.clone(), + )); + } attrs } } @@ -210,16 +240,22 @@ mod tests { namespace: Some("mydb".into()), network_peer_address: Some("127.0.0.1".into()), network_peer_port: Some(5432), + network_protocol_name: Some("postgresql".into()), + network_transport: Some("tcp".into()), + pool_name: Some("primary".into()), query_text_mode: QueryTextMode::Full, }; let kvs = attrs.base_key_values(); - assert_eq!(kvs.len(), 6); + assert_eq!(kvs.len(), 9); assert_eq!(kvs[0].key.as_str(), "db.system.name"); assert_eq!(kvs[1].key.as_str(), "server.address"); assert_eq!(kvs[2].key.as_str(), "server.port"); assert_eq!(kvs[3].key.as_str(), "db.namespace"); assert_eq!(kvs[4].key.as_str(), "network.peer.address"); assert_eq!(kvs[5].key.as_str(), "network.peer.port"); + assert_eq!(kvs[6].key.as_str(), "network.protocol.name"); + assert_eq!(kvs[7].key.as_str(), "network.transport"); + assert_eq!(kvs[8].key.as_str(), "db.client.connection.pool.name"); } #[test] @@ -231,6 +267,9 @@ mod tests { namespace: None, network_peer_address: None, network_peer_port: None, + network_protocol_name: None, + network_transport: None, + pool_name: None, query_text_mode: QueryTextMode::Off, }; let kvs = attrs.base_key_values(); @@ -328,6 +367,9 @@ mod tests { namespace in proptest::option::of("[a-z]{1,16}"), network_peer_address in proptest::option::of("[0-9.:]{1,32}"), network_peer_port in proptest::option::of(any::()), + network_protocol_name in proptest::option::of("[a-z]{1,16}"), + network_transport in proptest::option::of("[a-z]{1,8}"), + pool_name in proptest::option::of("[a-z0-9-]{1,32}"), ) { let attrs = ConnectionAttributes { system: "sqlite", @@ -336,6 +378,9 @@ mod tests { namespace: namespace.clone(), network_peer_address: network_peer_address.clone(), network_peer_port, + network_protocol_name: network_protocol_name.clone(), + network_transport: network_transport.clone(), + pool_name: pool_name.clone(), query_text_mode: QueryTextMode::Off, }; let kvs = attrs.base_key_values(); @@ -344,9 +389,26 @@ mod tests { + usize::from(port.is_some()) + usize::from(namespace.is_some()) + usize::from(network_peer_address.is_some()) - + usize::from(network_peer_port.is_some()); + + usize::from(network_peer_port.is_some()) + + usize::from(network_protocol_name.is_some()) + + usize::from(network_transport.is_some()) + + usize::from(pool_name.is_some()); prop_assert_eq!(kvs.len(), expected); prop_assert_eq!(kvs[0].key.as_str(), "db.system.name"); + + let keys: Vec<&str> = kvs.iter().map(|k| k.key.as_str()).collect(); + prop_assert_eq!( + keys.contains(&"network.protocol.name"), + network_protocol_name.is_some(), + ); + prop_assert_eq!( + keys.contains(&"network.transport"), + network_transport.is_some(), + ); + prop_assert_eq!( + keys.contains(&"db.client.connection.pool.name"), + pool_name.is_some(), + ); } } } diff --git a/src/database.rs b/src/database.rs index 5c9a445..1ae63ac 100644 --- a/src/database.rs +++ b/src/database.rs @@ -18,6 +18,13 @@ pub trait Database: sqlx::Database + sealed::Sealed { /// `"sqlite"`, `"mysql"`). const SYSTEM: &'static str; + /// Default `network.protocol.name` for this backend. `Some("postgresql")` / + /// `Some("mysql")` for the network-protocol backends; `None` for embedded backends + /// that do not speak a wire protocol (e.g. `SQLite`). Used by `PoolBuilder::from` to + /// pre-populate the attribute on every span and per-operation metric; override via + /// [`PoolBuilder::with_network_protocol_name`](crate::PoolBuilder::with_network_protocol_name). + const DEFAULT_NETWORK_PROTOCOL_NAME: Option<&'static str>; + /// Extract host, port, and database namespace from the backend's connect options. /// /// Returns `(host, port, namespace)` where any component may be `None` if the backend @@ -74,6 +81,7 @@ fn url_based_connection_attributes( #[cfg_attr(docsrs, doc(cfg(feature = "sqlite")))] impl Database for sqlx::Sqlite { const SYSTEM: &'static str = "sqlite"; + const DEFAULT_NETWORK_PROTOCOL_NAME: Option<&'static str> = None; fn connection_attributes( pool: &sqlx::Pool, @@ -95,6 +103,7 @@ impl Database for sqlx::Sqlite { #[cfg_attr(docsrs, doc(cfg(feature = "postgres")))] impl Database for sqlx::Postgres { const SYSTEM: &'static str = "postgresql"; + const DEFAULT_NETWORK_PROTOCOL_NAME: Option<&'static str> = Some("postgresql"); fn connection_attributes( pool: &sqlx::Pool, @@ -111,6 +120,7 @@ impl Database for sqlx::Postgres { #[cfg_attr(docsrs, doc(cfg(feature = "mysql")))] impl Database for sqlx::MySql { const SYSTEM: &'static str = "mysql"; + const DEFAULT_NETWORK_PROTOCOL_NAME: Option<&'static str> = Some("mysql"); fn connection_attributes( pool: &sqlx::Pool, diff --git a/src/executor.rs b/src/executor.rs index 19a2290..80b5c23 100644 --- a/src/executor.rs +++ b/src/executor.rs @@ -18,6 +18,30 @@ use crate::metrics::Metrics; // Span helpers // --------------------------------------------------------------------------- +/// Append the four per-query semantic convention annotation attributes +/// (`db.operation.name`, `db.collection.name`, `db.query.summary`, `db.stored_procedure.name`) onto +/// the supplied vector, one push per field that is `Some`. Used by both the span attribute builder +/// and `begin_query_span`'s metric attribute list so the two emit identical annotation-derived +/// keys. +fn append_annotation_attrs(kv: &mut Vec, annotations: Option<&QueryAnnotations>) { + let Some(ann) = annotations else { return }; + if let Some(ref op) = ann.operation { + kv.push(KeyValue::new(attribute::DB_OPERATION_NAME, op.clone())); + } + if let Some(ref coll) = ann.collection { + kv.push(KeyValue::new(attribute::DB_COLLECTION_NAME, coll.clone())); + } + if let Some(ref summary) = ann.query_summary { + kv.push(KeyValue::new(attribute::DB_QUERY_SUMMARY, summary.clone())); + } + if let Some(ref sp) = ann.stored_procedure { + kv.push(KeyValue::new( + attribute::DB_STORED_PROCEDURE_NAME, + sp.clone(), + )); + } +} + /// Build span attributes for a query, combining connection-level and per-query values. /// /// When `annotations` is provided, the four per-query semantic convention attributes @@ -29,23 +53,7 @@ fn build_attributes( annotations: Option<&QueryAnnotations>, ) -> Vec { let mut kv = attrs.base_key_values(); - if let Some(ann) = annotations { - if let Some(ref op) = ann.operation { - kv.push(KeyValue::new(attribute::DB_OPERATION_NAME, op.clone())); - } - if let Some(ref coll) = ann.collection { - kv.push(KeyValue::new(attribute::DB_COLLECTION_NAME, coll.clone())); - } - if let Some(ref summary) = ann.query_summary { - kv.push(KeyValue::new(attribute::DB_QUERY_SUMMARY, summary.clone())); - } - if let Some(ref sp) = ann.stored_procedure { - kv.push(KeyValue::new( - attribute::DB_STORED_PROCEDURE_NAME, - sp.clone(), - )); - } - } + append_annotation_attrs(&mut kv, annotations); if let Some(sql) = sql { match attrs.query_text_mode { QueryTextMode::Full => { @@ -75,12 +83,18 @@ fn start_span(name: &str, span_attrs: Vec) -> (OtelContext, Instant) { (cx, Instant::now()) } -/// Start an instrumented query: derive the span name from the connection attributes and -/// per-query annotations, build the span and metric attribute lists, and open the span. +/// Start an instrumented query: derive the span name from the connection attributes and per-query +/// annotations, build the span and metric attribute lists, and open the span. +/// +/// Returns the span's context, the timing reference for `finish()`, and the metric attribute list. +/// This consolidates the boilerplate that every `Executor` method shares before delegating to the +/// inner `SQLx` call. /// -/// Returns the span's context, the timing reference for `finish()`, and the metric -/// attribute list. This consolidates the boilerplate that every `Executor` method shares -/// before delegating to the inner `SQLx` call. +/// The returned `metric_attrs` mirror the bounded portion of the span attribute set: connection +/// 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. fn begin_query_span( attrs: &ConnectionAttributes, sql: Option<&str>, @@ -95,7 +109,8 @@ fn begin_query_span( }); let name = attributes::span_name(attrs.system, op, coll, summary); let span_attrs = build_attributes(attrs, sql, annotations); - let metric_attrs = attrs.base_key_values(); + let mut metric_attrs = attrs.base_key_values(); + append_annotation_attrs(&mut metric_attrs, annotations); let (cx, start) = start_span(&name, span_attrs); (cx, start, metric_attrs) } @@ -123,27 +138,34 @@ fn error_type(err: &sqlx::Error) -> &'static str { } } -/// Record an error on the span within the given context: set status, `error.type`, and -/// add an exception event. -fn record_error(cx: &OtelContext, err: &sqlx::Error) { +/// Record an error on the span within the given context: set status, `error.type`, and add an +/// exception event. Also append `error.type` and `db.response.status_code` (SQLSTATE for +/// `sqlx::Error::Database`) onto `metric_attrs` so the histogram emission carries the same +/// error-path dimensions as the span. Single source of truth for `error_type(err)` and SQLSTATE +/// extraction. +fn record_error(cx: &OtelContext, err: &sqlx::Error, metric_attrs: &mut Vec) { let span = cx.span(); + let kind = error_type(err); span.set_status(Status::Error { description: Cow::Owned(err.to_string()), }); - span.set_attribute(KeyValue::new(attribute::ERROR_TYPE, error_type(err))); + span.set_attribute(KeyValue::new(attribute::ERROR_TYPE, kind)); + metric_attrs.push(KeyValue::new(attribute::ERROR_TYPE, kind)); // Extract SQLSTATE or database-specific error code when available. if let sqlx::Error::Database(db_err) = err { if let Some(code) = db_err.code() { + let code = code.into_owned(); span.set_attribute(KeyValue::new( attribute::DB_RESPONSE_STATUS_CODE, - code.into_owned(), + code.clone(), )); + metric_attrs.push(KeyValue::new(attribute::DB_RESPONSE_STATUS_CODE, code)); } } span.add_event( "exception", vec![ - KeyValue::new("exception.type", error_type(err)), + KeyValue::new("exception.type", kind), KeyValue::new("exception.message", err.to_string()), ], ); @@ -184,11 +206,11 @@ async fn execute_instrumented( cx: OtelContext, start: Instant, metrics: std::sync::Arc, - metric_attrs: Vec, + mut metric_attrs: Vec, ) -> Result { let result = fut.await; if let Err(err) = &result { - record_error(&cx, err); + record_error(&cx, err, &mut metric_attrs); } finish(&cx, start, None, &metrics, &metric_attrs); result @@ -241,6 +263,7 @@ struct InstrumentedStream { rows: u64, metrics: std::sync::Arc, metric_attrs: Vec, + error_recorded: bool, finished: bool, _counter: std::marker::PhantomData, } @@ -260,6 +283,7 @@ impl InstrumentedStream { rows: 0, metrics, metric_attrs, + error_recorded: false, finished: false, _counter: std::marker::PhantomData, } @@ -298,7 +322,16 @@ where Poll::Ready(Some(Ok(item))) } Poll::Ready(Some(Err(err))) => { - record_error(&self.cx, &err); + if !self.error_recorded { + self.error_recorded = true; + // Re-borrow `&mut *self` to split the disjoint-field borrow: + // `record_error` needs `&self.cx` (immutable) and `&mut self.metric_attrs` + // (mutable) simultaneously. Going through `Pin<&mut Self>::deref_mut` + // (sound here because of the explicit `Unpin` impl below) lets the + // borrow checker see the two fields as distinct. + let this = &mut *self; + record_error(&this.cx, &err, &mut this.metric_attrs); + } Poll::Ready(Some(Err(err))) } Poll::Ready(None) => { @@ -357,7 +390,7 @@ macro_rules! impl_executor { { let sql = query.sql().to_owned(); let state = $self_.state.clone(); - let (cx, start, metric_attrs) = + let (cx, start, mut metric_attrs) = begin_query_span(&state.attrs, Some(&sql), $ann); let fut = ($inner).execute(query); Box::pin(async move { @@ -367,7 +400,7 @@ macro_rules! impl_executor { record_affected_rows(&cx, DB::rows_affected(qr)); } Err(err) => { - record_error(&cx, err); + record_error(&cx, err, &mut metric_attrs); } } finish(&cx, start, None, &state.metrics, &metric_attrs); @@ -470,7 +503,7 @@ macro_rules! impl_executor { { let sql = query.sql().to_owned(); let state = $self_.state.clone(); - let (cx, start, metric_attrs) = + let (cx, start, mut metric_attrs) = begin_query_span(&state.attrs, Some(&sql), $ann); let fut = ($inner).fetch_all(query); Box::pin(async move { @@ -482,7 +515,7 @@ macro_rules! impl_executor { finish(&cx, start, Some(count), &state.metrics, &metric_attrs); } Err(err) => { - record_error(&cx, err); + record_error(&cx, err, &mut metric_attrs); finish(&cx, start, None, &state.metrics, &metric_attrs); } } @@ -504,7 +537,7 @@ macro_rules! impl_executor { { let sql = query.sql().to_owned(); let state = $self_.state.clone(); - let (cx, start, metric_attrs) = + let (cx, start, mut metric_attrs) = begin_query_span(&state.attrs, Some(&sql), $ann); let fut = ($inner).fetch_one(query); Box::pin(async move { @@ -515,7 +548,7 @@ macro_rules! impl_executor { finish(&cx, start, Some(1), &state.metrics, &metric_attrs); } Err(err) => { - record_error(&cx, err); + record_error(&cx, err, &mut metric_attrs); finish(&cx, start, None, &state.metrics, &metric_attrs); } } @@ -537,7 +570,7 @@ macro_rules! impl_executor { { let sql = query.sql().to_owned(); let state = $self_.state.clone(); - let (cx, start, metric_attrs) = + let (cx, start, mut metric_attrs) = begin_query_span(&state.attrs, Some(&sql), $ann); let fut = ($inner).fetch_optional(query); Box::pin(async move { @@ -549,7 +582,7 @@ macro_rules! impl_executor { finish(&cx, start, Some(count), &state.metrics, &metric_attrs); } Err(err) => { - record_error(&cx, err); + record_error(&cx, err, &mut metric_attrs); finish(&cx, start, None, &state.metrics, &metric_attrs); } } @@ -728,6 +761,53 @@ mod tests { // construct an unknown variant, but it ensures forward compatibility. } + /// `InstrumentedStream::poll_next`'s `error_recorded` guard prevents `record_error` + /// from running more than once when the underlying stream yields multiple `Err`s + /// before terminating. Without the guard, the metric's attribute slice would + /// accumulate duplicate `error.type` (and `db.response.status_code`) `KeyValue`s, + /// producing a malformed histogram data point. Driven directly via a mock stream so + /// the assertion does not depend on backend stream-termination semantics. + #[test] + fn instrumented_stream_records_error_only_once_when_polled_past_err() { + use futures::StreamExt as _; + use futures::executor::block_on; + use futures::stream; + + let metrics = std::sync::Arc::new(crate::metrics::Metrics::new()); + let metric_attrs = vec![KeyValue::new(attribute::DB_SYSTEM_NAME, "postgresql")]; + let (cx, start) = start_span("test", Vec::new()); + + // Yield two distinct `Err`s back-to-back, then `None`. + let inner = stream::iter(vec![ + Err::(sqlx::Error::ColumnNotFound("x".into())), + Err(sqlx::Error::ColumnNotFound("y".into())), + ]); + let mut s = InstrumentedStream::<_, CountAll>::new(inner, cx, start, metrics, metric_attrs); + + block_on(async { + assert!(matches!(s.next().await, Some(Err(_))), "expected first Err"); + assert!( + matches!(s.next().await, Some(Err(_))), + "expected second Err" + ); + assert!(s.next().await.is_none(), "expected stream to terminate"); + }); + + let error_type_count = s + .metric_attrs + .iter() + .filter(|kv| kv.key.as_str() == "error.type") + .count(); + assert_eq!( + error_type_count, 1, + "error.type must appear exactly once even when the stream yields multiple Err items", + ); + assert!( + s.error_recorded, + "error_recorded should latch true after the first Err", + ); + } + fn test_attrs() -> ConnectionAttributes { ConnectionAttributes { system: "postgresql", @@ -736,6 +816,9 @@ mod tests { namespace: Some("mydb".into()), network_peer_address: None, network_peer_port: None, + network_protocol_name: None, + network_transport: None, + pool_name: None, query_text_mode: QueryTextMode::Full, } } @@ -835,6 +918,53 @@ mod tests { ); } + #[test] + fn append_annotation_attrs_pushes_all_four_when_set() { + let ann = QueryAnnotations::new() + .operation("SELECT") + .collection("users") + .query_summary("users by id") + .stored_procedure("sp_get_users"); + let mut kv = Vec::new(); + append_annotation_attrs(&mut kv, Some(&ann)); + let pairs: Vec<(&str, &opentelemetry::Value)> = + kv.iter().map(|k| (k.key.as_str(), &k.value)).collect(); + assert_eq!(pairs.len(), 4, "expected one push per annotation field"); + assert!(pairs.contains(&( + "db.operation.name", + &opentelemetry::Value::String("SELECT".into()) + ))); + assert!(pairs.contains(&( + "db.collection.name", + &opentelemetry::Value::String("users".into()) + ))); + assert!(pairs.contains(&( + "db.query.summary", + &opentelemetry::Value::String("users by id".into()) + ))); + assert!(pairs.contains(&( + "db.stored_procedure.name", + &opentelemetry::Value::String("sp_get_users".into()) + ))); + } + + #[test] + fn append_annotation_attrs_none_pushes_nothing() { + let mut kv = Vec::new(); + append_annotation_attrs(&mut kv, None); + assert!(kv.is_empty(), "no pushes expected when annotations is None"); + } + + #[test] + fn append_annotation_attrs_default_pushes_nothing() { + let mut kv = Vec::new(); + append_annotation_attrs(&mut kv, Some(&QueryAnnotations::new())); + assert!( + kv.is_empty(), + "no pushes expected when every annotation field is None" + ); + } + #[test] fn build_attributes_annotation_field_permutations() { type Setter = fn(QueryAnnotations) -> QueryAnnotations; @@ -901,6 +1031,9 @@ mod tests { namespace, network_peer_address, network_peer_port, + network_protocol_name: None, + network_transport: None, + pool_name: None, query_text_mode, } } @@ -1039,5 +1172,39 @@ mod tests { prop_assert!(!keys.contains(&"db.query.summary")); prop_assert!(!keys.contains(&"db.stored_procedure.name")); } + + /// `append_annotation_attrs` membership invariant: starting from an empty vector, + /// 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 + /// connection or query-text keys leak through. + #[test] + fn append_annotation_attrs_membership_invariant(ann in any_annotations()) { + let mut kv = Vec::new(); + append_annotation_attrs(&mut kv, Some(&ann)); + let keys: Vec<&str> = kv.iter().map(|k| k.key.as_str()).collect(); + + prop_assert_eq!(keys.contains(&"db.operation.name"), ann.operation.is_some()); + prop_assert_eq!(keys.contains(&"db.collection.name"), ann.collection.is_some()); + prop_assert_eq!(keys.contains(&"db.query.summary"), ann.query_summary.is_some()); + prop_assert_eq!( + keys.contains(&"db.stored_procedure.name"), + ann.stored_procedure.is_some(), + ); + + // No connection or query-text keys leak in from a stray copy-paste of + // `build_attributes` semantics. + prop_assert!(!keys.contains(&"db.system.name")); + prop_assert!(!keys.contains(&"db.namespace")); + prop_assert!(!keys.contains(&"db.query.text")); + + // Cardinality matches the count of `Some` annotation fields. + let expected_count = usize::from(ann.operation.is_some()) + + usize::from(ann.collection.is_some()) + + usize::from(ann.query_summary.is_some()) + + usize::from(ann.stored_procedure.is_some()); + prop_assert_eq!(kv.len(), expected_count); + } } } diff --git a/src/pool.rs b/src/pool.rs index 1ad6a3a..1c6fa17 100644 --- a/src/pool.rs +++ b/src/pool.rs @@ -53,6 +53,8 @@ pub struct PoolBuilder { namespace: Option, network_peer_address: Option, network_peer_port: Option, + network_protocol_name: Option, + network_transport: Option, query_text_mode: QueryTextMode, pool_name: Option, pool_metrics_interval: Duration, @@ -60,7 +62,10 @@ pub struct PoolBuilder { impl From> for PoolBuilder { /// Create a builder from an existing `sqlx::Pool`, auto-extracting connection - /// attributes from the backend's connect options. + /// attributes from the backend's connect options. `network.protocol.name` is + /// pre-populated from [`Database::DEFAULT_NETWORK_PROTOCOL_NAME`] (the wire protocol + /// for Postgres / `MySQL`; absent for `SQLite`); override via + /// [`with_network_protocol_name`](Self::with_network_protocol_name). fn from(pool: sqlx::Pool) -> Self { let (host, port, namespace) = DB::connection_attributes(&pool); Self { @@ -70,6 +75,8 @@ impl From> for PoolBuilder { namespace, network_peer_address: None, network_peer_port: None, + network_protocol_name: DB::DEFAULT_NETWORK_PROTOCOL_NAME.map(String::from), + network_transport: None, query_text_mode: QueryTextMode::default(), pool_name: None, pool_metrics_interval: Duration::from_secs(10), @@ -113,6 +120,27 @@ impl PoolBuilder { self } + /// Override the `network.protocol.name` attribute. Defaults to the backend's wire + /// protocol via [`Database::DEFAULT_NETWORK_PROTOCOL_NAME`] (`"postgresql"` / + /// `"mysql"`; absent for `SQLite`). Override when the connection is tunnelled through + /// a different application-layer protocol or when reporting to a system that expects a + /// specific name. + #[must_use] + pub fn with_network_protocol_name(mut self, name: impl Into) -> Self { + self.network_protocol_name = Some(name.into()); + self + } + + /// Set the `network.transport` attribute (the OSI L4 transport: `"tcp"`, `"udp"`, + /// `"pipe"`, `"unix"`, `"inproc"`). The wrapper does not infer transport from the + /// connect string – callers who want this attribute on spans / metrics must set it + /// explicitly so the value reflects the deployment configuration rather than a guess. + #[must_use] + pub fn with_network_transport(mut self, transport: impl Into) -> Self { + self.network_transport = Some(transport.into()); + self + } + /// Configure how `db.query.text` is captured on spans. Defaults to /// [`QueryTextMode::Full`]. #[must_use] @@ -167,6 +195,9 @@ impl PoolBuilder { namespace: self.namespace, network_peer_address: self.network_peer_address, network_peer_port: self.network_peer_port, + network_protocol_name: self.network_protocol_name, + network_transport: self.network_transport, + pool_name: self.pool_name, query_text_mode: self.query_text_mode, }); let metrics = Arc::new(Metrics::new()); diff --git a/tests/common.rs b/tests/common.rs index eb5345a..67dfce6 100644 --- a/tests/common.rs +++ b/tests/common.rs @@ -87,6 +87,227 @@ pub fn attr(span: &SpanData, key: &str) -> Option { .map(|kv| kv.value.clone()) } +/// Find the attribute value for a given key on a histogram data point. +/// +/// Mirrors [`attr`] for the metric side: walks the data point's attribute iterator and +/// returns the cloned value for the first matching key. Used by the metric-attribute +/// assertions in `test_operation_duration_metric_carries_*` to confirm the histogram +/// emits the same dimensions the span carries. +pub fn metric_attr( + dp: &opentelemetry_sdk::metrics::data::HistogramDataPoint, + key: &str, +) -> Option { + dp.attributes() + .find(|kv| kv.key.as_str() == key) + .map(|kv| kv.value.clone()) +} + +/// Locate the `db.client.operation.duration` histogram in a `ResourceMetrics` snapshot and +/// return the first data point. +/// +/// Returns `None` if the metric is absent or has no data points. Tests that assert on the +/// data point's attributes should `unwrap()` the result so absence fails the test loudly +/// rather than silently passing an empty-attribute set. +pub fn find_duration_data_point( + metrics: &[opentelemetry_sdk::metrics::data::ResourceMetrics], +) -> 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() != "db.client.operation.duration" { + continue; + } + if let AggregatedMetrics::F64(MetricData::Histogram(hist)) = metric.data() { + if let Some(dp) = hist.data_points().next() { + return Some(dp.clone()); + } + } + } + } + } + None +} + +/// Find the first `db.client.operation.duration` data point whose attribute set contains +/// every `(key, value)` pair in `expected`. Each expected value is matched as a string +/// `opentelemetry::Value::String`. Returns `None` when no matching data point exists. +/// +/// 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. +pub fn find_duration_data_point_with( + metrics: &[opentelemetry_sdk::metrics::data::ResourceMetrics], + 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() != "db.client.operation.duration" { + 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 that the `db.client.operation.duration` histogram has a data point matching +/// every `(key, value)` pair in `expected`, and that its `count() > 0`. Panics with a +/// helpful message naming the missing pairs when no data point matches – the panic body +/// is intentionally verbose so assertion failures point at the unsatisfied dimension +/// rather than a generic "metric missing". +pub fn assert_metric_data_point(tel: &TestTelemetry, expected: &[(&str, &str)]) { + let metrics = tel.metrics(); + let dp = find_duration_data_point_with(&metrics, expected).unwrap_or_else(|| { + panic!( + "no db.client.operation.duration data point found matching expected attrs {expected:?}", + ) + }); + assert!( + dp.count() > 0, + "matching data point has zero count for expected attrs {expected:?}", + ); +} + +/// Assert that the duration histogram recorded a data point for the given backend +/// `system`. The minimum bar every per-method macro should clear: this is the metric +/// equivalent of [`assert_common_span_attributes`] and verifies that the method's +/// instrumentation reached the meter at all. +pub fn assert_metric_for_system(tel: &TestTelemetry, system: &str) { + assert_metric_data_point(tel, &[("db.system.name", system)]); +} + +/// Assert that the duration histogram recorded a data point matching the standard +/// [`test_annotations`] keys (`db.operation.name = "SELECT"`, +/// `db.collection.name = "users"`) plus `db.system.name`. Used by every macro that +/// exercises the annotated path so the metric carries the same dimensions the span +/// assertions check via [`assert_annotated_span`]. +pub fn assert_annotated_metric(tel: &TestTelemetry, dialect: &Dialect) { + assert_metric_data_point( + tel, + &[ + ("db.system.name", dialect.system), + ("db.operation.name", "SELECT"), + ("db.collection.name", "users"), + ], + ); +} + +/// Assert that the duration histogram recorded an *annotated* error-path data point +/// matching the standard [`test_annotations`] keys (`db.operation.name = "SELECT"`, +/// `db.collection.name = "users"`) plus a non-empty `error.type` plus the expected +/// `db.system.name`. Used by every `*_records_error` macro that exercises an annotated +/// failing call: pins the contract that `record_error` keeps annotation attrs and error +/// attrs on the same `Vec` so a single histogram data point carries both +/// dimensions, which dashboards rely on to slice latency by failed operation + failed +/// collection. +pub fn assert_annotated_error_metric(tel: &TestTelemetry, dialect: &Dialect) { + use opentelemetry_sdk::metrics::data::{AggregatedMetrics, MetricData}; + let metrics = tel.metrics(); + let dp = metrics + .iter() + .flat_map(opentelemetry_sdk::metrics::data::ResourceMetrics::scope_metrics) + .flat_map(opentelemetry_sdk::metrics::data::ScopeMetrics::metrics) + .filter(|m| m.name() == "db.client.operation.duration") + .filter_map(|m| { + if let AggregatedMetrics::F64(MetricData::Histogram(hist)) = m.data() { + Some(hist) + } else { + None + } + }) + .flat_map(opentelemetry_sdk::metrics::data::Histogram::data_points) + .find(|dp| { + let attr_eq = |key: &str, value: &str| { + dp.attributes().any(|kv| { + kv.key.as_str() == key + && matches!( + &kv.value, + opentelemetry::Value::String(s) if s.as_str() == value + ) + }) + }; + let attr_present = |key: &str| { + dp.attributes().any(|kv| { + kv.key.as_str() == key + && matches!( + &kv.value, + opentelemetry::Value::String(s) if !s.as_str().is_empty() + ) + }) + }; + attr_eq("db.system.name", dialect.system) + && attr_eq("db.operation.name", "SELECT") + && attr_eq("db.collection.name", "users") + && attr_present("error.type") + }); + assert!( + dp.is_some(), + "no annotated db.client.operation.duration data point with error.type and \ + db.system.name = {:?} found", + dialect.system, + ); +} + +/// Assert that the duration histogram recorded a data point on the error path: the data +/// point must carry `error.type` (any non-empty string – the exact value depends on the +/// `sqlx::Error` variant under test) and the expected `db.system.name`. Used by every +/// `*_records_error` macro. +pub fn assert_error_metric(tel: &TestTelemetry, dialect: &Dialect) { + use opentelemetry_sdk::metrics::data::{AggregatedMetrics, MetricData}; + let metrics = tel.metrics(); + let dp = metrics + .iter() + .flat_map(opentelemetry_sdk::metrics::data::ResourceMetrics::scope_metrics) + .flat_map(opentelemetry_sdk::metrics::data::ScopeMetrics::metrics) + .filter(|m| m.name() == "db.client.operation.duration") + .filter_map(|m| { + if let AggregatedMetrics::F64(MetricData::Histogram(hist)) = m.data() { + Some(hist) + } else { + None + } + }) + .flat_map(opentelemetry_sdk::metrics::data::Histogram::data_points) + .find(|dp| { + let has_error_type = dp.attributes().any(|kv| { + kv.key.as_str() == "error.type" + && matches!(&kv.value, opentelemetry::Value::String(s) if !s.as_str().is_empty()) + }); + let has_system = dp.attributes().any(|kv| { + kv.key.as_str() == "db.system.name" + && matches!( + &kv.value, + opentelemetry::Value::String(s) if s.as_str() == dialect.system + ) + }); + has_error_type && has_system + }); + assert!( + dp.is_some(), + "no db.client.operation.duration data point with error.type and db.system.name = {:?} found", + dialect.system, + ); +} + /// Assert that a span carries the common attributes every instrumented operation must have. /// /// `system` is the expected `db.system.name` value (e.g. `"sqlite"`, `"postgresql"`). @@ -419,6 +640,8 @@ macro_rules! test_execute_creates_span_via_pool { .await .unwrap(); $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -461,6 +684,8 @@ macro_rules! test_execute_creates_span_via_connection { .await .unwrap(); $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -505,6 +730,8 @@ macro_rules! test_execute_creates_span_via_transaction { assert!($crate::common::attr(&spans[0], "db.response.affected_rows").is_some()); $crate::common::assert_annotated_span(&spans[1], &$dialect); $crate::common::assert_annotated_span(&spans[2], &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -543,6 +770,8 @@ macro_rules! test_execute_records_error { let last = tel.spans().last().unwrap().clone(); $crate::common::assert_annotated_span(&last, &$dialect); $crate::common::assert_error_span(&last); + + $crate::common::assert_annotated_error_metric(&tel, &$dialect); }}; } @@ -581,6 +810,8 @@ macro_rules! test_execute_many_via_pool { while stream.next().await.is_some() {} drop(stream); $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -620,6 +851,8 @@ macro_rules! test_execute_many_via_connection { while stream.next().await.is_some() {} drop(stream); $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -661,6 +894,8 @@ macro_rules! test_execute_many_via_transaction { ); $crate::common::assert_annotated_span(&spans[1], &$dialect); $crate::common::assert_annotated_span(&spans[2], &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -706,6 +941,8 @@ macro_rules! test_execute_many_records_error { let last = tel.spans().last().unwrap().clone(); $crate::common::assert_annotated_span(&last, &$dialect); $crate::common::assert_error_span(&last); + + $crate::common::assert_annotated_error_metric(&tel, &$dialect); }}; } @@ -748,6 +985,8 @@ macro_rules! test_fetch_via_pool { while stream.next().await.is_some() {} drop(stream); $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -786,6 +1025,8 @@ macro_rules! test_fetch_via_connection { while stream.next().await.is_some() {} drop(stream); $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -826,6 +1067,8 @@ macro_rules! test_fetch_via_transaction { ); $crate::common::assert_annotated_span(&spans[1], &$dialect); $crate::common::assert_annotated_span(&spans[2], &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -855,6 +1098,8 @@ macro_rules! test_fetch_stream_dropped_early_still_records_span { $crate::common::attr(&spans[0], "db.response.returned_rows"), Some(opentelemetry::Value::I64(1)) ); + + $crate::common::assert_metric_for_system(&tel, $dialect.system); }}; } @@ -898,6 +1143,8 @@ macro_rules! test_fetch_stream_records_error { let last = tel.spans().last().unwrap().clone(); $crate::common::assert_annotated_span(&last, &$dialect); $crate::common::assert_error_span(&last); + + $crate::common::assert_annotated_error_metric(&tel, &$dialect); }}; } @@ -945,6 +1192,8 @@ macro_rules! test_fetch_many_via_pool { while stream.next().await.is_some() {} drop(stream); $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -983,6 +1232,8 @@ macro_rules! test_fetch_many_via_connection { while stream.next().await.is_some() {} drop(stream); $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -1023,6 +1274,8 @@ macro_rules! test_fetch_many_via_transaction { ); $crate::common::assert_annotated_span(&spans[1], &$dialect); $crate::common::assert_annotated_span(&spans[2], &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -1052,6 +1305,8 @@ macro_rules! test_fetch_many_dropped_early_still_records_span { $crate::common::attr(&spans[0], "db.response.returned_rows"), Some(opentelemetry::Value::I64(1)) ); + + $crate::common::assert_metric_for_system(&tel, $dialect.system); }}; } @@ -1097,6 +1352,8 @@ macro_rules! test_fetch_many_records_error { let last = tel.spans().last().unwrap().clone(); $crate::common::assert_annotated_span(&last, &$dialect); $crate::common::assert_error_span(&last); + + $crate::common::assert_annotated_error_metric(&tel, &$dialect); }}; } @@ -1134,6 +1391,8 @@ macro_rules! test_fetch_all_via_pool { .await .unwrap(); $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -1171,6 +1430,8 @@ macro_rules! test_fetch_all_via_connection { .await .unwrap(); $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -1210,6 +1471,8 @@ macro_rules! test_fetch_all_via_transaction { ); $crate::common::assert_annotated_span(&spans[1], &$dialect); $crate::common::assert_annotated_span(&spans[2], &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -1247,6 +1510,8 @@ macro_rules! test_fetch_all_records_error { let last = tel.spans().last().unwrap().clone(); $crate::common::assert_annotated_span(&last, &$dialect); $crate::common::assert_error_span(&last); + + $crate::common::assert_annotated_error_metric(&tel, &$dialect); }}; } @@ -1279,6 +1544,8 @@ macro_rules! test_fetch_one_via_pool { .await .unwrap(); $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -1312,6 +1579,8 @@ macro_rules! test_fetch_one_via_connection { .await .unwrap(); $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -1347,6 +1616,8 @@ macro_rules! test_fetch_one_via_transaction { ); $crate::common::assert_annotated_span(&spans[1], &$dialect); $crate::common::assert_annotated_span(&spans[2], &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -1384,6 +1655,8 @@ macro_rules! test_fetch_one_records_error { let last = tel.spans().last().unwrap().clone(); $crate::common::assert_annotated_span(&last, &$dialect); $crate::common::assert_error_span(&last); + + $crate::common::assert_annotated_error_metric(&tel, &$dialect); }}; } @@ -1417,6 +1690,8 @@ macro_rules! test_fetch_optional_records_one_row { .await .unwrap(); $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -1451,6 +1726,8 @@ macro_rules! test_fetch_optional_via_connection { .await .unwrap(); $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -1487,6 +1764,8 @@ macro_rules! test_fetch_optional_via_transaction { ); $crate::common::assert_annotated_span(&spans[1], &$dialect); $crate::common::assert_annotated_span(&spans[2], &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -1527,6 +1806,8 @@ macro_rules! test_fetch_optional_records_error { let last = tel.spans().last().unwrap().clone(); $crate::common::assert_annotated_span(&last, &$dialect); $crate::common::assert_error_span(&last); + + $crate::common::assert_annotated_error_metric(&tel, &$dialect); }}; } @@ -1560,6 +1841,8 @@ macro_rules! test_prepare_via_pool { .await .unwrap(); $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -1590,6 +1873,8 @@ macro_rules! test_prepare_via_connection { .await .unwrap(); $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -1622,6 +1907,8 @@ macro_rules! test_prepare_via_transaction { assert!($crate::common::attr(&spans[0], "db.response.returned_rows").is_none()); $crate::common::assert_annotated_span(&spans[1], &$dialect); $crate::common::assert_annotated_span(&spans[2], &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -1660,6 +1947,8 @@ macro_rules! test_prepare_records_error { let last = tel.spans().last().unwrap().clone(); $crate::common::assert_annotated_span(&last, &$dialect); $crate::common::assert_error_span(&last); + + $crate::common::assert_annotated_error_metric(&tel, &$dialect); }}; } @@ -1692,6 +1981,8 @@ macro_rules! test_prepare_with_via_pool { .await .unwrap(); $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -1725,6 +2016,8 @@ macro_rules! test_prepare_with_via_connection { .await .unwrap(); $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -1760,6 +2053,8 @@ macro_rules! test_prepare_with_via_transaction { assert!($crate::common::attr(&spans[0], "db.response.returned_rows").is_none()); $crate::common::assert_annotated_span(&spans[1], &$dialect); $crate::common::assert_annotated_span(&spans[2], &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -1798,6 +2093,8 @@ macro_rules! test_prepare_with_records_error { let last = tel.spans().last().unwrap().clone(); $crate::common::assert_annotated_span(&last, &$dialect); $crate::common::assert_error_span(&last); + + $crate::common::assert_annotated_error_metric(&tel, &$dialect); }}; } @@ -1827,6 +2124,8 @@ macro_rules! test_describe_via_pool { .await .unwrap(); $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -1857,6 +2156,8 @@ macro_rules! test_describe_via_connection { .await .unwrap(); $crate::common::assert_annotated_span(tel.spans().last().unwrap(), &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -1889,6 +2190,8 @@ macro_rules! test_describe_via_transaction { assert!($crate::common::attr(&spans[0], "db.response.returned_rows").is_none()); $crate::common::assert_annotated_span(&spans[1], &$dialect); $crate::common::assert_annotated_span(&spans[2], &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -1927,6 +2230,8 @@ macro_rules! test_describe_records_error { let last = tel.spans().last().unwrap().clone(); $crate::common::assert_annotated_span(&last, &$dialect); $crate::common::assert_error_span(&last); + + $crate::common::assert_annotated_error_metric(&tel, &$dialect); }}; } @@ -1934,45 +2239,68 @@ macro_rules! test_describe_records_error { // Misc: metrics, annotations // --------------------------------------------------------------------------- -/// `db.client.operation.duration` histogram is populated for any executed query. +/// All four annotation fields set together must surface on the `db.client.operation.duration` +/// histogram data point. Per-method macros only exercise the standard +/// `test_annotations()` shape (`db.operation.name = "SELECT"`, +/// `db.collection.name = "users"`), so this macro pins the integration-level guarantee +/// for `db.query.summary` and `db.stored_procedure.name` propagation. #[macro_export] -macro_rules! test_operation_duration_metric_is_recorded { +macro_rules! test_operation_duration_metric_carries_full_annotations { ($pool_factory:expr, $dialect:expr) => {{ - use opentelemetry_sdk::metrics::data::{AggregatedMetrics, MetricData}; - let _ = $dialect; // unused – backend doesn't influence the metric shape let tel = $crate::common::TestTelemetry::install(); let pool = $pool_factory; - let _: (i32,) = sqlx::query_as("SELECT 1").fetch_one(&pool).await.unwrap(); + pool.with_annotations( + sqlx_otel::QueryAnnotations::new() + .operation("SELECT") + .collection("users") + .query_summary("users by id") + .stored_procedure("sp_get_users"), + ) + .fetch_one("SELECT 1") + .await + .unwrap(); + + $crate::common::assert_metric_data_point( + &tel, + &[ + ("db.system.name", $dialect.system), + ("db.operation.name", "SELECT"), + ("db.collection.name", "users"), + ("db.query.summary", "users by id"), + ("db.stored_procedure.name", "sp_get_users"), + ], + ); + }}; +} + +/// 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` +/// 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. +#[macro_export] +macro_rules! test_operation_duration_metric_carries_sqlstate { + ($pool_factory:expr, $expected_code:expr) => {{ + let tel = $crate::common::TestTelemetry::install(); + let pool = $pool_factory; + + let result = pool + .with_operation("SELECT", "nonexistent_table_xyz") + .fetch_one("SELECT * FROM nonexistent_table_xyz") + .await; + assert!(result.is_err(), "expected fetch_one to fail"); let resource_metrics = tel.metrics(); - assert!(!resource_metrics.is_empty(), "should have metric data"); - - let mut found_duration = false; - for rm in &resource_metrics { - for sm in rm.scope_metrics() { - for metric in sm.metrics() { - if metric.name() == "db.client.operation.duration" { - found_duration = true; - assert_eq!(metric.unit(), "s"); - if let AggregatedMetrics::F64(MetricData::Histogram(hist)) = metric.data() { - let dp: Vec<_> = hist.data_points().collect(); - assert!(!dp.is_empty(), "histogram should have data points"); - assert!(dp[0].count() > 0, "data point count should be > 0"); - let has_system = dp[0] - .attributes() - .any(|kv| kv.key.as_str() == "db.system.name"); - assert!(has_system, "metric should have db.system.name attribute"); - } else { - panic!("db.client.operation.duration should be an f64 histogram"); - } - } - } - } - } - assert!( - found_duration, - "db.client.operation.duration metric not found" + let dp = $crate::common::find_duration_data_point(&resource_metrics) + .expect("db.client.operation.duration data point missing"); + + assert_eq!( + $crate::common::metric_attr(&dp, "db.response.status_code"), + Some(opentelemetry::Value::String($expected_code.into())), + "metric must carry db.response.status_code for sqlx::Error::Database", ); }}; } @@ -2079,6 +2407,8 @@ macro_rules! test_query_execute_many_with_annotations_via_pool { let spans = tel.spans(); assert_eq!(spans.len(), 1); $crate::common::assert_annotated_span(&spans[0], &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -2104,6 +2434,8 @@ macro_rules! test_query_fetch_with_annotations_via_pool { $crate::common::attr(&spans[0], "db.response.returned_rows"), Some(opentelemetry::Value::I64(2)) ); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -2126,6 +2458,8 @@ macro_rules! test_query_fetch_many_with_annotations_via_pool { let spans = tel.spans(); assert_eq!(spans.len(), 1); $crate::common::assert_annotated_span(&spans[0], &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -2151,6 +2485,8 @@ macro_rules! test_query_fetch_all_with_annotations_via_pool { $crate::common::attr(&spans[0], "db.response.returned_rows"), Some(opentelemetry::Value::I64(3)) ); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -2175,6 +2511,8 @@ macro_rules! test_query_fetch_one_with_annotations_via_pool { $crate::common::attr(&spans[0], "db.response.returned_rows"), Some(opentelemetry::Value::I64(1)) ); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -2196,6 +2534,8 @@ macro_rules! test_query_execute_with_annotations_records_error { assert_eq!(spans.len(), 1); $crate::common::assert_annotated_span(&spans[0], &$dialect); $crate::common::assert_error_span(&spans[0]); + + $crate::common::assert_annotated_error_metric(&tel, &$dialect); }}; } @@ -2217,6 +2557,8 @@ macro_rules! test_query_as_fetch_with_annotations_via_pool { let spans = tel.spans(); assert_eq!(spans.len(), 1); $crate::common::assert_annotated_span(&spans[0], &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -2239,6 +2581,8 @@ macro_rules! test_query_as_fetch_many_with_annotations_via_pool { let spans = tel.spans(); assert_eq!(spans.len(), 1); $crate::common::assert_annotated_span(&spans[0], &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -2274,6 +2618,8 @@ macro_rules! test_query_as_fetch_all_with_annotations_via_pool { for span in &spans { $crate::common::assert_annotated_span(span, &$dialect); } + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -2310,6 +2656,8 @@ macro_rules! test_query_as_fetch_one_with_annotations_via_pool { for span in &spans { $crate::common::assert_annotated_span(span, &$dialect); } + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -2346,6 +2694,8 @@ macro_rules! test_query_as_fetch_optional_with_annotations_via_pool { for span in &spans { $crate::common::assert_annotated_span(span, &$dialect); } + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -2367,6 +2717,8 @@ macro_rules! test_query_as_fetch_one_with_annotations_records_error { assert_eq!(spans.len(), 1); $crate::common::assert_annotated_span(&spans[0], &$dialect); $crate::common::assert_error_span(&spans[0]); + + $crate::common::assert_annotated_error_metric(&tel, &$dialect); }}; } @@ -2388,6 +2740,8 @@ macro_rules! test_query_scalar_fetch_with_annotations_via_pool { let spans = tel.spans(); assert_eq!(spans.len(), 1); $crate::common::assert_annotated_span(&spans[0], &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -2410,6 +2764,8 @@ macro_rules! test_query_scalar_fetch_many_with_annotations_via_pool { let spans = tel.spans(); assert_eq!(spans.len(), 1); $crate::common::assert_annotated_span(&spans[0], &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -2431,6 +2787,8 @@ macro_rules! test_query_scalar_fetch_all_with_annotations_via_pool { let spans = tel.spans(); assert_eq!(spans.len(), 1); $crate::common::assert_annotated_span(&spans[0], &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -2452,6 +2810,8 @@ macro_rules! test_query_scalar_fetch_one_with_annotations_via_pool { let spans = tel.spans(); assert_eq!(spans.len(), 1); $crate::common::assert_annotated_span(&spans[0], &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -2473,6 +2833,8 @@ macro_rules! test_query_scalar_fetch_optional_with_annotations_via_pool { let spans = tel.spans(); assert_eq!(spans.len(), 1); $crate::common::assert_annotated_span(&spans[0], &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -2504,6 +2866,8 @@ macro_rules! test_query_map_position_1_via_pool { let spans = tel.spans(); assert_eq!(spans.len(), 1); $crate::common::assert_annotated_span(&spans[0], &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -2527,6 +2891,8 @@ macro_rules! test_query_map_position_2_via_pool { let spans = tel.spans(); assert_eq!(spans.len(), 1); $crate::common::assert_annotated_span(&spans[0], &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -2550,6 +2916,8 @@ macro_rules! test_query_map_position_3_via_pool { let spans = tel.spans(); assert_eq!(spans.len(), 1); $crate::common::assert_annotated_span(&spans[0], &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -2573,6 +2941,8 @@ macro_rules! test_query_try_map_position_3_via_pool { let spans = tel.spans(); assert_eq!(spans.len(), 1); $crate::common::assert_annotated_span(&spans[0], &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -2600,6 +2970,8 @@ macro_rules! test_map_fetch_with_annotations_via_pool { $crate::common::attr(&spans[0], "db.response.returned_rows"), Some(opentelemetry::Value::I64(2)) ); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -2624,6 +2996,8 @@ macro_rules! test_map_fetch_many_with_annotations_via_pool { let spans = tel.spans(); assert_eq!(spans.len(), 1); $crate::common::assert_annotated_span(&spans[0], &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -2647,6 +3021,8 @@ macro_rules! test_map_fetch_all_with_annotations_via_pool { let spans = tel.spans(); assert_eq!(spans.len(), 1); $crate::common::assert_annotated_span(&spans[0], &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -2670,6 +3046,8 @@ macro_rules! test_map_fetch_one_with_annotations_via_pool { let spans = tel.spans(); assert_eq!(spans.len(), 1); $crate::common::assert_annotated_span(&spans[0], &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -2693,6 +3071,8 @@ macro_rules! test_map_fetch_optional_with_annotations_via_pool { let spans = tel.spans(); assert_eq!(spans.len(), 1); $crate::common::assert_annotated_span(&spans[0], &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -2717,6 +3097,8 @@ macro_rules! test_map_compose_after_annotations_via_pool { let spans = tel.spans(); assert_eq!(spans.len(), 1); $crate::common::assert_annotated_span(&spans[0], &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -2741,6 +3123,8 @@ macro_rules! test_map_try_map_compose_after_annotations_via_pool { let spans = tel.spans(); assert_eq!(spans.len(), 1); $crate::common::assert_annotated_span(&spans[0], &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -2765,6 +3149,8 @@ macro_rules! test_query_map_with_annotations_via_connection { let spans = tel.spans(); assert_eq!(spans.len(), 1); $crate::common::assert_annotated_span(&spans[0], &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -2790,6 +3176,8 @@ macro_rules! test_query_map_with_annotations_via_transaction { let spans = tel.spans(); assert_eq!(spans.len(), 1); $crate::common::assert_annotated_span(&spans[0], &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -2813,6 +3201,8 @@ macro_rules! test_query_map_with_annotations_records_error { assert_eq!(spans.len(), 1); $crate::common::assert_annotated_span(&spans[0], &$dialect); $crate::common::assert_error_span(&spans[0]); + + $crate::common::assert_annotated_error_metric(&tel, &$dialect); }}; } @@ -2839,6 +3229,8 @@ macro_rules! test_query_try_map_with_annotations_propagates_mapper_error { let spans = tel.spans(); assert_eq!(spans.len(), 1); $crate::common::assert_annotated_span(&spans[0], &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -2939,6 +3331,145 @@ macro_rules! test_builder_with_network_peer_address { }}; } +/// `PoolBuilder::with_pool_name` propagates `db.client.connection.pool.name` onto every +/// span and per-operation metric so dashboards can correlate pool-level signals (the +/// `db.client.connection.*` family) with query-level latency. +#[macro_export] +macro_rules! test_builder_with_pool_name_propagates_to_span_and_metric { + ($raw_pool_factory:expr, $dialect:expr) => {{ + use sqlx::Executor as _; + let _ = $dialect; + let tel = $crate::common::TestTelemetry::install(); + let raw = $raw_pool_factory; + let pool = sqlx_otel::PoolBuilder::from(raw) + .with_pool_name("primary-rw") + .build(); + + let _ = (&pool).fetch_optional("SELECT 1").await.unwrap(); + + let spans = tel.spans(); + assert_eq!(spans.len(), 1); + assert_eq!( + $crate::common::attr(&spans[0], "db.client.connection.pool.name"), + Some(opentelemetry::Value::String("primary-rw".into())), + "span must carry db.client.connection.pool.name set via with_pool_name", + ); + + let resource_metrics = tel.metrics(); + let dp = $crate::common::find_duration_data_point(&resource_metrics) + .expect("db.client.operation.duration data point missing"); + assert_eq!( + $crate::common::metric_attr(&dp, "db.client.connection.pool.name"), + Some(opentelemetry::Value::String("primary-rw".into())), + "metric must carry db.client.connection.pool.name set via with_pool_name", + ); + }}; +} + +/// Default `network.protocol.name` is the backend's wire protocol (Postgres / `MySQL`) +/// or absent (`SQLite`). The expected value is supplied per backend because `Dialect` +/// does not currently distinguish wire-protocol-bearing backends from embedded ones. +#[macro_export] +macro_rules! test_builder_default_network_protocol_name { + ($raw_pool_factory:expr, $expected:expr) => {{ + use sqlx::Executor as _; + let tel = $crate::common::TestTelemetry::install(); + let raw = $raw_pool_factory; + let pool = sqlx_otel::PoolBuilder::from(raw).build(); + + let _ = (&pool).fetch_optional("SELECT 1").await.unwrap(); + + let spans = tel.spans(); + assert_eq!(spans.len(), 1); + let actual_span = $crate::common::attr(&spans[0], "network.protocol.name"); + let expected: Option<&str> = $expected; + assert_eq!( + actual_span, + expected.map(|s| opentelemetry::Value::String(s.to_owned().into())), + "span network.protocol.name must match the backend default", + ); + + let resource_metrics = tel.metrics(); + let dp = $crate::common::find_duration_data_point(&resource_metrics) + .expect("db.client.operation.duration data point missing"); + let actual_metric = $crate::common::metric_attr(&dp, "network.protocol.name"); + assert_eq!( + actual_metric, + expected.map(|s| opentelemetry::Value::String(s.to_owned().into())), + "metric network.protocol.name must match the backend default", + ); + }}; +} + +/// `PoolBuilder::with_network_protocol_name` overrides the backend default and the value +/// surfaces on both spans and per-operation metrics. +#[macro_export] +macro_rules! test_builder_with_network_protocol_name_overrides { + ($raw_pool_factory:expr, $dialect:expr) => {{ + use sqlx::Executor as _; + let _ = $dialect; + let tel = $crate::common::TestTelemetry::install(); + let raw = $raw_pool_factory; + let pool = sqlx_otel::PoolBuilder::from(raw) + .with_network_protocol_name("custom-proto") + .build(); + + let _ = (&pool).fetch_optional("SELECT 1").await.unwrap(); + + let spans = tel.spans(); + assert_eq!(spans.len(), 1); + assert_eq!( + $crate::common::attr(&spans[0], "network.protocol.name"), + Some(opentelemetry::Value::String("custom-proto".into())), + "with_network_protocol_name must override the default on the span", + ); + + let resource_metrics = tel.metrics(); + let dp = $crate::common::find_duration_data_point(&resource_metrics) + .expect("db.client.operation.duration data point missing"); + assert_eq!( + $crate::common::metric_attr(&dp, "network.protocol.name"), + Some(opentelemetry::Value::String("custom-proto".into())), + "with_network_protocol_name must override the default on the metric", + ); + }}; +} + +/// `PoolBuilder::with_network_transport` propagates `network.transport` onto spans and +/// per-operation metrics. The wrapper does not infer transport from the connect string, +/// so the attribute reflects the deployment configuration the caller declared. +#[macro_export] +macro_rules! test_builder_with_network_transport { + ($raw_pool_factory:expr, $dialect:expr) => {{ + use sqlx::Executor as _; + let _ = $dialect; + let tel = $crate::common::TestTelemetry::install(); + let raw = $raw_pool_factory; + let pool = sqlx_otel::PoolBuilder::from(raw) + .with_network_transport("tcp") + .build(); + + let _ = (&pool).fetch_optional("SELECT 1").await.unwrap(); + + let spans = tel.spans(); + assert_eq!(spans.len(), 1); + assert_eq!( + $crate::common::attr(&spans[0], "network.transport"), + Some(opentelemetry::Value::String("tcp".into())), + "span must carry network.transport set via with_network_transport", + ); + + let resource_metrics = tel.metrics(); + let dp = $crate::common::find_duration_data_point(&resource_metrics) + .expect("db.client.operation.duration data point missing"); + assert_eq!( + $crate::common::metric_attr(&dp, "network.transport"), + Some(opentelemetry::Value::String("tcp".into())), + "metric must carry network.transport set via with_network_transport", + ); + }}; +} + /// `PoolBuilder::with_network_peer_port` populates `network.peer.port`. #[macro_export] macro_rules! test_builder_with_network_peer_port { @@ -3003,6 +3534,8 @@ macro_rules! test_query_text_mode_off_suppresses_sql { $crate::common::attr(&spans[0], "db.query.text").is_none(), "db.query.text should not be present when QueryTextMode::Off" ); + + $crate::common::assert_metric_for_system(&tel, $dialect.system); }}; } @@ -3041,6 +3574,7 @@ macro_rules! test_execute_records_affected_rows { Some(opentelemetry::Value::I64(3)), "inserting 3 rows should affect 3 rows" ); + $crate::common::assert_metric_for_system(&tel, $dialect.system); tel.reset(); // --- Upsert (dialect-specific) --- @@ -3094,6 +3628,7 @@ macro_rules! test_execute_records_affected_rows { Some(opentelemetry::Value::I64(0)), "deleting non-existent rows should affect 0 rows" ); + $crate::common::assert_metric_for_system(&tel, $dialect.system); }}; } @@ -3129,6 +3664,10 @@ macro_rules! test_transaction_rollback { #[macro_export] macro_rules! test_query_text_mode_obfuscated_replaces_literals { ($raw_pool_factory:expr, $dialect:expr) => {{ + // This macro intentionally narrows scope to span-side `db.query.text` capture; it + // does not assert on metrics because the pool is constructed before the test + // telemetry is installed, so the pool's metric instruments bind to the no-op meter. + // Per-method metric coverage lives in every other `test__*` macro. use sqlx::Executor as _; let _ = $dialect; let raw = $raw_pool_factory; @@ -3179,6 +3718,8 @@ macro_rules! test_fetch_optional_records_zero_rows { $crate::common::attr(&spans[0], "db.response.returned_rows"), Some(opentelemetry::Value::I64(0)) ); + + $crate::common::assert_metric_for_system(&tel, $dialect.system); }}; } @@ -3205,6 +3746,8 @@ macro_rules! test_query_bind_first_then_annotations_via_pool { let spans = tel.spans(); assert_eq!(spans.len(), 1); $crate::common::assert_annotated_span(&spans[0], &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -3231,6 +3774,8 @@ macro_rules! test_query_annotations_first_then_bind_via_pool { let spans = tel.spans(); assert_eq!(spans.len(), 1); $crate::common::assert_annotated_span(&spans[0], &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -3270,6 +3815,8 @@ macro_rules! test_query_execute_with_annotations_via_pool { $crate::common::assert_annotated_span(span, &$dialect); assert!($crate::common::attr(span, "db.response.affected_rows").is_some()); } + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -3308,6 +3855,8 @@ macro_rules! test_query_execute_with_annotations_via_connection { for span in &spans { $crate::common::assert_annotated_span(span, &$dialect); } + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -3347,6 +3896,8 @@ macro_rules! test_query_execute_with_annotations_via_transaction { for span in &spans { $crate::common::assert_annotated_span(span, &$dialect); } + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -3368,6 +3919,8 @@ macro_rules! test_query_with_operation_shorthand_via_pool { let spans = tel.spans(); assert_eq!(spans.len(), 1); $crate::common::assert_annotated_span(&spans[0], &$dialect); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -3394,6 +3947,8 @@ macro_rules! test_query_fetch_optional_with_annotations_via_pool { $crate::common::attr(&spans[0], "db.response.returned_rows"), Some(opentelemetry::Value::I64(0)) ); + + $crate::common::assert_annotated_metric(&tel, &$dialect); }}; } @@ -3407,7 +3962,13 @@ macro_rules! test_query_fetch_optional_with_annotations_via_pool { #[macro_export] macro_rules! test_executor_side_query_side_parity_via_pool { ($pool_factory:expr, $dialect:expr) => {{ + // This macro asserts span-side parity between the executor and query annotation + // surfaces. It does not assert on metrics because the pool is constructed before + // the test telemetry is installed, so the pool's metric instruments bind to the + // no-op meter. Per-method metric coverage lives in every other `test__*` + // macro. use sqlx_otel::QueryAnnotateExt as _; + let _ = $dialect; let pool = $pool_factory; let tel = $crate::common::TestTelemetry::install(); @@ -3463,7 +4024,5 @@ macro_rules! test_executor_side_query_side_parity_via_pool { let query_spans = tel.spans(); $crate::common::assert_span_parity("query_scalar::fetch_one", &exec_spans, &query_spans); - - let _ = $dialect; }}; } diff --git a/tests/mysql.rs b/tests/mysql.rs index c741822..b6ac0fa 100644 --- a/tests/mysql.rs +++ b/tests/mysql.rs @@ -467,11 +467,25 @@ async fn sqlstate_recorded_on_constraint_violation() { // =========================================================================== // Metrics // =========================================================================== +// Per-method coverage of `db.client.operation.duration` lives inside each +// `test__*` macro via `assert_metric_for_system` / `assert_annotated_metric` / +// `assert_error_metric`. The targeted SQLSTATE assertion below pins the backend-specific +// `db.response.status_code` value because that dimension varies per driver. #[tokio::test] #[serial] -async fn operation_duration_metric_is_recorded() { - test_operation_duration_metric_is_recorded!(test_pool().await, common::MYSQL_DIALECT); +async fn operation_duration_metric_carries_full_annotations() { + test_operation_duration_metric_carries_full_annotations!( + test_pool().await, + common::MYSQL_DIALECT + ); +} + +#[tokio::test] +#[serial] +async fn operation_duration_metric_carries_sqlstate() { + // MySQL SQLSTATE 42S02 = base table or view not found. + test_operation_duration_metric_carries_sqlstate!(test_pool().await, "42S02"); } // =========================================================================== @@ -518,6 +532,33 @@ async fn builder_with_network_peer_port() { test_builder_with_network_peer_port!(raw_pool().await, common::MYSQL_DIALECT); } +#[tokio::test] +#[serial] +async fn builder_with_pool_name_propagates_to_span_and_metric() { + test_builder_with_pool_name_propagates_to_span_and_metric!( + raw_pool().await, + common::MYSQL_DIALECT + ); +} + +#[tokio::test] +#[serial] +async fn builder_default_network_protocol_name_is_mysql() { + test_builder_default_network_protocol_name!(raw_pool().await, Some("mysql")); +} + +#[tokio::test] +#[serial] +async fn builder_with_network_protocol_name_overrides() { + test_builder_with_network_protocol_name_overrides!(raw_pool().await, common::MYSQL_DIALECT); +} + +#[tokio::test] +#[serial] +async fn builder_with_network_transport() { + test_builder_with_network_transport!(raw_pool().await, common::MYSQL_DIALECT); +} + // =========================================================================== // Pool close / is_closed // =========================================================================== diff --git a/tests/postgres.rs b/tests/postgres.rs index e1a3a78..3c75ce2 100644 --- a/tests/postgres.rs +++ b/tests/postgres.rs @@ -470,11 +470,25 @@ async fn sqlstate_recorded_on_constraint_violation() { // =========================================================================== // Metrics // =========================================================================== +// Per-method coverage of `db.client.operation.duration` lives inside each +// `test__*` macro via `assert_metric_for_system` / `assert_annotated_metric` / +// `assert_error_metric`. The targeted SQLSTATE assertion below pins the backend-specific +// `db.response.status_code` value because that dimension varies per driver. #[tokio::test] #[serial] -async fn operation_duration_metric_is_recorded() { - test_operation_duration_metric_is_recorded!(test_pool().await, common::POSTGRES_DIALECT); +async fn operation_duration_metric_carries_full_annotations() { + test_operation_duration_metric_carries_full_annotations!( + test_pool().await, + common::POSTGRES_DIALECT + ); +} + +#[tokio::test] +#[serial] +async fn operation_duration_metric_carries_sqlstate() { + // Postgres SQLSTATE 42P01 = undefined_table. + test_operation_duration_metric_carries_sqlstate!(test_pool().await, "42P01"); } // =========================================================================== @@ -521,6 +535,33 @@ async fn builder_with_network_peer_port() { test_builder_with_network_peer_port!(raw_pool().await, common::POSTGRES_DIALECT); } +#[tokio::test] +#[serial] +async fn builder_with_pool_name_propagates_to_span_and_metric() { + test_builder_with_pool_name_propagates_to_span_and_metric!( + raw_pool().await, + common::POSTGRES_DIALECT + ); +} + +#[tokio::test] +#[serial] +async fn builder_default_network_protocol_name_is_postgresql() { + test_builder_default_network_protocol_name!(raw_pool().await, Some("postgresql")); +} + +#[tokio::test] +#[serial] +async fn builder_with_network_protocol_name_overrides() { + test_builder_with_network_protocol_name_overrides!(raw_pool().await, common::POSTGRES_DIALECT); +} + +#[tokio::test] +#[serial] +async fn builder_with_network_transport() { + test_builder_with_network_transport!(raw_pool().await, common::POSTGRES_DIALECT); +} + // =========================================================================== // Pool close / is_closed // =========================================================================== diff --git a/tests/sqlite.rs b/tests/sqlite.rs index 2139423..ba26580 100644 --- a/tests/sqlite.rs +++ b/tests/sqlite.rs @@ -334,11 +334,26 @@ async fn describe_records_error() { // =========================================================================== // Metrics // =========================================================================== +// Per-method coverage of `db.client.operation.duration` lives inside each +// `test__*` macro via `assert_metric_for_system` / `assert_annotated_metric` / +// `assert_error_metric`. The targeted SQLSTATE assertion below pins the backend-specific +// `db.response.status_code` value because that dimension varies per driver and would +// otherwise need a per-backend value plumbed into every error-path macro. #[tokio::test] #[serial] -async fn operation_duration_metric_is_recorded() { - test_operation_duration_metric_is_recorded!(test_pool().await, common::SQLITE_DIALECT); +async fn operation_duration_metric_carries_full_annotations() { + test_operation_duration_metric_carries_full_annotations!( + test_pool().await, + common::SQLITE_DIALECT + ); +} + +#[tokio::test] +#[serial] +async fn operation_duration_metric_carries_sqlstate() { + // SQLite extended result code `1` = SQLITE_ERROR (returned for `no such table`). + test_operation_duration_metric_carries_sqlstate!(test_pool().await, "1"); } // =========================================================================== @@ -401,6 +416,36 @@ async fn builder_with_network_peer_port() { test_builder_with_network_peer_port!(raw_pool().await, common::SQLITE_DIALECT); } +#[tokio::test] +#[serial] +async fn builder_with_pool_name_propagates_to_span_and_metric() { + test_builder_with_pool_name_propagates_to_span_and_metric!( + raw_pool().await, + common::SQLITE_DIALECT + ); +} + +#[tokio::test] +#[serial] +async fn builder_default_network_protocol_name_absent_for_sqlite() { + // SQLite is an embedded backend with no wire protocol; default omits + // `network.protocol.name`. + let expected: Option<&str> = None; + test_builder_default_network_protocol_name!(raw_pool().await, expected); +} + +#[tokio::test] +#[serial] +async fn builder_with_network_protocol_name_overrides() { + test_builder_with_network_protocol_name_overrides!(raw_pool().await, common::SQLITE_DIALECT); +} + +#[tokio::test] +#[serial] +async fn builder_with_network_transport() { + test_builder_with_network_transport!(raw_pool().await, common::SQLITE_DIALECT); +} + // =========================================================================== // Pool close / is_closed // ===========================================================================