From 861c71c5d4380759ef296db57661a3668a154e55 Mon Sep 17 00:00:00 2001 From: Patrick Lee Scott Date: Tue, 7 Jul 2026 02:21:28 -0500 Subject: [PATCH 1/2] feat: add distributed HTTP telemetry Implements [[tasks/distributed-http-telemetry-1]] --- README.md | 7 + docs/metrics.md | 30 ++++- docs/observability.md | 57 +++++--- docs/transports.md | 4 + src/http_telemetry.rs | 128 ++++++++++++++++++ src/lib.rs | 2 + src/metrics.rs | 211 ++++++++++++++++++++++++++++++ src/microsvc/http.rs | 6 + src/microsvc/knative_ingress.rs | 6 + src/telemetry.rs | 83 ++++++------ src/trace_context.rs | 43 ++++++ tests/knative_cloudevents/main.rs | 28 +++- tests/metrics_exposition/main.rs | 43 +++++- tests/microsvc/transport_http.rs | 131 ++++++++++++++++++- tests/otel_export/main.rs | 99 +++++++++++++- 15 files changed, 814 insertions(+), 64 deletions(-) create mode 100644 src/http_telemetry.rs diff --git a/README.md b/README.md index ce07c9aa..b7a5c911 100644 --- a/README.md +++ b/README.md @@ -1122,6 +1122,7 @@ Routes: |---|---|---| | `POST` | `/:command` | Dispatch a command. Body = JSON input, headers = session variables. | | `GET` | `/health` | Health check: `{ "ok": true, "commands": ["counter.initialize", ...] }` | +| `GET` | `/metrics` | Prometheus text metrics when the `metrics` feature is enabled. | ```bash curl -X POST http://localhost:3000/counter.initialize \ @@ -1486,6 +1487,12 @@ Bus-only and worker services can expose the same registry on a side port with `distributed::metrics::serve_http`. See [`docs/metrics.md`](docs/metrics.md) for metric names, label rules, and GitOps details. +HTTP services record request-count and duration metrics for Distributed-owned +route templates (`/health`, `/metrics`, `/:command` as `/{command}`, `/`, and +`/cloudevent/{type}`) separately from dispatch metrics. Request metrics capture +pre-dispatch failures such as malformed JSON or body-limit rejection without +recording raw paths, query strings, headers, session variables, or payload data. + `describe`/`schema` compile your crate and call its `distributed_manifest()` entrypoint (override with `--entrypoint`), which registers the [read models](#read-models) and tables that define the schema: diff --git a/docs/metrics.md b/docs/metrics.md index 0c08aac2..7831a199 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -24,7 +24,8 @@ distributed::microsvc::serve(service, "0.0.0.0:3000").await?; Prometheus can scrape `GET /metrics` on the service's HTTP port. Hops ObserveStack should use the same scrape target and labels; dashboards can join -these metrics with trace data by the stable `service` and message labels. +HTTP request metrics with dispatch metrics and trace data by the stable +`service` label. The scrape route is unauthenticated by design. Do not expose `/metrics` on a public listener; keep it behind a private network, ingress policy, security @@ -50,6 +51,12 @@ This listener exposes only `GET /metrics`; it does not expose command dispatch. Metric labels are intentionally bounded: - `service`: `Service::named(...)`, or `unnamed` when no service name was set. +- `method`: normalized HTTP method. Standard methods keep their uppercase token; + non-standard methods are bucketed as `_OTHER`. +- `route`: Distributed-owned HTTP route template only: `/health`, `/metrics`, + `/{command}`, `/`, `/cloudevent/{type}`, or `unmatched`. Raw paths are never + recorded. +- `status_code`: numeric HTTP response status as a string. - `message_kind`: `command` or `event`. - `message`: registered command/event name; unknown-command failures are bucketed as `unknown` rather than recording the unrecognized input. @@ -65,8 +72,9 @@ Metric labels are intentionally bounded: `log_and_ack`, `stop`, `recv_error`, or a settle failure label such as `settle_ack`. -Do not add IDs, trace IDs, user IDs, aggregate IDs, or other high-cardinality -values as framework labels. +Do not add raw URI paths, query strings, IDs, trace IDs, request IDs, user IDs, +aggregate IDs, headers, cookies, session variables, payload data, arbitrary +metadata values, or other high-cardinality values as framework labels. The label names, framework status values, transport outcomes, failure actions, and outbox outcomes live in one internal telemetry vocabulary. New framework @@ -75,6 +83,10 @@ metrics should use that vocabulary rather than introducing ad hoc labels. ## Metric Families - `distributed_service_info{service,version}` gauge. +- `distributed_http_server_requests_total{service,method,route,status_code}` + counter. +- `distributed_http_server_request_duration_seconds{service,method,route,status_code}` + histogram. - `distributed_microsvc_dispatch_total{service,message_kind,message,status}` counter. - `distributed_microsvc_dispatch_duration_seconds{service,message_kind,message,status}` @@ -96,8 +108,16 @@ shape. ## Boundaries The `metrics` feature owns Prometheus text exposition for framework metrics. It -does not install an OpenTelemetry metrics SDK, emit request-level HTTP metrics, -or record user payload data. +does not install an OpenTelemetry metrics SDK or record user payload data. + +HTTP request metrics measure the Distributed-owned HTTP boundary: health checks, +metrics scrapes, command ingress, CloudEvents ingress, and framework rejections +that happen before dispatch such as malformed JSON or body-limit failures. +Dispatch metrics measure the command/event routing and handler boundary after a +request has been accepted. A single successful `POST /{command}` normally +records both HTTP request metrics and `distributed_microsvc_dispatch_*`; a bad +JSON body records only HTTP request metrics because no command dispatch occurs. +Unknown command dispatches still use `message="unknown"` in dispatch metrics. Direct transport receive paths emit receive/settle counters and failure counters. Outbox dispatch emits publish outcomes and backlog gauges. Direct diff --git a/docs/observability.md b/docs/observability.md index 8fc2db1a..990a04a9 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -59,10 +59,11 @@ 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: +The `otel` feature adds framework-owned `tracing` spans around HTTP request, +dispatch, handler execution, transport receive, and outbox publish boundaries. +When the incoming HTTP request or message carries W3C `traceparent` / +`tracestate`, Distributed extracts that context and sets it as the OpenTelemetry +parent for the framework span: ```toml [dependencies] @@ -77,20 +78,42 @@ setup in the generated `main.rs` for the common case. Framework span names are intentionally bounded: +- `GET /health` +- `GET /metrics` +- `POST /{command}` +- `POST /` +- `POST /cloudevent/{type}` - `distributed.microsvc.dispatch` - `distributed.handler` - `distributed.transport.receive` - `distributed.outbox.publish` -Each span uses the same framework-owned message attributes: +HTTP server spans are created only for Distributed-owned routers and are span +kind `SERVER`. They use bounded HTTP attributes: + +- `http.request.method` +- `http.route` +- `http.response.status_code` +- `network.protocol.name = "http"` + +Distributed intentionally emits `http.route` and omits `url.path` and +`url.query` in this first pass so route telemetry cannot leak raw paths or query +strings. A command HTTP request carrying `traceparent` creates a request span +parented to that incoming context; `distributed.microsvc.dispatch` is then a +child of the HTTP span. If a caller has already opened a local span before +entering Distributed, the existing ambient-span rule still wins and framework +spans nest under that local span. + +Message-oriented spans use these framework-owned message attributes: - `distributed.message.name` - `distributed.message.kind` - `messaging.message.id` -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. +Future spans should use the same helper paths so new attributes are reviewed in +one place. Do not add payload fields, user ids, aggregate ids, raw URI paths, +query strings, headers, cookies, session variables, or raw metadata values as +framework span attributes. Recommended environment variables for OTLP exporters: @@ -151,10 +174,12 @@ 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 a message carrying a W3C `traceparent`, sends an HTTP + command request carrying `traceparent`, and asserts a real OpenTelemetry + Collector received the request span and `distributed.microsvc.dispatch` with + the expected parent/child relationships (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`. @@ -167,11 +192,11 @@ platform concern, verified on a live cluster rather than per PR. The default feature set propagates W3C trace context metadata only. The `metrics` feature records framework metrics and renders Prometheus text. It -does not expose OpenTelemetry metrics or request-level HTTP telemetry. +does not expose OpenTelemetry metrics. -The `otel` feature creates framework spans and parents them from W3C metadata -when available. It does not install subscribers, exporters, sampling rules, or -resource attributes. +The `otel` feature creates framework spans and parents them from W3C HTTP +headers or message metadata when available. It does not install subscribers, +exporters, sampling rules, or resource attributes. Future `logs` or private `diagnostics` features should build on the same bounded telemetry vocabulary. Logs may carry structured failure context, but diff --git a/docs/transports.md b/docs/transports.md index 33c6b0e9..400dd240 100644 --- a/docs/transports.md +++ b/docs/transports.md @@ -180,6 +180,10 @@ in-process consume loop: `KnativeBus::manifests(&plan, &subscriptions)` renders the role-based `Broker` + per-name `Trigger` YAML (subscriber URIs `/cloudevent/`, with a `.local(addr)` kubefwd variant), and the service mounts `cloud_events_router` so those Triggers reach `dispatch_message`. +When `metrics` is enabled, the router also exposes `GET /metrics` and records +bounded HTTP request metrics for `/`, `/cloudevent/{type}`, and `/metrics`; +dispatch metrics remain separate and describe the command/event handler +boundary after a CloudEvent is parsed. `PostgresBus` uses the claim-lease work queue (not `sqlxmq`) for the same reason the low-level adapter does — sqlxmq's always-on push runner doesn't compose with diff --git a/src/http_telemetry.rs b/src/http_telemetry.rs new file mode 100644 index 00000000..d6baec51 --- /dev/null +++ b/src/http_telemetry.rs @@ -0,0 +1,128 @@ +//! Shared telemetry middleware for Distributed-owned HTTP routers. + +#[cfg(feature = "metrics")] +use std::time::Instant; + +#[cfg(any(feature = "metrics", feature = "otel"))] +use axum::extract::MatchedPath; +use axum::extract::{Request, State}; +#[cfg(any(feature = "metrics", feature = "otel"))] +use axum::http::Method; +use axum::middleware::Next; +use axum::response::Response; + +#[derive(Clone, Debug, Default)] +pub(crate) struct HttpTelemetryState { + #[cfg(feature = "metrics")] + service: Option, +} + +impl HttpTelemetryState { + pub(crate) fn new(service: Option) -> Self { + #[cfg(not(feature = "metrics"))] + let _ = service; + #[cfg(not(feature = "metrics"))] + return Self {}; + + #[cfg(feature = "metrics")] + Self { service } + } + + #[cfg(feature = "metrics")] + fn service(&self) -> Option<&str> { + self.service.as_deref() + } +} + +pub(crate) async fn middleware( + State(state): State, + req: Request, + next: Next, +) -> Response { + #[cfg(not(any(feature = "metrics", feature = "otel")))] + { + let _ = state; + return next.run(req).await; + } + + #[cfg(any(feature = "metrics", feature = "otel"))] + { + #[cfg(not(feature = "metrics"))] + let _ = state; + + let method = normalize_method(req.method()); + let route = route_label(req.extensions().get::()); + #[cfg(feature = "metrics")] + let started = Instant::now(); + + #[cfg(feature = "otel")] + let span = http_server_span(method, route); + #[cfg(feature = "otel")] + crate::trace_context::set_span_parent_from_headers_if_no_current_span(&span, req.headers()); + + #[cfg(feature = "otel")] + let response = { + use tracing::Instrument as _; + + next.run(req).instrument(span.clone()).await + }; + #[cfg(not(feature = "otel"))] + let response = next.run(req).await; + + let status = response.status(); + #[cfg(feature = "otel")] + span.record("http.response.status_code", status.as_u16()); + #[cfg(feature = "metrics")] + crate::metrics::record_http_server_request( + state.service(), + method, + route, + status.as_u16(), + started.elapsed(), + ); + + response + } +} + +#[cfg(any(feature = "metrics", feature = "otel"))] +fn normalize_method(method: &Method) -> &'static str { + match method.as_str() { + "GET" => "GET", + "HEAD" => "HEAD", + "POST" => "POST", + "PUT" => "PUT", + "PATCH" => "PATCH", + "DELETE" => "DELETE", + "OPTIONS" => "OPTIONS", + "TRACE" => "TRACE", + "CONNECT" => "CONNECT", + _ => "_OTHER", + } +} + +#[cfg(any(feature = "metrics", feature = "otel"))] +fn route_label(matched_path: Option<&MatchedPath>) -> &'static str { + match matched_path.map(MatchedPath::as_str) { + Some("/health") => "/health", + Some("/metrics") => "/metrics", + Some("/{command}") => "/{command}", + Some("/") => "/", + Some("/cloudevent/{type}") => "/cloudevent/{type}", + _ => "unmatched", + } +} + +#[cfg(feature = "otel")] +fn http_server_span(method: &'static str, route: &'static str) -> tracing::Span { + let span_name = format!("{method} {route}"); + tracing::info_span!( + "distributed.http.server", + otel.name = span_name.as_str(), + otel.kind = "server", + http.request.method = method, + http.route = route, + http.response.status_code = tracing::field::Empty, + network.protocol.name = "http", + ) +} diff --git a/src/lib.rs b/src/lib.rs index d6a0112c..6178f0b9 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -12,6 +12,8 @@ pub mod repository; mod commit_builder; #[cfg(feature = "emitter")] pub mod emitter; +#[cfg(feature = "http")] +mod http_telemetry; mod in_memory_repo; pub mod lock; pub mod manifest; diff --git a/src/metrics.rs b/src/metrics.rs index 799b5401..3adc886f 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -28,6 +28,14 @@ const MICROSVC_DISPATCH_DURATION_FAMILY: MetricFamily = MetricFamily::histogram( metric_names::MICROSVC_DISPATCH_DURATION_SECONDS, "Microsvc dispatch duration in seconds.", ); +const HTTP_SERVER_REQUESTS_TOTAL_FAMILY: MetricFamily = MetricFamily::counter( + metric_names::HTTP_SERVER_REQUESTS_TOTAL, + "Total HTTP server requests by service, method, route, and status code.", +); +const HTTP_SERVER_REQUEST_DURATION_FAMILY: MetricFamily = MetricFamily::histogram( + metric_names::HTTP_SERVER_REQUEST_DURATION_SECONDS, + "HTTP server request duration in seconds.", +); const TRANSPORT_MESSAGES_TOTAL_FAMILY: MetricFamily = MetricFamily::counter( metric_names::TRANSPORT_MESSAGES_TOTAL, "Total transport receive/settle outcomes.", @@ -77,6 +85,23 @@ pub fn record_microsvc_dispatch( }); } +/// Record one Distributed-owned HTTP server request. +pub fn record_http_server_request( + service: Option<&str>, + method: &str, + route: &str, + status_code: u16, + duration: Duration, +) { + registry().record_http_server_request(HttpServerRequestKey { + service: service_label(service), + method: normalize_http_method(method).to_string(), + route: normalize_http_route(route).to_string(), + status_code: status_code.to_string(), + duration_seconds: duration.as_secs_f64(), + }); +} + /// Record a transport receive/settle outcome. pub fn record_transport_message( service: Option<&str>, @@ -237,8 +262,13 @@ struct MetricsHttpState { #[cfg(feature = "http")] fn http_router_with_state(state: MetricsHttpState) -> axum::Router { + let telemetry = crate::http_telemetry::HttpTelemetryState::new(state.service.clone()); axum::Router::new() .route("/metrics", axum::routing::get(metrics_http_handler)) + .layer(axum::middleware::from_fn_with_state( + telemetry, + crate::http_telemetry::middleware, + )) .with_state(state) } @@ -390,6 +420,8 @@ struct MetricsRegistry { service_info: Mutex>, dispatch_total: Mutex>, dispatch_duration: Mutex>, + http_server_requests_total: Mutex>, + http_server_request_duration: Mutex>, transport_messages_total: Mutex>, transport_failures_total: Mutex>, outbox_messages_total: Mutex>, @@ -415,6 +447,19 @@ impl MetricsRegistry { self.note_service(service); } + fn record_http_server_request(&self, key: HttpServerRequestKey) { + let service = key.service.clone(); + self.lock(&self.http_server_requests_total) + .entry(key.counter_key()) + .and_modify(|value| *value += 1) + .or_insert(1); + self.lock(&self.http_server_request_duration) + .entry(key.histogram_key()) + .or_insert_with(Histogram::new) + .observe(key.duration_seconds); + self.note_service(service); + } + fn record_transport_message(&self, key: TransportMessageKey) { let service = key.service.clone(); self.lock(&self.transport_messages_total) @@ -459,6 +504,8 @@ impl MetricsRegistry { let service_info = self.clone_locked(&self.service_info); let dispatch_total = self.clone_locked(&self.dispatch_total); let dispatch_duration = self.clone_locked(&self.dispatch_duration); + let http_server_requests_total = self.clone_locked(&self.http_server_requests_total); + let http_server_request_duration = self.clone_locked(&self.http_server_request_duration); let transport_messages_total = self.clone_locked(&self.transport_messages_total); let transport_failures_total = self.clone_locked(&self.transport_failures_total); let outbox_messages_total = self.clone_locked(&self.outbox_messages_total); @@ -497,6 +544,18 @@ impl MetricsRegistry { .map(|(key, histogram)| MetricSample::histogram(key.labels(), histogram)) .collect(), ), + HTTP_SERVER_REQUESTS_TOTAL_FAMILY.snapshot( + http_server_requests_total + .iter() + .map(|(key, value)| MetricSample::counter(key.labels(), *value)) + .collect(), + ), + HTTP_SERVER_REQUEST_DURATION_FAMILY.snapshot( + http_server_request_duration + .iter() + .map(|(key, histogram)| MetricSample::histogram(key.labels(), histogram)) + .collect(), + ), TRANSPORT_MESSAGES_TOTAL_FAMILY.snapshot( transport_messages_total .iter() @@ -540,6 +599,8 @@ impl MetricsRegistry { self.lock(&self.service_info).clear(); self.lock(&self.dispatch_total).clear(); self.lock(&self.dispatch_duration).clear(); + self.lock(&self.http_server_requests_total).clear(); + self.lock(&self.http_server_request_duration).clear(); self.lock(&self.transport_messages_total).clear(); self.lock(&self.transport_failures_total).clear(); self.lock(&self.outbox_messages_total).clear(); @@ -635,6 +696,63 @@ impl DispatchHistogramKey { } } +#[derive(Clone)] +struct HttpServerRequestKey { + service: String, + method: String, + route: String, + status_code: String, + duration_seconds: f64, +} + +impl HttpServerRequestKey { + fn counter_key(&self) -> HttpServerRequestCounterKey { + HttpServerRequestCounterKey { + service: self.service.clone(), + method: self.method.clone(), + route: self.route.clone(), + status_code: self.status_code.clone(), + } + } + + fn histogram_key(&self) -> HttpServerRequestHistogramKey { + HttpServerRequestHistogramKey { + service: self.service.clone(), + method: self.method.clone(), + route: self.route.clone(), + status_code: self.status_code.clone(), + } + } +} + +#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)] +struct HttpServerRequestCounterKey { + service: String, + method: String, + route: String, + status_code: String, +} + +impl HttpServerRequestCounterKey { + fn labels(&self) -> Vec<(String, String)> { + http_server_request_labels(&self.service, &self.method, &self.route, &self.status_code) + } +} + +#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)] +struct HttpServerRequestHistogramKey { + service: String, + method: String, + route: String, + status_code: String, +} + +impl HttpServerRequestHistogramKey { + fn labels(&self) -> Vec<(String, String)> { + http_server_request_labels(&self.service, &self.method, &self.route, &self.status_code) + } +} + #[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)] struct TransportMessageKey { service: String, @@ -725,6 +843,49 @@ fn service_labels(service: &str) -> Vec<(String, String)> { vec![(metric_labels::SERVICE.to_string(), service.to_string())] } +fn http_server_request_labels( + service: &str, + method: &str, + route: &str, + status_code: &str, +) -> Vec<(String, String)> { + vec![ + (metric_labels::SERVICE.to_string(), service.to_string()), + (metric_labels::METHOD.to_string(), method.to_string()), + (metric_labels::ROUTE.to_string(), route.to_string()), + ( + metric_labels::STATUS_CODE.to_string(), + status_code.to_string(), + ), + ] +} + +fn normalize_http_method(method: &str) -> &'static str { + match method { + "GET" => "GET", + "HEAD" => "HEAD", + "POST" => "POST", + "PUT" => "PUT", + "PATCH" => "PATCH", + "DELETE" => "DELETE", + "OPTIONS" => "OPTIONS", + "TRACE" => "TRACE", + "CONNECT" => "CONNECT", + _ => "_OTHER", + } +} + +fn normalize_http_route(route: &str) -> &'static str { + match route { + "/health" => "/health", + "/metrics" => "/metrics", + "/{command}" => "/{command}", + "/" => "/", + "/cloudevent/{type}" => "/cloudevent/{type}", + _ => "unmatched", + } +} + fn render_prometheus(snapshot: &MetricsSnapshot) -> String { let mut output = String::new(); for family_snapshot in snapshot.families() { @@ -909,6 +1070,13 @@ mod tests { dispatch_status::SUCCESS, Duration::from_millis(7), ); + record_http_server_request( + Some("orders"), + "POST", + "/{command}", + 200, + Duration::from_millis(11), + ); record_transport_message( Some("orders"), "nats", @@ -937,6 +1105,8 @@ mod tests { metric_names::SERVICE_INFO, metric_names::MICROSVC_DISPATCH_TOTAL, metric_names::MICROSVC_DISPATCH_DURATION_SECONDS, + metric_names::HTTP_SERVER_REQUESTS_TOTAL, + metric_names::HTTP_SERVER_REQUEST_DURATION_SECONDS, metric_names::TRANSPORT_MESSAGES_TOTAL, metric_names::TRANSPORT_FAILURES_TOTAL, metric_names::OUTBOX_MESSAGES_TOTAL, @@ -971,6 +1141,40 @@ mod tests { &duration_family.samples[0].value, MetricSampleValue::Histogram(_) )); + + let http_duration_family = snapshot + .families() + .iter() + .find(|family| family.family.name == metric_names::HTTP_SERVER_REQUEST_DURATION_SECONDS) + .expect("http duration family is present"); + assert!(matches!( + &http_duration_family.samples[0].value, + MetricSampleValue::Histogram(_) + )); + } + + #[test] + fn http_request_metrics_clamp_method_and_route_labels() { + let _guard = lock_for_tests(); + reset_for_tests(); + + record_http_server_request( + Some("orders"), + "BREW", + "/orders/o-1?user_id=alice", + 418, + Duration::from_millis(3), + ); + + let text = prometheus_text(); + + assert!(text.contains( + "distributed_http_server_requests_total{service=\"orders\",method=\"_OTHER\",route=\"unmatched\",status_code=\"418\"} 1" + )); + assert!( + !text.contains("orders/o-1") && !text.contains("user_id"), + "raw path and query values must not appear in metrics:\n{text}" + ); } #[test] @@ -986,6 +1190,13 @@ mod tests { dispatch_status::SUCCESS, Duration::from_millis(2), ); + record_http_server_request( + Some("orders"), + "POST", + "/{command}", + 200, + Duration::from_millis(4), + ); record_transport_message( Some("orders"), "rabbitmq", diff --git a/src/microsvc/http.rs b/src/microsvc/http.rs index 68bf25bb..92360b3f 100644 --- a/src/microsvc/http.rs +++ b/src/microsvc/http.rs @@ -47,6 +47,8 @@ use super::MAX_HTTP_BODY_BYTES; /// Build an axum `Router` that dispatches commands via the given service. pub fn router(service: Arc) -> Router { + let telemetry = + crate::http_telemetry::HttpTelemetryState::new(service.name().map(str::to_string)); let router = Router::new() .route("/health", get(health_handler)) .route("/{command}", axum::routing::post(command_handler)); @@ -57,6 +59,10 @@ pub fn router(service: Arc) -> Router { // Pin the body limit explicitly rather than relying on axum's default; // the command handler buffers the JSON body into memory. .layer(DefaultBodyLimit::max(MAX_HTTP_BODY_BYTES)) + .layer(axum::middleware::from_fn_with_state( + telemetry, + crate::http_telemetry::middleware, + )) .with_state(service) } diff --git a/src/microsvc/knative_ingress.rs b/src/microsvc/knative_ingress.rs index 292ced55..eb8b2892 100644 --- a/src/microsvc/knative_ingress.rs +++ b/src/microsvc/knative_ingress.rs @@ -44,6 +44,8 @@ const STRUCTURED_CONTENT_TYPE: &str = "application/cloudevents+json"; /// alignment only), so a Knative Trigger can target either a single shared `ref` /// (`/`) or the per-type subscriber URI `KnativeBus` emits (`/cloudevent/`). pub fn cloud_events_router(service: Arc) -> Router { + let telemetry = + crate::http_telemetry::HttpTelemetryState::new(service.name().map(str::to_string)); let router = Router::new() .route("/", axum::routing::post(ingress_handler)) .route("/cloudevent/{type}", axum::routing::post(ingress_handler)); @@ -54,6 +56,10 @@ pub fn cloud_events_router(service: Arc) -> Router { // Pin the body limit explicitly rather than relying on axum's default, // since the handler buffers the whole body into memory. .layer(DefaultBodyLimit::max(MAX_HTTP_BODY_BYTES)) + .layer(axum::middleware::from_fn_with_state( + telemetry, + crate::http_telemetry::middleware, + )) .with_state(service) } diff --git a/src/telemetry.rs b/src/telemetry.rs index bff32be1..b516f80e 100644 --- a/src/telemetry.rs +++ b/src/telemetry.rs @@ -19,6 +19,9 @@ pub(crate) mod metric_names { pub(crate) const MICROSVC_DISPATCH_TOTAL: &str = "distributed_microsvc_dispatch_total"; pub(crate) const MICROSVC_DISPATCH_DURATION_SECONDS: &str = "distributed_microsvc_dispatch_duration_seconds"; + pub(crate) const HTTP_SERVER_REQUESTS_TOTAL: &str = "distributed_http_server_requests_total"; + pub(crate) const HTTP_SERVER_REQUEST_DURATION_SECONDS: &str = + "distributed_http_server_request_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 OUTBOX_MESSAGES_TOTAL: &str = "distributed_outbox_messages_total"; @@ -34,6 +37,9 @@ pub(crate) mod metric_labels { pub(crate) const MESSAGE_KIND: &str = "message_kind"; pub(crate) const MESSAGE: &str = "message"; pub(crate) const STATUS: &str = "status"; + pub(crate) const METHOD: &str = "method"; + pub(crate) const ROUTE: &str = "route"; + pub(crate) const STATUS_CODE: &str = "status_code"; pub(crate) const TRANSPORT: &str = "transport"; pub(crate) const OUTCOME: &str = "outcome"; pub(crate) const FAILURE_CLASS: &str = "failure_class"; @@ -96,43 +102,6 @@ pub(crate) mod outbox_outcome { pub(crate) const FAILED: &str = "failed"; } -#[cfg(test)] -#[cfg(feature = "metrics")] -pub(crate) mod privacy_policy { - pub(crate) const ALLOWED_METRIC_LABELS: &[&str] = &[ - super::metric_labels::SERVICE, - super::metric_labels::VERSION, - super::metric_labels::MESSAGE_KIND, - super::metric_labels::MESSAGE, - super::metric_labels::STATUS, - super::metric_labels::TRANSPORT, - super::metric_labels::OUTCOME, - super::metric_labels::FAILURE_CLASS, - super::metric_labels::ACTION, - super::metric_labels::LE, - ]; - - pub(crate) const FORBIDDEN_METRIC_LABELS: &[&str] = &[ - "request_id", - "trace_id", - "span_id", - "correlation_id", - "causation_id", - "message_id", - "aggregate_id", - "aggregate_type", - "stream_id", - "user_id", - "tenant_id", - "payload", - "metadata", - "http_path", - "http_target", - "http_route", - "http_user_agent", - ]; -} - #[cfg(feature = "metrics")] pub(crate) fn service_label(service: Option<&str>) -> String { service @@ -229,3 +198,43 @@ 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(test)] +#[cfg(feature = "metrics")] +pub(crate) mod privacy_policy { + pub(crate) const ALLOWED_METRIC_LABELS: &[&str] = &[ + super::metric_labels::SERVICE, + super::metric_labels::VERSION, + super::metric_labels::MESSAGE_KIND, + super::metric_labels::MESSAGE, + super::metric_labels::STATUS, + super::metric_labels::METHOD, + super::metric_labels::ROUTE, + super::metric_labels::STATUS_CODE, + super::metric_labels::TRANSPORT, + super::metric_labels::OUTCOME, + super::metric_labels::FAILURE_CLASS, + super::metric_labels::ACTION, + super::metric_labels::LE, + ]; + + pub(crate) const FORBIDDEN_METRIC_LABELS: &[&str] = &[ + "request_id", + "trace_id", + "span_id", + "correlation_id", + "causation_id", + "message_id", + "aggregate_id", + "aggregate_type", + "stream_id", + "user_id", + "tenant_id", + "payload", + "metadata", + "http_path", + "http_target", + "http_route", + "http_user_agent", + ]; +} diff --git a/src/trace_context.rs b/src/trace_context.rs index 2792fe57..b0b8acac 100644 --- a/src/trace_context.rs +++ b/src/trace_context.rs @@ -103,17 +103,49 @@ fn set_span_parent_from_metadata(span: &tracing::Span, metadata: &[(String, Stri } } +#[cfg(all(feature = "otel", feature = "http"))] +pub(crate) fn set_span_parent_from_headers_if_no_current_span( + span: &tracing::Span, + headers: &axum::http::HeaderMap, +) { + if tracing::Span::current().id().is_none() { + set_span_parent_from_headers(span, headers); + } +} + +#[cfg(all(feature = "otel", feature = "http"))] +fn set_span_parent_from_headers(span: &tracing::Span, headers: &axum::http::HeaderMap) { + use opentelemetry::trace::TraceContextExt as _; + use tracing_opentelemetry::OpenTelemetrySpanExt as _; + + let parent_context = extract_otel_context_from_headers(headers); + if parent_context.span().span_context().is_valid() { + let _ = span.set_parent(parent_context); + } +} + #[cfg(feature = "otel")] fn extract_otel_context_from_metadata(metadata: &[(String, String)]) -> opentelemetry::Context { opentelemetry_sdk::propagation::TraceContextPropagator::new() .extract(&MetadataExtractor { metadata }) } +#[cfg(all(feature = "otel", feature = "http"))] +fn extract_otel_context_from_headers(headers: &axum::http::HeaderMap) -> opentelemetry::Context { + opentelemetry_sdk::propagation::TraceContextPropagator::new() + .extract(&HeaderExtractor { headers }) +} + #[cfg(feature = "otel")] struct MetadataExtractor<'a> { metadata: &'a [(String, String)], } +#[cfg(all(feature = "otel", feature = "http"))] +struct HeaderExtractor<'a> { + headers: &'a axum::http::HeaderMap, +} + #[cfg(feature = "otel")] impl Extractor for MetadataExtractor<'_> { fn get(&self, key: &str) -> Option<&str> { @@ -128,6 +160,17 @@ impl Extractor for MetadataExtractor<'_> { } } +#[cfg(all(feature = "otel", feature = "http"))] +impl Extractor for HeaderExtractor<'_> { + fn get(&self, key: &str) -> Option<&str> { + self.headers.get(key).and_then(|value| value.to_str().ok()) + } + + fn keys(&self) -> Vec<&str> { + self.headers.keys().map(|key| key.as_str()).collect() + } +} + fn replace_vec_key(metadata: &mut Vec<(String, String)>, key: &'static str, value: Option<&str>) { metadata.retain(|(existing, _)| !existing.eq_ignore_ascii_case(key)); if let Some(value) = value { diff --git a/tests/knative_cloudevents/main.rs b/tests/knative_cloudevents/main.rs index 4742a178..f9d65492 100644 --- a/tests/knative_cloudevents/main.rs +++ b/tests/knative_cloudevents/main.rs @@ -160,14 +160,36 @@ async fn metrics_endpoint_exposes_cloud_event_dispatch_counters() { .unwrap(); assert_eq!(resp.status(), 200); - let resp = client.get(format!("{url}metrics")).send().await.unwrap(); - assert_eq!(resp.status(), 200); + let per_type = client + .post(format!("{url}cloudevent/order.initialized")) + .header("ce-id", "evt-metrics-typed") + .header("ce-type", "order.initialized") + .header("ce-source", "/orders") + .header("content-type", "application/json") + .body(r#"{"order":"o-metrics-typed"}"#) + .send() + .await + .unwrap(); + assert_eq!(per_type.status(), 200); + + let first_scrape = client.get(format!("{url}metrics")).send().await.unwrap(); + assert_eq!(first_scrape.status(), 200); + let final_scrape = client.get(format!("{url}metrics")).send().await.unwrap(); + assert_eq!(final_scrape.status(), 200); - let body = resp.text().await.unwrap(); + let body = final_scrape.text().await.unwrap(); assert!( body.contains("distributed_microsvc_dispatch_total{service=\"unnamed\",message_kind=\"event\",message=\"order.initialized\",status=\"success\"}"), "metrics body should include the CloudEvents dispatch counter:\n{body}" ); + assert!( + body.contains("distributed_http_server_requests_total{service=\"unnamed\",method=\"POST\",route=\"/\",status_code=\"200\"}"), + "metrics body should include the shared CloudEvents ingress route:\n{body}" + ); + assert!( + body.contains("distributed_http_server_requests_total{service=\"unnamed\",method=\"POST\",route=\"/cloudevent/{type}\",status_code=\"200\"}"), + "metrics body should include the per-type CloudEvents ingress route:\n{body}" + ); } #[tokio::test] diff --git a/tests/metrics_exposition/main.rs b/tests/metrics_exposition/main.rs index 1ca102cf..904aa6be 100644 --- a/tests/metrics_exposition/main.rs +++ b/tests/metrics_exposition/main.rs @@ -19,7 +19,7 @@ use std::sync::{Arc, Mutex}; use std::time::Duration; use distributed::bus::{MessagePublisher, TransportError}; -use distributed::microsvc::{self, Context, Message, Routes, Service}; +use distributed::microsvc::{self, Context, Message, Routes, Service, MAX_HTTP_BODY_BYTES}; use distributed::outbox_worker::OutboxDispatcher; use distributed::{CommitBatch, InMemoryRepository, OutboxMessage, TransactionalCommit}; use serde_json::json; @@ -112,9 +112,33 @@ async fn exposition_covers_framework_families_and_passes_promtool() { .await .unwrap(); assert_ne!(resp.status(), 200); + let unknown_status = resp.status().as_u16(); + + let health = client.get(format!("{base}/health")).send().await.unwrap(); + assert_eq!(health.status(), 200); + + let bad_json = client + .post(format!("{base}/orders.create")) + .header("content-type", "application/json") + .body("{not valid json") + .send() + .await + .unwrap(); + assert!(bad_json.status().is_client_error()); + + let oversized = client + .post(format!("{base}/orders.create")) + .header("content-type", "application/json") + .body("x".repeat(MAX_HTTP_BODY_BYTES + 1)) + .send() + .await + .unwrap(); + assert!(oversized.status().is_client_error()); drive_outbox("orders-exposition").await; + let first_scrape = client.get(format!("{base}/metrics")).send().await.unwrap(); + assert_eq!(first_scrape.status(), 200); let scrape = client.get(format!("{base}/metrics")).send().await.unwrap(); assert_eq!(scrape.status(), 200); let content_type = scrape @@ -131,6 +155,23 @@ async fn exposition_covers_framework_families_and_passes_promtool() { let body = scrape.text().await.unwrap(); for family in [ "distributed_service_info{service=\"orders-exposition\"", + "distributed_http_server_requests_total{service=\"orders-exposition\",method=\"GET\",route=\"/health\",status_code=\"200\"}", + "distributed_http_server_requests_total{service=\"orders-exposition\",method=\"GET\",route=\"/metrics\",status_code=\"200\"}", + "distributed_http_server_requests_total{service=\"orders-exposition\",method=\"POST\",route=\"/{command}\",status_code=\"200\"}", + &format!( + "distributed_http_server_requests_total{{service=\"orders-exposition\",method=\"POST\",route=\"/{{command}}\",status_code=\"{unknown_status}\"}}" + ), + &format!( + "distributed_http_server_requests_total{{service=\"orders-exposition\",method=\"POST\",route=\"/{{command}}\",status_code=\"{}\"}}", + bad_json.status().as_u16() + ), + &format!( + "distributed_http_server_requests_total{{service=\"orders-exposition\",method=\"POST\",route=\"/{{command}}\",status_code=\"{}\"}}", + oversized.status().as_u16() + ), + "distributed_http_server_request_duration_seconds_bucket{service=\"orders-exposition\"", + "distributed_http_server_request_duration_seconds_sum{service=\"orders-exposition\"", + "distributed_http_server_request_duration_seconds_count{service=\"orders-exposition\"", "distributed_microsvc_dispatch_total{service=\"orders-exposition\"", "distributed_microsvc_dispatch_duration_seconds_bucket{service=\"orders-exposition\"", "distributed_microsvc_dispatch_duration_seconds_sum{service=\"orders-exposition\"", diff --git a/tests/microsvc/transport_http.rs b/tests/microsvc/transport_http.rs index a4194269..91c61674 100644 --- a/tests/microsvc/transport_http.rs +++ b/tests/microsvc/transport_http.rs @@ -4,7 +4,7 @@ use std::sync::Arc; -use distributed::microsvc::{self, Routes, Service}; +use distributed::microsvc::{self, Routes, Service, MAX_HTTP_BODY_BYTES}; use distributed::{AggregateBuilder, InMemoryRepository, Queueable}; use serde_json::json; @@ -20,6 +20,15 @@ fn counter_service() -> Arc { ))) } +fn named_counter_service(name: &str) -> Arc { + Arc::new(Service::new().named(name).routes(distributed::routes!( + Routes::new().with_repo(InMemoryRepository::new().queued().aggregate::()), + command handlers::counter_create, + command handlers::counter_increment, + command handlers::whoami, + ))) +} + /// Bind to port 0 and return the actual address. async fn start_server(service: Arc) -> String { start_app(microsvc::router(service)).await @@ -125,6 +134,126 @@ async fn standalone_metrics_router_exposes_worker_metrics() { ); } +#[cfg(feature = "metrics")] +#[tokio::test] +async fn http_request_metrics_cover_framework_routes_and_rejections() { + let service_name = "http-telemetry-routes"; + let service = named_counter_service(service_name); + let base = start_server(service).await; + let client = reqwest::Client::new(); + let traceparent = "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"; + + let health = client + .get(format!("{base}/health?user_id=user-42&aggregate_id=agg-1")) + .header("traceparent", traceparent) + .send() + .await + .unwrap(); + assert_eq!(health.status(), 200); + + let success = client + .post(format!("{base}/counter.initialize?request_id=req-1")) + .header("traceparent", traceparent) + .header("x-hasura-user-id", "user-42") + .json(&json!({ "id": "http-telemetry-counter" })) + .send() + .await + .unwrap(); + assert_eq!(success.status(), 200); + + let unknown = client + .post(format!("{base}/definitely.missing?aggregate_id=agg-1")) + .json(&json!({})) + .send() + .await + .unwrap(); + assert_eq!(unknown.status(), 404); + + let bad_json = client + .post(format!("{base}/counter.initialize")) + .header("content-type", "application/json") + .body("{not valid json") + .send() + .await + .unwrap(); + assert!(bad_json.status().is_client_error()); + + let oversized = client + .post(format!("{base}/counter.initialize")) + .header("content-type", "application/json") + .body("x".repeat(MAX_HTTP_BODY_BYTES + 1)) + .send() + .await + .unwrap(); + assert!(oversized.status().is_client_error()); + + let first_scrape = client.get(format!("{base}/metrics")).send().await.unwrap(); + assert_eq!(first_scrape.status(), 200); + let final_scrape = client.get(format!("{base}/metrics")).send().await.unwrap(); + assert_eq!(final_scrape.status(), 200); + let body = final_scrape.text().await.unwrap(); + + assert_http_request_metric(&body, service_name, "GET", "/health", 200); + assert_http_request_metric(&body, service_name, "POST", "/{command}", 200); + assert_http_request_metric(&body, service_name, "POST", "/{command}", 404); + assert_http_request_metric( + &body, + service_name, + "POST", + "/{command}", + bad_json.status().as_u16(), + ); + assert_http_request_metric( + &body, + service_name, + "POST", + "/{command}", + oversized.status().as_u16(), + ); + assert_http_request_metric(&body, service_name, "GET", "/metrics", 200); + assert!( + body.contains(&format!( + "distributed_microsvc_dispatch_total{{service=\"{service_name}\",message_kind=\"command\",message=\"unknown\",status=\"unknown_command\"}}" + )), + "dispatch metrics should still bucket unknown commands under a fixed message label:\n{body}" + ); + + for forbidden in [ + "/counter.initialize", + "definitely.missing", + "request_id", + "req-1", + "user-42", + "agg-1", + "http-telemetry-counter", + "x-hasura-user-id", + "4bf92f3577b34da6a3ce929d0e0e4736", + "00f067aa0ba902b7", + ] { + assert!( + !body.contains(forbidden), + "metrics body must not contain `{forbidden}`:\n{body}" + ); + } +} + +#[cfg(feature = "metrics")] +fn assert_http_request_metric( + body: &str, + service: &str, + method: &str, + route: &str, + status_code: u16, +) { + let expected = format!( + "distributed_http_server_requests_total{{service=\"{service}\",method=\"{method}\",route=\"{route}\",status_code=\"{status_code}\"}}" + ); + assert!( + body.contains(&expected), + "metrics body should contain `{expected}`:\n{body}" + ); +} + #[tokio::test] async fn create_and_increment_counter() { let service = counter_service(); diff --git a/tests/otel_export/main.rs b/tests/otel_export/main.rs index bafb642b..8439865c 100644 --- a/tests/otel_export/main.rs +++ b/tests/otel_export/main.rs @@ -21,7 +21,7 @@ use std::sync::Arc; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; -use distributed::microsvc::{Context, Message, MessageKind, Routes, Service}; +use distributed::microsvc::{self, Context, Message, MessageKind, Routes, Service}; use opentelemetry::propagation::{Extractor, TextMapPropagator as _}; use opentelemetry::trace::TracerProvider as _; use opentelemetry_otlp::WithExportConfig as _; @@ -112,6 +112,18 @@ async fn dispatch_span_reaches_collector_with_propagated_parent() { .await .expect("nested dispatch succeeds"); + let http_trace_id = fresh_trace_id(31); + let http_traceparent = format!("00-{http_trace_id}-{REMOTE_PARENT_SPAN_ID}-01"); + let base = spawn_http_service(service.clone()).await; + let response = reqwest::Client::new() + .post(format!("{base}/orders.create")) + .header("traceparent", &http_traceparent) + .json(&json!({"id": "o-http"})) + .send() + .await + .expect("http request succeeds"); + assert_eq!(response.status(), 200); + provider.force_flush().expect("flush spans to collector"); provider.shutdown().expect("shutdown provider"); @@ -149,6 +161,72 @@ 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 http_span = poll_for_span( + &traces_file, + &http_trace_id, + "POST /{command}", + Duration::from_secs(20), + ); + let http_dispatch_span = poll_for_span( + &traces_file, + &http_trace_id, + "distributed.microsvc.dispatch", + Duration::from_secs(20), + ); + assert_eq!( + http_span["parentSpanId"].as_str(), + Some(REMOTE_PARENT_SPAN_ID), + "HTTP request span must be parented to incoming traceparent: {http_span}" + ); + assert_eq!( + http_span["kind"].as_i64(), + Some(2), + "HTTP request span must be exported as SERVER kind: {http_span}" + ); + assert_eq!( + span_string_attribute(&http_span, "http.request.method"), + Some("POST"), + "HTTP request span should record bounded method attribute: {http_span}" + ); + assert_eq!( + span_string_attribute(&http_span, "http.route"), + Some("/{command}"), + "HTTP request span should record bounded route template: {http_span}" + ); + assert_eq!( + span_string_attribute(&http_span, "network.protocol.name"), + Some("http"), + "HTTP request span should record the protocol name: {http_span}" + ); + assert_eq!( + span_int_attribute(&http_span, "http.response.status_code"), + Some(200), + "HTTP request span should record the response status: {http_span}" + ); + assert!( + span_attribute(&http_span, "url.query").is_none() + && span_attribute(&http_span, "url.path").is_none(), + "HTTP request span must not emit raw URL path or query attributes: {http_span}" + ); + assert_eq!( + http_dispatch_span["parentSpanId"].as_str(), + http_span["spanId"].as_str(), + "dispatch span must be a child of the HTTP request span: {http_dispatch_span}" + ); +} + +async fn spawn_http_service(service: Arc) -> String { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind http listener"); + let addr = listener.local_addr().expect("local addr"); + tokio::spawn(async move { + axum::serve(listener, microsvc::router(service)) + .await + .expect("serve http router"); + }); + format!("http://{addr}") } /// Poll the collector's file-exporter output until the framework dispatch span @@ -197,6 +275,25 @@ fn find_span(batch: &Value, trace_id: &str, span_name: &str) -> Option { None } +fn span_attribute<'a>(span: &'a Value, key: &str) -> Option<&'a Value> { + span["attributes"] + .as_array()? + .iter() + .find(|attribute| attribute["key"].as_str() == Some(key)) + .and_then(|attribute| attribute.get("value")) +} + +fn span_string_attribute<'a>(span: &'a Value, key: &str) -> Option<&'a str> { + span_attribute(span, key)?.get("stringValue")?.as_str() +} + +fn span_int_attribute(span: &Value, key: &str) -> Option { + let value = span_attribute(span, key)?.get("intValue")?; + value + .as_i64() + .or_else(|| value.as_str().and_then(|value| value.parse().ok())) +} + fn fresh_trace_id(salt: u128) -> String { format!( "{:032x}", From 82f5a821ac52d7d8e24b99a801a2d6d084d9bf21 Mon Sep 17 00:00:00 2001 From: Patrick Lee Scott Date: Tue, 7 Jul 2026 02:28:00 -0500 Subject: [PATCH 2/2] fix: export HTTP status span attribute as integer Implements [[tasks/distributed-http-telemetry-1]] --- src/http_telemetry.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/http_telemetry.rs b/src/http_telemetry.rs index d6baec51..bb9e2e5d 100644 --- a/src/http_telemetry.rs +++ b/src/http_telemetry.rs @@ -71,7 +71,7 @@ pub(crate) async fn middleware( let status = response.status(); #[cfg(feature = "otel")] - span.record("http.response.status_code", status.as_u16()); + span.record("http.response.status_code", i64::from(status.as_u16())); #[cfg(feature = "metrics")] crate::metrics::record_http_server_request( state.service(),