feat: add telemetry instrumentation
This commit is contained in:
@@ -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"] }
|
||||
|
||||
155
src/app.rs
155
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<PokedexService>,
|
||||
telemetry: Arc<Telemetry>,
|
||||
}
|
||||
|
||||
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<Self, AppError> {
|
||||
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<HealthResponse> {
|
||||
Json(HealthResponse { status: "ok" })
|
||||
}
|
||||
|
||||
async fn metrics(State(state): State<AppState>) -> Result<impl IntoResponse, AppError> {
|
||||
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<AppState>,
|
||||
req: Request,
|
||||
next: middleware::Next,
|
||||
) -> Response {
|
||||
let start = Instant::now();
|
||||
let method = req.method().clone();
|
||||
let path = req.extensions().get::<MatchedPath>().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<AppState>,
|
||||
Path(name): Path<String>,
|
||||
@@ -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"),
|
||||
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<Body> {
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -8,6 +8,7 @@ pub mod domain;
|
||||
pub mod error;
|
||||
pub mod pokemon;
|
||||
pub mod service;
|
||||
pub mod telemetry;
|
||||
pub mod translation;
|
||||
|
||||
#[must_use]
|
||||
|
||||
@@ -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<dyn std::error::Error>> {
|
||||
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)?;
|
||||
|
||||
@@ -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<PokemonInfo, AppError> {
|
||||
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<PokemonInfo, AppError> {
|
||||
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<PokemonInfo, AppError> {
|
||||
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(),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
173
src/telemetry.rs
Normal file
173
src/telemetry.rs
Normal file
@@ -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<u64>,
|
||||
http_errors: opentelemetry::metrics::Counter<u64>,
|
||||
http_request_duration: opentelemetry::metrics::Histogram<f64>,
|
||||
upstream_requests: opentelemetry::metrics::Counter<u64>,
|
||||
upstream_errors: opentelemetry::metrics::Counter<u64>,
|
||||
upstream_request_duration: opentelemetry::metrics::Histogram<f64>,
|
||||
}
|
||||
|
||||
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<String, AppError> {
|
||||
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\""));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user