From a18a602d87ef2ee1c119cf52418dfceda0f51aef Mon Sep 17 00:00:00 2001 From: Patrick Lee Scott Date: Tue, 7 Jul 2026 02:37:52 -0500 Subject: [PATCH] feat: add direct publish telemetry Implements [[tasks/distributed-direct-publish-telemetry-1]] --- docs/metrics.md | 28 +++- docs/observability.md | 30 +++- docs/transports.md | 15 +- src/bus/bus.rs | 24 +++ src/bus/in_memory_bus.rs | 25 ++- src/bus/kafka_bus.rs | 37 ++++- src/bus/knative_bus.rs | 35 ++++- src/bus/mod.rs | 1 + src/bus/nats_bus.rs | 31 +++- src/bus/producer_telemetry.rs | 137 +++++++++++++++++ src/bus/rabbitmq_bus.rs | 37 ++++- src/bus/sql_bus_common.rs | 35 ++++- src/metrics.rs | 234 ++++++++++++++++++++++++++++- src/outbox_worker/bus_publisher.rs | 27 +++- src/telemetry.rs | 28 ++++ tests/metrics_exposition/main.rs | 75 ++++++++- tests/otel_export/main.rs | 100 ++++++++++++ 17 files changed, 871 insertions(+), 28 deletions(-) create mode 100644 src/bus/producer_telemetry.rs diff --git a/docs/metrics.md b/docs/metrics.md index 0c08aac2..90b35c45 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -56,7 +56,7 @@ Metric labels are intentionally bounded: - `status`: bounded dispatch status such as `success`, `unknown_command`, `decode_failed`, `guard_rejected`, `repository_error`, or `other_error`. - `transport`: built-in source label such as `in_memory`, `sqlite`, - `postgres`, `rabbitmq`, `kafka`, `nats`, or `outbox`. + `postgres`, `rabbitmq`, `kafka`, `nats`, or `knative`. - `outcome`: bounded settle/publish outcome such as `ack`, `nack`, `dead_letter`, `park`, `ignored`, `log_and_ack`, `published`, `released`, or `failed`. @@ -83,6 +83,12 @@ metrics should use that vocabulary rather than introducing ad hoc labels. counter. - `distributed_transport_failures_total{service,transport,failure_class,action}` counter. +- `distributed_transport_publish_total{service,transport,message_kind,outcome}` + counter for direct bus producer calls. `outcome` is `published` or `failed`. +- `distributed_transport_publish_duration_seconds{service,transport,message_kind,outcome}` + histogram for direct bus producer call duration. +- `distributed_transport_publish_failures_total{service,transport,message_kind,failure_class}` + counter for failed direct bus producer calls. - `distributed_outbox_messages_total{service,outcome}` counter. - `distributed_outbox_pending_messages{service}` gauge. - `distributed_outbox_oldest_pending_age_seconds{service}` gauge. @@ -100,10 +106,22 @@ does not install an OpenTelemetry metrics SDK, emit request-level HTTP metrics, or record user payload data. Direct transport receive paths emit receive/settle counters and failure -counters. Outbox dispatch emits publish outcomes and backlog gauges. Direct -`MessagePublisher` calls outside the outbox path do not emit publish metrics in -this release; instrumenting those calls should use the same bounded vocabulary -and should stay separate from outbox publish outcomes. +counters. Direct `Bus::send`, `Bus::publish`, `send_message`, and +`publish_message` calls on built-in buses emit `distributed_transport_publish_*` +metrics. A direct publish success means the adapter's durable publish threshold +resolved `Ok`: in-memory accepted the message, SQL committed the insert, +RabbitMQ confirmed the publish, Kafka acknowledged per `acks`, NATS JetStream +returned a publish ack, or Knative/HTTP returned a successful broker response. +If that threshold returns `Err`, the publish outcome is `failed` and the +failure counter uses the error's `TransportErrorKind` as `retryable` or +`permanent`. + +Outbox-derived publishes are intentionally separate. `BusOutboxPublishHook`, +`BusPublisher`, `DynBusPublisher`, and `OutboxDispatcher` record outbox +`published`, `released`, `failed`, and backlog metrics, but they use the buses' +raw outbox publish path so those rows do not increment direct producer metrics. +This avoids double-counting; outbox metrics and the `distributed.outbox.publish` +span remain the authoritative producer signal for outbox rows. ## Scaffolded GitOps diff --git a/docs/observability.md b/docs/observability.md index 8fc2db1a..1bff915b 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -60,9 +60,10 @@ metadata already stored with events. ## Optional Span Feature The `otel` feature adds framework-owned `tracing` spans around dispatch, -handler execution, transport receive, and outbox publish boundaries. When the -incoming message carries W3C `traceparent` / `tracestate`, Distributed extracts -that context and sets it as the OpenTelemetry parent for the framework span: +handler execution, direct transport publish, transport receive, and outbox +publish boundaries. When the incoming message carries W3C `traceparent` / +`tracestate`, Distributed extracts that context and sets it as the +OpenTelemetry parent for the framework span: ```toml [dependencies] @@ -79,6 +80,7 @@ Framework span names are intentionally bounded: - `distributed.microsvc.dispatch` - `distributed.handler` +- `distributed.transport.publish` - `distributed.transport.receive` - `distributed.outbox.publish` @@ -88,6 +90,17 @@ Each span uses the same framework-owned message attributes: - `distributed.message.kind` - `messaging.message.id` +Direct producer spans also include: + +- `distributed.transport.name` +- `distributed.bus.operation`: `send` or `publish` +- `distributed.producer.source`: `direct` + +Outbox rows keep using `distributed.outbox.publish` as their producer span and +do not emit `distributed.transport.publish`, so traces match the metrics +boundary: direct bus calls are direct producer telemetry; outbox dispatch is +outbox producer telemetry. + Future spans should use the same helper path so new attributes are reviewed in one place. Do not add payload fields, user ids, aggregate ids, or raw metadata values as framework span attributes. @@ -151,10 +164,13 @@ CI verifies the observability surface at the docker level (see `GET /metrics`, and lints the exposition with `promtool check metrics` (skips when `PROMTOOL` is unset). - `tests/otel_export` builds a real OTLP pipeline the way a service binary - would, dispatches a message carrying a W3C `traceparent`, and asserts a real - OpenTelemetry Collector received `distributed.microsvc.dispatch` parented to - the incoming span (skips when `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT` / - `OTEL_COLLECTOR_TRACES_FILE` are unset). + would, dispatches and directly publishes messages carrying W3C `traceparent`, + and asserts a real OpenTelemetry Collector received + `distributed.microsvc.dispatch` and `distributed.transport.publish` parented + to the incoming span. It also verifies active local spans remain the direct + parent for nested framework spans (skips when + `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT` / `OTEL_COLLECTOR_TRACES_FILE` are + unset). - The scaffolded `ServiceMonitor` / `PrometheusRule` / OTLP env output is rendered with `helm template` and validated against published CRD schemas with `kubeconform`. diff --git a/docs/transports.md b/docs/transports.md index 33c6b0e9..b07e13cb 100644 --- a/docs/transports.md +++ b/docs/transports.md @@ -75,9 +75,18 @@ metadata (codec, destination, source aggregate) is namespaced under the reserved Telemetry follows the same boundary. Consumer-side direct transports report receive/settle outcomes and classified failures. Producer-side outbox dispatch reports `published`, `released`, `failed`, and backlog gauges. A direct -`MessagePublisher` call outside the outbox path is not counted as an outbox -publish and currently has no direct publish metric; adding that should be a -separate producer telemetry path with its own bounded labels. +`Bus::send`, `Bus::publish`, `send_message`, or `publish_message` call on a +built-in bus reports `distributed_transport_publish_*` metrics and, with +`otel`, a `distributed.transport.publish` span. Direct publish success means the +same adapter durable publish threshold resolved `Ok`; a failed threshold records +`outcome="failed"` and a `retryable` or `permanent` failure class from +`TransportErrorKind`. + +Outbox-derived publishes use the buses' raw outbox path through +`BusPublisher`/`DynBusPublisher`, so they do not increment direct publish +metrics. That avoids double-counting and keeps `distributed_outbox_*` metrics +plus `distributed.outbox.publish` as the authoritative producer signal for +outbox rows. Trace context is normal metadata. Distributed preserves W3C `traceparent` and `tracestate` across `Message`, `EventRecord`, and `OutboxMessage` carriers diff --git a/src/bus/bus.rs b/src/bus/bus.rs index a897f47c..055ba5da 100644 --- a/src/bus/bus.rs +++ b/src/bus/bus.rs @@ -59,6 +59,30 @@ pub trait Bus: Send + Sync { &self, message: Message, ) -> impl Future> + Send; + + /// Send an outbox-derived command without direct producer telemetry. + /// + /// This default preserves downstream custom bus compatibility. Built-in + /// buses override it to bypass their direct producer instrumentation so the + /// outbox remains the authoritative producer signal for outbox rows. + fn send_outbox_message( + &self, + message: Message, + ) -> impl Future> + Send { + self.send_message(message) + } + + /// Publish an outbox-derived event without direct producer telemetry. + /// + /// See [`send_outbox_message`](Self::send_outbox_message) for why this + /// exists as a provided compatibility method instead of a required trait + /// method. + fn publish_outbox_message( + &self, + message: Message, + ) -> impl Future> + Send { + self.publish_message(message) + } } /// Consume side of the bus — pull transports that run a [`run_source`] loop. diff --git a/src/bus/in_memory_bus.rs b/src/bus/in_memory_bus.rs index 4e476255..a28e80d2 100644 --- a/src/bus/in_memory_bus.rs +++ b/src/bus/in_memory_bus.rs @@ -13,6 +13,7 @@ use std::collections::{HashMap, VecDeque}; use std::sync::{Arc, Mutex}; +use super::producer_telemetry::{record_direct_publish, BusOperation}; use super::source::{MessageSource, ReceivedMessage}; use super::Message; use super::{run_source, Bus, BusConsumer, MessageRouter, RunOptions, TransportError}; @@ -62,10 +63,32 @@ impl InMemoryBus { impl Bus for InMemoryBus { async fn send_message(&self, message: Message) -> Result<(), TransportError> { - self.enqueue(message) + record_direct_publish( + None, + "in_memory", + BusOperation::Send, + message, + |message| async { self.enqueue(message) }, + ) + .await } async fn publish_message(&self, message: Message) -> Result<(), TransportError> { + record_direct_publish( + None, + "in_memory", + BusOperation::Publish, + message, + |message| async { self.append(message) }, + ) + .await + } + + async fn send_outbox_message(&self, message: Message) -> Result<(), TransportError> { + self.enqueue(message) + } + + async fn publish_outbox_message(&self, message: Message) -> Result<(), TransportError> { self.append(message) } } diff --git a/src/bus/kafka_bus.rs b/src/bus/kafka_bus.rs index d21320e8..718b6db9 100644 --- a/src/bus/kafka_bus.rs +++ b/src/bus/kafka_bus.rs @@ -24,6 +24,7 @@ use std::sync::Arc; use std::time::Duration; use super::kafka::{KafkaPublisher, KafkaSource}; +use super::producer_telemetry::{record_direct_publish, BusOperation}; use super::{ run_source, Bus, BusConsumer, BusTopologyConfig, MessagePublisher, MessageRouter, RunOptions, TransportError, @@ -185,13 +186,45 @@ impl KafkaBus { } impl Bus for KafkaBus { - async fn send_message(&self, mut message: Message) -> Result<(), TransportError> { + async fn send_message(&self, message: Message) -> Result<(), TransportError> { + record_direct_publish( + None, + "kafka", + BusOperation::Send, + message, + |message| async { self.send_message_raw(message).await }, + ) + .await + } + + async fn publish_message(&self, message: Message) -> Result<(), TransportError> { + record_direct_publish( + None, + "kafka", + BusOperation::Publish, + message, + |message| async { self.publish_message_raw(message).await }, + ) + .await + } + + async fn send_outbox_message(&self, message: Message) -> Result<(), TransportError> { + self.send_message_raw(message).await + } + + async fn publish_outbox_message(&self, message: Message) -> Result<(), TransportError> { + self.publish_message_raw(message).await + } +} + +impl KafkaBus { + async fn send_message_raw(&self, mut message: Message) -> Result<(), TransportError> { // The publisher uses the message name as the topic; namespace it. message.name = format!("{}{}", self.command_prefix()?, message.name); self.publisher.publish(message).await } - async fn publish_message(&self, mut message: Message) -> Result<(), TransportError> { + async fn publish_message_raw(&self, mut message: Message) -> Result<(), TransportError> { message.name = format!("{}{}", self.event_prefix()?, message.name); self.publisher.publish(message).await } diff --git a/src/bus/knative_bus.rs b/src/bus/knative_bus.rs index ab08e17b..64ff1865 100644 --- a/src/bus/knative_bus.rs +++ b/src/bus/knative_bus.rs @@ -23,6 +23,7 @@ use std::time::Duration; use std::collections::HashSet; use super::knative::unique_k8s_name; +use super::producer_telemetry::{record_direct_publish, BusOperation}; use super::{Bus, TransportError}; use super::{Message, SubscriptionPlan}; @@ -223,10 +224,42 @@ impl KnativeBus { impl Bus for KnativeBus { async fn send_message(&self, message: Message) -> Result<(), TransportError> { - self.post_cloud_event(&self.commands_broker, &message).await + record_direct_publish( + None, + "knative", + BusOperation::Send, + message, + |message| async { self.send_message_raw(message).await }, + ) + .await } async fn publish_message(&self, message: Message) -> Result<(), TransportError> { + record_direct_publish( + None, + "knative", + BusOperation::Publish, + message, + |message| async { self.publish_message_raw(message).await }, + ) + .await + } + + async fn send_outbox_message(&self, message: Message) -> Result<(), TransportError> { + self.send_message_raw(message).await + } + + async fn publish_outbox_message(&self, message: Message) -> Result<(), TransportError> { + self.publish_message_raw(message).await + } +} + +impl KnativeBus { + async fn send_message_raw(&self, message: Message) -> Result<(), TransportError> { + self.post_cloud_event(&self.commands_broker, &message).await + } + + async fn publish_message_raw(&self, message: Message) -> Result<(), TransportError> { self.post_cloud_event(&self.events_broker, &message).await } } diff --git a/src/bus/mod.rs b/src/bus/mod.rs index 7ee59ec6..b096b1b2 100644 --- a/src/bus/mod.rs +++ b/src/bus/mod.rs @@ -103,6 +103,7 @@ mod nats; mod nats_bus; #[cfg(feature = "postgres")] mod postgres_bus; +mod producer_telemetry; mod publisher; #[cfg(feature = "rabbitmq")] mod rabbitmq; diff --git a/src/bus/nats_bus.rs b/src/bus/nats_bus.rs index c9ae5d5c..f4ce80a0 100644 --- a/src/bus/nats_bus.rs +++ b/src/bus/nats_bus.rs @@ -28,6 +28,7 @@ use async_nats::jetstream::consumer::pull::Config as PullConfig; use async_nats::jetstream::stream::{Config as StreamConfig, Stream}; use super::nats::{NatsJetStreamSource, NatsPublisher}; +use super::producer_telemetry::{record_direct_publish, BusOperation}; use super::{ retryable, run_source, Bus, BusConsumer, BusTopologyConfig, MessagePublisher, MessageRouter, RunOptions, TransportError, @@ -270,11 +271,39 @@ impl NatsBus { impl Bus for NatsBus { async fn send_message(&self, message: Message) -> Result<(), TransportError> { + record_direct_publish(None, "nats", BusOperation::Send, message, |message| async { + self.send_message_raw(message).await + }) + .await + } + + async fn publish_message(&self, message: Message) -> Result<(), TransportError> { + record_direct_publish( + None, + "nats", + BusOperation::Publish, + message, + |message| async { self.publish_message_raw(message).await }, + ) + .await + } + + async fn send_outbox_message(&self, message: Message) -> Result<(), TransportError> { + self.send_message_raw(message).await + } + + async fn publish_outbox_message(&self, message: Message) -> Result<(), TransportError> { + self.publish_message_raw(message).await + } +} + +impl NatsBus { + async fn send_message_raw(&self, message: Message) -> Result<(), TransportError> { self.validated_namespace()?; self.cmd_publisher.publish(message).await } - async fn publish_message(&self, message: Message) -> Result<(), TransportError> { + async fn publish_message_raw(&self, message: Message) -> Result<(), TransportError> { self.validated_namespace()?; self.evt_publisher.publish(message).await } diff --git a/src/bus/producer_telemetry.rs b/src/bus/producer_telemetry.rs new file mode 100644 index 00000000..a22d8e5e --- /dev/null +++ b/src/bus/producer_telemetry.rs @@ -0,0 +1,137 @@ +use std::future::Future; + +#[cfg(feature = "metrics")] +use std::time::{Duration, Instant}; + +use super::{Message, TransportError}; + +#[cfg(feature = "metrics")] +use super::MessageKind; + +#[derive(Clone, Copy, Debug)] +pub(crate) enum BusOperation { + Send, + Publish, +} + +impl BusOperation { + #[cfg(feature = "otel")] + fn as_str(self) -> &'static str { + match self { + Self::Send => "send", + Self::Publish => "publish", + } + } +} + +pub(crate) async fn record_direct_publish( + service: Option<&str>, + transport: &'static str, + operation: BusOperation, + message: Message, + publish: F, +) -> Result<(), TransportError> +where + F: FnOnce(Message) -> Fut, + Fut: Future> + Send, +{ + #[cfg(not(feature = "metrics"))] + let _ = service; + #[cfg(not(feature = "otel"))] + let _ = operation; + #[cfg(not(any(feature = "metrics", feature = "otel")))] + let _ = transport; + + #[cfg(feature = "metrics")] + let started = Instant::now(); + #[cfg(feature = "metrics")] + let kind = message.kind; + + #[cfg(feature = "otel")] + let result = { + use tracing::Instrument as _; + + let span = + crate::telemetry::transport_publish_span(transport, operation.as_str(), &message); + crate::trace_context::set_span_parent_from_metadata_if_no_current_span( + &span, + &message.metadata, + ); + publish(message).instrument(span).await + }; + + #[cfg(not(feature = "otel"))] + let result = publish(message).await; + + #[cfg(feature = "metrics")] + record_direct_publish_result(service, transport, kind, started.elapsed(), &result); + + result +} + +#[cfg(feature = "metrics")] +fn record_direct_publish_result( + service: Option<&str>, + transport: &str, + kind: MessageKind, + duration: Duration, + result: &Result<(), TransportError>, +) { + let outcome = match result { + Ok(()) => crate::telemetry::transport_publish_outcome::PUBLISHED, + Err(_) => crate::telemetry::transport_publish_outcome::FAILED, + }; + crate::metrics::record_transport_publish(service, transport, kind, outcome, duration); + + if let Err(error) = result { + crate::metrics::record_transport_publish_failure( + service, + transport, + kind, + crate::telemetry::transport_failure_class(error.kind()), + ); + } +} + +#[cfg(all(test, feature = "metrics"))] +mod tests { + use super::*; + use crate::bus::MessageKind; + + #[tokio::test] + async fn failed_direct_publish_records_outcome_and_failure_class_without_ids() { + let _guard = crate::metrics::async_lock_for_tests().await; + crate::metrics::reset_for_tests(); + + let message = Message::new("secret.direct.event", MessageKind::Event, b"{}".to_vec()) + .with_id("msg-secret-1") + .with_metadata( + "traceparent", + "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01", + ); + + let result = record_direct_publish( + Some("orders"), + "in_memory", + BusOperation::Publish, + message, + |_message| async { Err(TransportError::permanent("broker rejected")) }, + ) + .await; + assert!(result.is_err()); + + let text = crate::metrics::prometheus_text(); + assert!(text.contains( + "distributed_transport_publish_total{service=\"orders\",transport=\"in_memory\",message_kind=\"event\",outcome=\"failed\"} 1" + )); + assert!(text.contains( + "distributed_transport_publish_duration_seconds_count{service=\"orders\",transport=\"in_memory\",message_kind=\"event\",outcome=\"failed\"} 1" + )); + assert!(text.contains( + "distributed_transport_publish_failures_total{service=\"orders\",transport=\"in_memory\",message_kind=\"event\",failure_class=\"permanent\"} 1" + )); + assert!(!text.contains("secret.direct.event")); + assert!(!text.contains("msg-secret-1")); + assert!(!text.contains("4bf92f3577b34da6a3ce929d0e0e4736")); + } +} diff --git a/src/bus/rabbitmq_bus.rs b/src/bus/rabbitmq_bus.rs index f62f906b..2af432cc 100644 --- a/src/bus/rabbitmq_bus.rs +++ b/src/bus/rabbitmq_bus.rs @@ -34,6 +34,7 @@ use lapin::options::{ use lapin::types::{FieldTable, ShortString}; use lapin::{Channel, ExchangeKind}; +use super::producer_telemetry::{record_direct_publish, BusOperation}; use super::rabbitmq::{connect_channel, message_properties, RabbitSource}; use super::{ retryable, run_source, Bus, BusConsumer, BusTopologyConfig, MessageRouter, RunOptions, @@ -313,7 +314,39 @@ impl RabbitBus { } impl Bus for RabbitBus { - async fn send_message(&self, mut message: Message) -> Result<(), TransportError> { + async fn send_message(&self, message: Message) -> Result<(), TransportError> { + record_direct_publish( + None, + "rabbitmq", + BusOperation::Send, + message, + |message| async { self.send_message_raw(message).await }, + ) + .await + } + + async fn publish_message(&self, message: Message) -> Result<(), TransportError> { + record_direct_publish( + None, + "rabbitmq", + BusOperation::Publish, + message, + |message| async { self.publish_message_raw(message).await }, + ) + .await + } + + async fn send_outbox_message(&self, message: Message) -> Result<(), TransportError> { + self.send_message_raw(message).await + } + + async fn publish_outbox_message(&self, message: Message) -> Result<(), TransportError> { + self.publish_message_raw(message).await + } +} + +impl RabbitBus { + async fn send_message_raw(&self, mut message: Message) -> Result<(), TransportError> { checked_name(message.name())?; // Default exchange routes by routing key == queue name; declare the queue // (once per process) so the command is retained until a listener consumes it. @@ -323,7 +356,7 @@ impl Bus for RabbitBus { self.publish_confirmed("", &queue, &message).await } - async fn publish_message(&self, message: Message) -> Result<(), TransportError> { + async fn publish_message_raw(&self, message: Message) -> Result<(), TransportError> { checked_name(message.name())?; let exchange = self.events_exchange()?; self.ensure_events_exchange(&exchange).await?; diff --git a/src/bus/sql_bus_common.rs b/src/bus/sql_bus_common.rs index 0497b02e..32711295 100644 --- a/src/bus/sql_bus_common.rs +++ b/src/bus/sql_bus_common.rs @@ -40,6 +40,7 @@ use sqlx::{ColumnIndex, Decode, Row, Type}; use crate::sqlx_repo::is_sqlx_transient; +use super::producer_telemetry::{record_direct_publish, BusOperation}; use super::source::{MessageSource, ReceivedMessage}; use super::{ run_source, Bus, BusConsumer, BusTopologyConfig, MessageRouter, RunOptions, TransportError, @@ -315,10 +316,42 @@ impl SqlBus { impl Bus for SqlBus { async fn send_message(&self, message: Message) -> Result<(), TransportError> { - self.dialect.insert_queue(&message).await + record_direct_publish( + None, + B::BACKEND, + BusOperation::Send, + message, + |message| async { self.send_message_raw(message).await }, + ) + .await } async fn publish_message(&self, message: Message) -> Result<(), TransportError> { + record_direct_publish( + None, + B::BACKEND, + BusOperation::Publish, + message, + |message| async { self.publish_message_raw(message).await }, + ) + .await + } + + async fn send_outbox_message(&self, message: Message) -> Result<(), TransportError> { + self.send_message_raw(message).await + } + + async fn publish_outbox_message(&self, message: Message) -> Result<(), TransportError> { + self.publish_message_raw(message).await + } +} + +impl SqlBus { + async fn send_message_raw(&self, message: Message) -> Result<(), TransportError> { + self.dialect.insert_queue(&message).await + } + + async fn publish_message_raw(&self, message: Message) -> Result<(), TransportError> { self.dialect.insert_log(&message).await } } diff --git a/src/metrics.rs b/src/metrics.rs index 799b5401..01e180af 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -36,6 +36,18 @@ const TRANSPORT_FAILURES_TOTAL_FAMILY: MetricFamily = MetricFamily::counter( metric_names::TRANSPORT_FAILURES_TOTAL, "Total transport failures by class and chosen action.", ); +const TRANSPORT_PUBLISH_TOTAL_FAMILY: MetricFamily = MetricFamily::counter( + metric_names::TRANSPORT_PUBLISH_TOTAL, + "Total direct transport publish outcomes.", +); +const TRANSPORT_PUBLISH_DURATION_FAMILY: MetricFamily = MetricFamily::histogram( + metric_names::TRANSPORT_PUBLISH_DURATION_SECONDS, + "Direct transport publish duration in seconds.", +); +const TRANSPORT_PUBLISH_FAILURES_TOTAL_FAMILY: MetricFamily = MetricFamily::counter( + metric_names::TRANSPORT_PUBLISH_FAILURES_TOTAL, + "Total direct transport publish failures by class.", +); const OUTBOX_MESSAGES_TOTAL_FAMILY: MetricFamily = MetricFamily::counter( metric_names::OUTBOX_MESSAGES_TOTAL, "Total outbox publish outcomes.", @@ -107,6 +119,38 @@ pub fn record_transport_failure( }); } +/// Record one direct producer publish attempt. +pub fn record_transport_publish( + service: Option<&str>, + transport: &str, + kind: MessageKind, + outcome: &str, + duration: Duration, +) { + registry().record_transport_publish(TransportPublishKey { + service: service_label(service), + transport: transport.to_string(), + message_kind: kind.as_str().to_string(), + outcome: outcome.to_string(), + duration_seconds: duration.as_secs_f64(), + }); +} + +/// Record a classified direct producer publish failure. +pub fn record_transport_publish_failure( + service: Option<&str>, + transport: &str, + kind: MessageKind, + failure_class: &str, +) { + registry().record_transport_publish_failure(TransportPublishFailureKey { + service: service_label(service), + transport: transport.to_string(), + message_kind: kind.as_str().to_string(), + failure_class: failure_class.to_string(), + }); +} + /// Record one outbox dispatch state transition. pub fn record_outbox_message(service: Option<&str>, outcome: &str) { record_outbox_messages(service, outcome, 1); @@ -392,6 +436,9 @@ struct MetricsRegistry { dispatch_duration: Mutex>, transport_messages_total: Mutex>, transport_failures_total: Mutex>, + transport_publish_total: Mutex>, + transport_publish_duration: Mutex>, + transport_publish_failures_total: Mutex>, outbox_messages_total: Mutex>, outbox_pending_messages: Mutex>, outbox_oldest_pending_age_seconds: Mutex>, @@ -433,6 +480,28 @@ impl MetricsRegistry { self.note_service(service); } + fn record_transport_publish(&self, key: TransportPublishKey) { + let service = key.service.clone(); + self.lock(&self.transport_publish_total) + .entry(key.counter_key()) + .and_modify(|value| *value += 1) + .or_insert(1); + self.lock(&self.transport_publish_duration) + .entry(key.histogram_key()) + .or_insert_with(Histogram::new) + .observe(key.duration_seconds); + self.note_service(service); + } + + fn record_transport_publish_failure(&self, key: TransportPublishFailureKey) { + let service = key.service.clone(); + self.lock(&self.transport_publish_failures_total) + .entry(key) + .and_modify(|value| *value += 1) + .or_insert(1); + self.note_service(service); + } + fn record_outbox_messages(&self, key: OutboxMessageKey, count: u64) { let service = key.service.clone(); self.lock(&self.outbox_messages_total) @@ -461,6 +530,10 @@ impl MetricsRegistry { let dispatch_duration = self.clone_locked(&self.dispatch_duration); let transport_messages_total = self.clone_locked(&self.transport_messages_total); let transport_failures_total = self.clone_locked(&self.transport_failures_total); + let transport_publish_total = self.clone_locked(&self.transport_publish_total); + let transport_publish_duration = self.clone_locked(&self.transport_publish_duration); + let transport_publish_failures_total = + self.clone_locked(&self.transport_publish_failures_total); let outbox_messages_total = self.clone_locked(&self.outbox_messages_total); let outbox_pending_messages = self.clone_locked(&self.outbox_pending_messages); let outbox_oldest_pending_age_seconds = @@ -509,6 +582,24 @@ impl MetricsRegistry { .map(|(key, value)| MetricSample::counter(key.labels(), *value)) .collect(), ), + TRANSPORT_PUBLISH_TOTAL_FAMILY.snapshot( + transport_publish_total + .iter() + .map(|(key, value)| MetricSample::counter(key.labels(), *value)) + .collect(), + ), + TRANSPORT_PUBLISH_DURATION_FAMILY.snapshot( + transport_publish_duration + .iter() + .map(|(key, histogram)| MetricSample::histogram(key.labels(), histogram)) + .collect(), + ), + TRANSPORT_PUBLISH_FAILURES_TOTAL_FAMILY.snapshot( + transport_publish_failures_total + .iter() + .map(|(key, value)| MetricSample::counter(key.labels(), *value)) + .collect(), + ), OUTBOX_MESSAGES_TOTAL_FAMILY.snapshot( outbox_messages_total .iter() @@ -542,6 +633,9 @@ impl MetricsRegistry { self.lock(&self.dispatch_duration).clear(); self.lock(&self.transport_messages_total).clear(); self.lock(&self.transport_failures_total).clear(); + self.lock(&self.transport_publish_total).clear(); + self.lock(&self.transport_publish_duration).clear(); + self.lock(&self.transport_publish_failures_total).clear(); self.lock(&self.outbox_messages_total).clear(); self.lock(&self.outbox_pending_messages).clear(); self.lock(&self.outbox_oldest_pending_age_seconds).clear(); @@ -679,6 +773,98 @@ impl TransportFailureKey { } } +#[derive(Clone)] +struct TransportPublishKey { + service: String, + transport: String, + message_kind: String, + outcome: String, + duration_seconds: f64, +} + +impl TransportPublishKey { + fn counter_key(&self) -> TransportPublishCounterKey { + TransportPublishCounterKey { + service: self.service.clone(), + transport: self.transport.clone(), + message_kind: self.message_kind.clone(), + outcome: self.outcome.clone(), + } + } + + fn histogram_key(&self) -> TransportPublishHistogramKey { + TransportPublishHistogramKey { + service: self.service.clone(), + transport: self.transport.clone(), + message_kind: self.message_kind.clone(), + outcome: self.outcome.clone(), + } + } +} + +#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)] +struct TransportPublishCounterKey { + service: String, + transport: String, + message_kind: String, + outcome: String, +} + +impl TransportPublishCounterKey { + fn labels(&self) -> Vec<(String, String)> { + transport_publish_labels( + &self.service, + &self.transport, + &self.message_kind, + &self.outcome, + ) + } +} + +#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)] +struct TransportPublishHistogramKey { + service: String, + transport: String, + message_kind: String, + outcome: String, +} + +impl TransportPublishHistogramKey { + fn labels(&self) -> Vec<(String, String)> { + transport_publish_labels( + &self.service, + &self.transport, + &self.message_kind, + &self.outcome, + ) + } +} + +#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)] +struct TransportPublishFailureKey { + service: String, + transport: String, + message_kind: String, + failure_class: String, +} + +impl TransportPublishFailureKey { + fn labels(&self) -> Vec<(String, String)> { + vec![ + (metric_labels::SERVICE.to_string(), self.service.clone()), + (metric_labels::TRANSPORT.to_string(), self.transport.clone()), + ( + metric_labels::MESSAGE_KIND.to_string(), + self.message_kind.clone(), + ), + ( + metric_labels::FAILURE_CLASS.to_string(), + self.failure_class.clone(), + ), + ] + } +} + #[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)] struct OutboxMessageKey { service: String, @@ -725,6 +911,23 @@ fn service_labels(service: &str) -> Vec<(String, String)> { vec![(metric_labels::SERVICE.to_string(), service.to_string())] } +fn transport_publish_labels( + service: &str, + transport: &str, + message_kind: &str, + outcome: &str, +) -> Vec<(String, String)> { + vec![ + (metric_labels::SERVICE.to_string(), service.to_string()), + (metric_labels::TRANSPORT.to_string(), transport.to_string()), + ( + metric_labels::MESSAGE_KIND.to_string(), + message_kind.to_string(), + ), + (metric_labels::OUTCOME.to_string(), outcome.to_string()), + ] +} + fn render_prometheus(snapshot: &MetricsSnapshot) -> String { let mut output = String::new(); for family_snapshot in snapshot.families() { @@ -847,7 +1050,7 @@ mod tests { use super::*; use crate::telemetry::{ dispatch_status, failure_action, failure_class, metric_labels, metric_names, - outbox_outcome, transport_outcome, + outbox_outcome, transport_outcome, transport_publish_outcome, }; use std::collections::BTreeSet; @@ -921,6 +1124,19 @@ mod tests { failure_class::RETRYABLE, failure_action::NACK, ); + record_transport_publish( + Some("orders"), + "nats", + MessageKind::Event, + transport_publish_outcome::PUBLISHED, + Duration::from_millis(9), + ); + record_transport_publish_failure( + Some("orders"), + "nats", + MessageKind::Command, + failure_class::PERMANENT, + ); record_outbox_messages(Some("orders"), outbox_outcome::PUBLISHED, 2); set_outbox_backlog(Some("orders"), 3, Some(Duration::from_secs(4))); @@ -939,6 +1155,9 @@ mod tests { metric_names::MICROSVC_DISPATCH_DURATION_SECONDS, metric_names::TRANSPORT_MESSAGES_TOTAL, metric_names::TRANSPORT_FAILURES_TOTAL, + metric_names::TRANSPORT_PUBLISH_TOTAL, + metric_names::TRANSPORT_PUBLISH_DURATION_SECONDS, + metric_names::TRANSPORT_PUBLISH_FAILURES_TOTAL, metric_names::OUTBOX_MESSAGES_TOTAL, metric_names::OUTBOX_PENDING_MESSAGES, metric_names::OUTBOX_OLDEST_PENDING_AGE_SECONDS, @@ -998,6 +1217,19 @@ mod tests { failure_class::PERMANENT, failure_action::DEAD_LETTER, ); + record_transport_publish( + Some("orders"), + "rabbitmq", + MessageKind::Command, + transport_publish_outcome::FAILED, + Duration::from_millis(5), + ); + record_transport_publish_failure( + Some("orders"), + "rabbitmq", + MessageKind::Command, + failure_class::PERMANENT, + ); record_outbox_messages(Some("orders"), outbox_outcome::RELEASED, 1); set_outbox_backlog(Some("orders"), 1, Some(Duration::from_secs(30))); diff --git a/src/outbox_worker/bus_publisher.rs b/src/outbox_worker/bus_publisher.rs index f1d03631..45a46eda 100644 --- a/src/outbox_worker/bus_publisher.rs +++ b/src/outbox_worker/bus_publisher.rs @@ -44,8 +44,8 @@ impl Clone for BusPublisher { impl MessagePublisher for BusPublisher { async fn publish(&self, message: Message) -> Result<(), TransportError> { match message.kind { - MessageKind::Command => self.bus.send_message(message).await, - MessageKind::Event => self.bus.publish_message(message).await, + MessageKind::Command => self.bus.send_outbox_message(message).await, + MessageKind::Event => self.bus.publish_outbox_message(message).await, } } } @@ -121,4 +121,27 @@ mod tests { ] ); } + + #[cfg(feature = "metrics")] + #[test] + fn built_in_bus_publisher_uses_outbox_raw_path_without_direct_metrics() { + let _guard = crate::metrics::lock_for_tests(); + crate::metrics::reset_for_tests(); + + let bus = Arc::new(crate::bus::InMemoryBus::new()); + let publisher = BusPublisher::new(bus); + + block_on(publisher.publish(Message::new( + "order.shipped", + MessageKind::Event, + b"{}".to_vec(), + ))) + .unwrap(); + + let text = crate::metrics::prometheus_text(); + assert!( + !text.contains("distributed_transport_publish_total{"), + "outbox-derived BusPublisher publishes must not record direct publish metrics:\n{text}" + ); + } } diff --git a/src/telemetry.rs b/src/telemetry.rs index bff32be1..3030d1eb 100644 --- a/src/telemetry.rs +++ b/src/telemetry.rs @@ -21,6 +21,11 @@ pub(crate) mod metric_names { "distributed_microsvc_dispatch_duration_seconds"; pub(crate) const TRANSPORT_MESSAGES_TOTAL: &str = "distributed_transport_messages_total"; pub(crate) const TRANSPORT_FAILURES_TOTAL: &str = "distributed_transport_failures_total"; + pub(crate) const TRANSPORT_PUBLISH_TOTAL: &str = "distributed_transport_publish_total"; + pub(crate) const TRANSPORT_PUBLISH_DURATION_SECONDS: &str = + "distributed_transport_publish_duration_seconds"; + pub(crate) const TRANSPORT_PUBLISH_FAILURES_TOTAL: &str = + "distributed_transport_publish_failures_total"; pub(crate) const OUTBOX_MESSAGES_TOTAL: &str = "distributed_outbox_messages_total"; pub(crate) const OUTBOX_PENDING_MESSAGES: &str = "distributed_outbox_pending_messages"; pub(crate) const OUTBOX_OLDEST_PENDING_AGE_SECONDS: &str = @@ -69,6 +74,12 @@ pub(crate) mod transport_outcome { pub(crate) const LOG_AND_ACK: &str = "log_and_ack"; } +#[cfg(feature = "metrics")] +pub(crate) mod transport_publish_outcome { + pub(crate) const PUBLISHED: &str = "published"; + pub(crate) const FAILED: &str = "failed"; +} + #[cfg(feature = "metrics")] pub(crate) mod failure_class { pub(crate) const RETRYABLE: &str = "retryable"; @@ -229,3 +240,20 @@ pub(crate) fn transport_receive_span(message: &Message) -> tracing::Span { pub(crate) fn outbox_publish_span(message: &Message) -> tracing::Span { framework_message_span!("distributed.outbox.publish", message) } + +#[cfg(feature = "otel")] +pub(crate) fn transport_publish_span( + transport: &str, + operation: &str, + message: &Message, +) -> tracing::Span { + tracing::info_span!( + "distributed.transport.publish", + distributed.transport.name = %transport, + distributed.bus.operation = %operation, + distributed.message.name = %message.name(), + distributed.message.kind = %message.kind.as_str(), + messaging.message.id = %message.id().unwrap_or(""), + distributed.producer.source = "direct", + ) +} diff --git a/tests/metrics_exposition/main.rs b/tests/metrics_exposition/main.rs index 1ca102cf..b7f0b6c5 100644 --- a/tests/metrics_exposition/main.rs +++ b/tests/metrics_exposition/main.rs @@ -18,10 +18,15 @@ use std::process::{Command, Stdio}; use std::sync::{Arc, Mutex}; use std::time::Duration; -use distributed::bus::{MessagePublisher, TransportError}; +use distributed::bus::{ + Bus, InMemoryBus, Message as BusMessage, MessageKind as BusMessageKind, MessagePublisher, + TransportError, +}; use distributed::microsvc::{self, Context, Message, Routes, Service}; use distributed::outbox_worker::OutboxDispatcher; -use distributed::{CommitBatch, InMemoryRepository, OutboxMessage, TransactionalCommit}; +use distributed::{ + BusPublisher, CommitBatch, InMemoryRepository, OutboxMessage, TransactionalCommit, +}; use serde_json::json; #[path = "../support/env.rs"] @@ -61,6 +66,30 @@ async fn spawn_http_service(name: &str) -> String { format!("http://{addr}") } +/// Drive direct producer traffic through the in-memory bus. The message names, +/// ids, and trace ids are intentionally unique so the exposition assertions can +/// prove direct publish metrics do not use high-cardinality labels. +async fn drive_direct_bus() { + let bus = InMemoryBus::new(); + bus.send("metrics.direct.command", b"{}".to_vec()) + .await + .unwrap(); + bus.publish_message( + BusMessage::new( + "metrics.direct.event", + BusMessageKind::Event, + b"{}".to_vec(), + ) + .with_id("direct-message-id") + .with_metadata( + "traceparent", + "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01", + ), + ) + .await + .unwrap(); +} + /// Stage outbox rows through a commit and drain them through the dispatcher so /// `distributed_outbox_messages_total` and the backlog gauges are populated. async fn drive_outbox(service_name: &str) { @@ -89,6 +118,30 @@ async fn drive_outbox(service_name: &str) { assert_eq!(outcome.released, 1); } +/// Drain an outbox row through the `Bus` -> `MessagePublisher` bridge. This must +/// contribute to outbox metrics without increasing direct producer metrics. +async fn drive_outbox_via_bus_publisher(service_name: &str) { + let repo = InMemoryRepository::new(); + let mut batch = CommitBatch::empty(); + batch.outbox_messages.push( + OutboxMessage::create("evt-bus-publisher", "orders.bus-published", b"{}".to_vec()).unwrap(), + ); + repo.commit_batch(batch).await.unwrap(); + + let publisher = BusPublisher::new(Arc::new(InMemoryBus::new())); + let dispatcher = OutboxDispatcher::new( + repo.outbox_store(), + publisher, + "metrics-exposition:bus-publisher", + Duration::from_secs(60), + 3, + ) + .with_service(service_name); + + let outcome = dispatcher.dispatch_batch(10).await.unwrap(); + assert_eq!(outcome.published, 1); +} + #[tokio::test] async fn exposition_covers_framework_families_and_passes_promtool() { let base = spawn_http_service("orders-exposition").await; @@ -113,7 +166,9 @@ async fn exposition_covers_framework_families_and_passes_promtool() { .unwrap(); assert_ne!(resp.status(), 200); + drive_direct_bus().await; drive_outbox("orders-exposition").await; + drive_outbox_via_bus_publisher("orders-bus-publisher").await; let scrape = client.get(format!("{base}/metrics")).send().await.unwrap(); assert_eq!(scrape.status(), 200); @@ -135,8 +190,13 @@ async fn exposition_covers_framework_families_and_passes_promtool() { "distributed_microsvc_dispatch_duration_seconds_bucket{service=\"orders-exposition\"", "distributed_microsvc_dispatch_duration_seconds_sum{service=\"orders-exposition\"", "distributed_microsvc_dispatch_duration_seconds_count{service=\"orders-exposition\"", + "distributed_transport_publish_total{service=\"unnamed\",transport=\"in_memory\",message_kind=\"command\",outcome=\"published\"} 1", + "distributed_transport_publish_total{service=\"unnamed\",transport=\"in_memory\",message_kind=\"event\",outcome=\"published\"} 1", + "distributed_transport_publish_duration_seconds_count{service=\"unnamed\",transport=\"in_memory\",message_kind=\"command\",outcome=\"published\"} 1", + "distributed_transport_publish_duration_seconds_count{service=\"unnamed\",transport=\"in_memory\",message_kind=\"event\",outcome=\"published\"} 1", "distributed_outbox_messages_total{service=\"orders-exposition\",outcome=\"published\"} 2", "distributed_outbox_messages_total{service=\"orders-exposition\",outcome=\"released\"} 1", + "distributed_outbox_messages_total{service=\"orders-bus-publisher\",outcome=\"published\"} 1", "distributed_outbox_pending_messages{service=\"orders-exposition\"} 1", "distributed_outbox_oldest_pending_age_seconds{service=\"orders-exposition\"}", ] { @@ -145,6 +205,17 @@ async fn exposition_covers_framework_families_and_passes_promtool() { "scrape must contain `{family}`:\n{body}" ); } + for forbidden in [ + "metrics.direct.command", + "metrics.direct.event", + "direct-message-id", + "4bf92f3577b34da6a3ce929d0e0e4736", + ] { + assert!( + !body.contains(forbidden), + "direct publish metrics must not expose high-cardinality value `{forbidden}`:\n{body}" + ); + } promtool_check(&body); } diff --git a/tests/otel_export/main.rs b/tests/otel_export/main.rs index bafb642b..5ee83619 100644 --- a/tests/otel_export/main.rs +++ b/tests/otel_export/main.rs @@ -21,6 +21,7 @@ use std::sync::Arc; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; +use distributed::bus::{Bus, InMemoryBus, Message as BusMessage, MessageKind as BusMessageKind}; use distributed::microsvc::{Context, Message, MessageKind, Routes, Service}; use opentelemetry::propagation::{Extractor, TextMapPropagator as _}; use opentelemetry::trace::TracerProvider as _; @@ -112,6 +113,42 @@ async fn dispatch_span_reaches_collector_with_propagated_parent() { .await .expect("nested dispatch succeeds"); + let bus = InMemoryBus::new(); + let publish_trace_id = fresh_trace_id(3); + let publish_traceparent = format!("00-{publish_trace_id}-{REMOTE_PARENT_SPAN_ID}-01"); + let publish_message = BusMessage::new( + "orders.created", + BusMessageKind::Event, + br#"{"id":"o-1"}"#.to_vec(), + ) + .with_id("evt-direct-publish") + .with_metadata("traceparent", &publish_traceparent); + bus.publish_message(publish_message) + .await + .expect("direct publish succeeds"); + + let nested_publish_trace_id = fresh_trace_id(19); + let nested_publish_traceparent = + format!("00-{nested_publish_trace_id}-{REMOTE_PARENT_SPAN_ID}-01"); + let publish_outer_span = tracing::info_span!("test.publish.outer"); + let parent_context = opentelemetry_sdk::propagation::TraceContextPropagator::new().extract( + &TraceparentExtractor { + traceparent: &nested_publish_traceparent, + }, + ); + let _ = publish_outer_span.set_parent(parent_context); + let nested_publish_message = BusMessage::new( + "orders.create", + BusMessageKind::Command, + br#"{"id":"o-2"}"#.to_vec(), + ) + .with_id("evt-direct-send") + .with_metadata("traceparent", &nested_publish_traceparent); + bus.send_message(nested_publish_message) + .instrument(publish_outer_span) + .await + .expect("nested direct send succeeds"); + provider.force_flush().expect("flush spans to collector"); provider.shutdown().expect("shutdown provider"); @@ -149,6 +186,61 @@ async fn dispatch_span_reaches_collector_with_propagated_parent() { outer_span["spanId"].as_str(), "dispatch span must preserve the active local span hierarchy: {nested_dispatch_span}" ); + + let publish_span = poll_for_span( + &traces_file, + &publish_trace_id, + "distributed.transport.publish", + Duration::from_secs(20), + ); + assert_eq!( + publish_span["parentSpanId"].as_str(), + Some(REMOTE_PARENT_SPAN_ID), + "direct publish span must be parented to the incoming traceparent: {publish_span}" + ); + assert_eq!( + span_attribute_str(&publish_span, "distributed.transport.name"), + Some("in_memory"), + "direct publish span must include the transport name: {publish_span}" + ); + assert_eq!( + span_attribute_str(&publish_span, "distributed.bus.operation"), + Some("publish"), + "direct publish span must include the bus operation: {publish_span}" + ); + assert_eq!( + span_attribute_str(&publish_span, "distributed.producer.source"), + Some("direct"), + "direct publish span must identify the producer source: {publish_span}" + ); + + let publish_outer_span = poll_for_span( + &traces_file, + &nested_publish_trace_id, + "test.publish.outer", + Duration::from_secs(20), + ); + let nested_publish_span = poll_for_span( + &traces_file, + &nested_publish_trace_id, + "distributed.transport.publish", + Duration::from_secs(20), + ); + assert_eq!( + publish_outer_span["parentSpanId"].as_str(), + Some(REMOTE_PARENT_SPAN_ID), + "publish outer span must keep the incoming traceparent as its parent: {publish_outer_span}" + ); + assert_eq!( + nested_publish_span["parentSpanId"].as_str(), + publish_outer_span["spanId"].as_str(), + "direct publish span must preserve the active local span hierarchy: {nested_publish_span}" + ); + assert_eq!( + span_attribute_str(&nested_publish_span, "distributed.bus.operation"), + Some("send"), + "direct send span must include the bus operation: {nested_publish_span}" + ); } /// Poll the collector's file-exporter output until the framework dispatch span @@ -197,6 +289,14 @@ fn find_span(batch: &Value, trace_id: &str, span_name: &str) -> Option { None } +fn span_attribute_str<'a>(span: &'a Value, key: &str) -> Option<&'a str> { + span["attributes"] + .as_array()? + .iter() + .find(|attribute| attribute["key"].as_str() == Some(key))?["value"]["stringValue"] + .as_str() +} + fn fresh_trace_id(salt: u128) -> String { format!( "{:032x}",