From e7cf7729ee09a8fe77e07fd98baa6c9e97ff6c4e Mon Sep 17 00:00:00 2001 From: Alberto Moretti Date: Sat, 13 Jun 2026 06:12:51 +0200 Subject: [PATCH] feat: add telemetry instrumentation --- Cargo.toml | 2 +- src/app.rs | 159 ++++++++++++++++++++++++++++++++++++++----- src/lib.rs | 1 + src/main.rs | 3 +- src/service.rs | 45 +++++++++--- src/telemetry.rs | 173 +++++++++++++++++++++++++++++++++++++++++++++++ 6 files changed, 357 insertions(+), 26 deletions(-) create mode 100644 src/telemetry.rs diff --git a/Cargo.toml b/Cargo.toml index 8a669d1..61cac98 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -16,7 +16,7 @@ serde_json = "1" thiserror = "2" tokio = { version = "1", features = ["macros", "net", "rt-multi-thread", "signal"] } tower = { version = "0.5", features = ["util"] } -tower-http = { version = "0.6", features = ["request-id", "trace"] } +tower-http = { version = "0.6", features = ["request-id", "trace", "util"] } tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] } uuid = { version = "1", features = ["v4"] } diff --git a/src/app.rs b/src/app.rs index 273f925..e1fe0aa 100644 --- a/src/app.rs +++ b/src/app.rs @@ -1,48 +1,102 @@ use crate::{ config::AppConfig, error::AppError, pokemon::PokeApiClient, service::PokedexService, - translation::TranslationClient, + telemetry::Telemetry, translation::TranslationClient, }; use axum::{ Json, Router, - extract::{Path, State}, + extract::{MatchedPath, Path, Request, State}, + http::header, + middleware, + response::{IntoResponse, Response}, routing::get, }; use serde::Serialize; -use std::sync::Arc; +use std::{sync::Arc, time::Instant}; +use tower::ServiceBuilder; +use tower_http::{ + ServiceBuilderExt, + request_id::MakeRequestUuid, + trace::{DefaultMakeSpan, DefaultOnResponse, TraceLayer}, +}; #[derive(Clone)] pub struct AppState { service: Arc, + telemetry: Arc, } impl AppState { #[must_use] - pub fn new(service: PokedexService) -> Self { + pub fn new(service: PokedexService, telemetry: Telemetry) -> Self { Self { service: Arc::new(service), + telemetry: Arc::new(telemetry), } } pub fn from_config(config: &AppConfig) -> Result { + let telemetry = Telemetry::new(&config.service_name); let pokemon = PokeApiClient::new(&config.pokeapi_base_url, config.request_timeout)?; let translations = TranslationClient::new(&config.translations_base_url, config.request_timeout)?; - Ok(Self::new(PokedexService::new(pokemon, translations))) + let service = PokedexService::new(pokemon, translations, telemetry.metrics()); + Ok(Self::new(service, telemetry)) } } pub fn create_app(state: AppState) -> Router { + let tracing_layers = ServiceBuilder::new() + .set_x_request_id(MakeRequestUuid) + .layer( + TraceLayer::new_for_http() + .make_span_with(DefaultMakeSpan::new().include_headers(true)) + .on_response(DefaultOnResponse::new().include_headers(true)), + ) + .propagate_x_request_id(); + Router::new() .route("/health", get(health)) + .route("/metrics", get(metrics)) .route("/pokemon/{name}", get(get_pokemon)) .route("/pokemon/translated/{name}", get(get_translated_pokemon)) - .with_state(state) + .with_state(state.clone()) + .layer(middleware::from_fn_with_state(state, record_http_metrics)) + .layer(tracing_layers) } async fn health() -> Json { Json(HealthResponse { status: "ok" }) } +async fn metrics(State(state): State) -> Result { + let body = state.telemetry.render_prometheus()?; + Ok(([(header::CONTENT_TYPE, "text/plain; version=0.0.4")], body)) +} + +async fn record_http_metrics( + State(state): State, + req: Request, + next: middleware::Next, +) -> Response { + let start = Instant::now(); + let method = req.method().clone(); + let path = req.extensions().get::().map_or_else( + || req.uri().path().to_owned(), + |path| path.as_str().to_owned(), + ); + + let response = next.run(req).await; + let status = response.status(); + state.telemetry.metrics().record_http_request( + method.as_str(), + &path, + status.as_u16(), + start.elapsed(), + ); + + response +} + async fn get_pokemon( State(state): State, Path(name): Path, @@ -65,10 +119,13 @@ struct HealthResponse { #[cfg(test)] mod tests { use super::{AppState, create_app}; - use crate::{pokemon::PokeApiClient, service::PokedexService, translation::TranslationClient}; + use crate::{ + pokemon::PokeApiClient, service::PokedexService, telemetry::Telemetry, + translation::TranslationClient, + }; use axum::{ body::{Body, to_bytes}, - http::{Request, StatusCode}, + http::{Request, StatusCode, header::HeaderName}, }; use serde_json::Value; use std::time::Duration; @@ -143,6 +200,62 @@ mod tests { ); } + #[tokio::test] + async fn every_response_has_a_request_id() { + let pokeapi = MockServer::start().await; + let translations = MockServer::start().await; + let app = create_app(state_for(&pokeapi, &translations)); + + let response = app + .oneshot(request("/health")) + .await + .expect("request should be handled"); + + assert!(response.headers().contains_key("x-request-id")); + } + + #[tokio::test] + async fn keeps_incoming_request_id_when_present() { + let pokeapi = MockServer::start().await; + let translations = MockServer::start().await; + let app = create_app(state_for(&pokeapi, &translations)); + let request = Request::builder() + .uri("/health") + .header(HeaderName::from_static("x-request-id"), "external-id") + .body(Body::empty()) + .expect("request should build"); + + let response = app + .oneshot(request) + .await + .expect("request should be handled"); + + assert_eq!(response.headers()["x-request-id"], "external-id"); + } + + #[tokio::test] + async fn metrics_endpoint_exposes_http_and_upstream_otel_metrics() { + let pokeapi = MockServer::start().await; + let translations = MockServer::start().await; + mount_species(&pokeapi, "mewtwo", true, "rare", "Created by science.").await; + let app = create_app(state_for(&pokeapi, &translations)); + let response = app + .clone() + .oneshot(request("/pokemon/mewtwo")) + .await + .expect("request should be handled"); + assert_eq!(response.status(), StatusCode::OK); + + let response = app + .oneshot(request("/metrics")) + .await + .expect("request should be handled"); + let body = text_body(response).await; + + assert!(body.contains("http_server_requests")); + assert!(body.contains("upstream_client_requests")); + } + #[tokio::test] async fn pokemon_endpoint_maps_missing_pokemon_to_404() { let pokeapi = MockServer::start().await; @@ -185,11 +298,17 @@ mod tests { } fn state_for(pokeapi: &MockServer, translations: &MockServer) -> AppState { - AppState::new(PokedexService::new( - PokeApiClient::new(pokeapi.uri(), Duration::from_secs(1)).expect("client should build"), - TranslationClient::new(translations.uri(), Duration::from_secs(1)) - .expect("client should build"), - )) + let telemetry = Telemetry::new("test-service"); + AppState::new( + PokedexService::new( + PokeApiClient::new(pokeapi.uri(), Duration::from_secs(1)) + .expect("client should build"), + TranslationClient::new(translations.uri(), Duration::from_secs(1)) + .expect("client should build"), + telemetry.metrics(), + ), + telemetry, + ) } fn request(path: &str) -> Request { @@ -200,9 +319,17 @@ mod tests { } async fn json_body(response: axum::response::Response) -> Value { - let bytes = to_bytes(response.into_body(), 1024 * 1024) - .await - .expect("body should be readable"); + let bytes = body_bytes(response).await; serde_json::from_slice(&bytes).expect("body should be valid JSON") } + + async fn text_body(response: axum::response::Response) -> String { + String::from_utf8(body_bytes(response).await.to_vec()).expect("body should be valid UTF-8") + } + + async fn body_bytes(response: axum::response::Response) -> axum::body::Bytes { + to_bytes(response.into_body(), 1024 * 1024) + .await + .expect("body should be readable") + } } diff --git a/src/lib.rs b/src/lib.rs index 96d546a..b397c01 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -8,6 +8,7 @@ pub mod domain; pub mod error; pub mod pokemon; pub mod service; +pub mod telemetry; pub mod translation; #[must_use] diff --git a/src/main.rs b/src/main.rs index db6ae50..861c8fa 100644 --- a/src/main.rs +++ b/src/main.rs @@ -2,10 +2,11 @@ #![deny(clippy::all, clippy::pedantic, clippy::nursery, clippy::cargo)] #![allow(clippy::missing_errors_doc, clippy::multiple_crate_versions)] -use pokedex_api::{app, config::AppConfig}; +use pokedex_api::{app, config::AppConfig, telemetry}; #[tokio::main] async fn main() -> Result<(), Box> { + telemetry::init_tracing(); let config = AppConfig::from_env()?; let listener = tokio::net::TcpListener::bind(config.bind_addr).await?; let state = app::AppState::from_config(&config)?; diff --git a/src/service.rs b/src/service.rs index 4078c1d..766442c 100644 --- a/src/service.rs +++ b/src/service.rs @@ -2,46 +2,74 @@ use crate::{ domain::PokemonInfo, error::AppError, pokemon::PokeApiClient, + telemetry::AppMetrics, translation::{TranslationClient, TranslationKind}, }; +use std::time::Instant; use tracing::{debug, instrument}; #[derive(Clone)] pub struct PokedexService { pokemon: PokeApiClient, translations: TranslationClient, + metrics: AppMetrics, } impl PokedexService { #[must_use] - pub const fn new(pokemon: PokeApiClient, translations: TranslationClient) -> Self { + pub const fn new( + pokemon: PokeApiClient, + translations: TranslationClient, + metrics: AppMetrics, + ) -> Self { Self { pokemon, translations, + metrics, } } #[instrument(skip(self), fields(pokemon.name = %name))] pub async fn get_basic(&self, name: &str) -> Result { - self.pokemon.get_pokemon_info(name).await + self.fetch_pokemon(name).await } #[instrument(skip(self), fields(pokemon.name = %name))] pub async fn get_translated(&self, name: &str) -> Result { - let mut pokemon = self.pokemon.get_pokemon_info(name).await?; + let mut pokemon = self.fetch_pokemon(name).await?; let translation_kind = translation_kind_for(&pokemon); - match self + let translation_start = Instant::now(); + let translation = self .translations .translate(translation_kind, &pokemon.description) - .await - { + .await; + self.metrics.record_upstream_request( + "funtranslations", + translation_kind.endpoint(), + translation.is_ok(), + translation_start.elapsed(), + ); + + match translation { Ok(translated) => pokemon.description = translated, Err(error) => debug!(%error, "using original description after translation failure"), } Ok(pokemon) } + + async fn fetch_pokemon(&self, name: &str) -> Result { + let start = Instant::now(); + let pokemon = self.pokemon.get_pokemon_info(name).await; + self.metrics.record_upstream_request( + "pokeapi", + "pokemon_species", + pokemon.is_ok(), + start.elapsed(), + ); + pokemon + } } #[must_use] @@ -57,8 +85,8 @@ pub fn translation_kind_for(pokemon: &PokemonInfo) -> TranslationKind { mod tests { use super::{PokedexService, translation_kind_for}; use crate::{ - domain::PokemonInfo, pokemon::PokeApiClient, translation::TranslationClient, - translation::TranslationKind, + domain::PokemonInfo, pokemon::PokeApiClient, telemetry::Telemetry, + translation::TranslationClient, translation::TranslationKind, }; use std::time::Duration; use wiremock::{Mock, MockServer, ResponseTemplate, matchers}; @@ -188,6 +216,7 @@ mod tests { PokeApiClient::new(pokeapi.uri(), Duration::from_secs(1)).expect("client should build"), TranslationClient::new(translations.uri(), Duration::from_secs(1)) .expect("client should build"), + Telemetry::new("test-service").metrics(), ) } } diff --git a/src/telemetry.rs b/src/telemetry.rs new file mode 100644 index 0000000..2c09abc --- /dev/null +++ b/src/telemetry.rs @@ -0,0 +1,173 @@ +use crate::error::AppError; +use opentelemetry::{KeyValue, metrics::MeterProvider}; +use opentelemetry_prometheus_text_exporter::PrometheusExporter; +use opentelemetry_sdk::{Resource, metrics::SdkMeterProvider}; +use std::time::Duration; +use tracing_subscriber::{EnvFilter, layer::SubscriberExt, util::SubscriberInitExt}; + +#[derive(Clone, Debug)] +pub struct AppMetrics { + http_requests: opentelemetry::metrics::Counter, + http_errors: opentelemetry::metrics::Counter, + http_request_duration: opentelemetry::metrics::Histogram, + upstream_requests: opentelemetry::metrics::Counter, + upstream_errors: opentelemetry::metrics::Counter, + upstream_request_duration: opentelemetry::metrics::Histogram, +} + +impl AppMetrics { + pub fn record_http_request(&self, method: &str, path: &str, status: u16, duration: Duration) { + let is_error = status_is_error(status); + let status = status.to_string(); + let attributes = [ + KeyValue::new("method", method.to_owned()), + KeyValue::new("path", path.to_owned()), + KeyValue::new("status", status), + ]; + self.http_requests.add(1, &attributes); + self.http_request_duration + .record(duration.as_secs_f64(), &attributes); + if is_error { + self.http_errors.add(1, &attributes); + } + } + + pub fn record_upstream_request( + &self, + service: &'static str, + operation: &'static str, + success: bool, + duration: Duration, + ) { + let status = if success { "ok" } else { "error" }; + let attributes = [ + KeyValue::new("service", service), + KeyValue::new("operation", operation), + KeyValue::new("status", status), + ]; + self.upstream_requests.add(1, &attributes); + self.upstream_request_duration + .record(duration.as_secs_f64(), &attributes); + if !success { + self.upstream_errors.add(1, &attributes); + } + } +} + +pub struct Telemetry { + provider: SdkMeterProvider, + exporter: PrometheusExporter, + metrics: AppMetrics, +} + +impl Telemetry { + #[must_use] + pub fn new(service_name: &str) -> Self { + let exporter = PrometheusExporter::builder().build(); + let provider = SdkMeterProvider::builder() + .with_reader(exporter.clone()) + .with_resource( + Resource::builder_empty() + .with_service_name(service_name.to_owned()) + .build(), + ) + .build(); + let meter = provider.meter(env!("CARGO_PKG_NAME")); + let metrics = AppMetrics { + http_requests: meter + .u64_counter("http.server.requests") + .with_description("Total HTTP requests") + .with_unit("requests") + .build(), + http_errors: meter + .u64_counter("http.server.errors") + .with_description("Total HTTP requests with error status codes") + .with_unit("errors") + .build(), + http_request_duration: meter + .f64_histogram("http.server.request.duration") + .with_description("HTTP request duration") + .with_unit("s") + .build(), + upstream_requests: meter + .u64_counter("upstream.client.requests") + .with_description("Total upstream requests") + .with_unit("requests") + .build(), + upstream_errors: meter + .u64_counter("upstream.client.errors") + .with_description("Total failed upstream requests") + .with_unit("errors") + .build(), + upstream_request_duration: meter + .f64_histogram("upstream.client.request.duration") + .with_description("Upstream request duration") + .with_unit("s") + .build(), + }; + + Self { + provider, + exporter, + metrics, + } + } + + #[must_use] + pub fn metrics(&self) -> AppMetrics { + self.metrics.clone() + } + + pub fn render_prometheus(&self) -> Result { + let mut output = Vec::new(); + self.exporter + .export(&mut output) + .map_err(|_| AppError::Internal)?; + String::from_utf8(output).map_err(|_| AppError::Internal) + } + + pub fn shutdown(self) -> Result<(), AppError> { + self.provider.shutdown().map_err(|_| AppError::Internal) + } +} + +pub fn init_tracing() { + let env_filter = EnvFilter::try_from_default_env() + .unwrap_or_else(|_| EnvFilter::new("pokedex_api=info,tower_http=info")); + + tracing_subscriber::registry() + .with(env_filter) + .with(tracing_subscriber::fmt::layer().json()) + .init(); +} + +const fn status_is_error(status: u16) -> bool { + status >= 400 +} + +#[cfg(test)] +mod tests { + use super::Telemetry; + use std::time::Duration; + + #[test] + fn records_http_and_upstream_metrics_in_prometheus_text_format() { + let telemetry = Telemetry::new("test-service"); + let metrics = telemetry.metrics(); + + metrics.record_http_request("GET", "/pokemon/{name}", 200, Duration::from_millis(10)); + metrics.record_upstream_request( + "pokeapi", + "pokemon_species", + true, + Duration::from_millis(5), + ); + let rendered = telemetry + .render_prometheus() + .expect("metrics should render as text"); + + assert!(rendered.contains("http_server_requests")); + assert!(rendered.contains("upstream_client_requests")); + assert!(rendered.contains("service_name=\"test-service\"")); + } +}