diff --git a/Cargo.lock b/Cargo.lock index 905a003c..113f68a4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3376,6 +3376,12 @@ dependencies = [ "tonic-prost", ] +[[package]] +name = "opentelemetry-semantic-conventions" +version = "0.31.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e62e29dfe041afb8ed2a6c9737ab57db4907285d999ef8ad3a59092a36bdc846" + [[package]] name = "opentelemetry_sdk" version = "0.31.0" @@ -4612,6 +4618,7 @@ dependencies = [ "openidconnect", "opentelemetry", "opentelemetry-otlp", + "opentelemetry-semantic-conventions", "opentelemetry_sdk", "password-hash", "percent-encoding", diff --git a/Cargo.toml b/Cargo.toml index 83d26619..e51cc5bd 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -88,8 +88,9 @@ tracing-opentelemetry = "0.32" tracing-actix-web = { version = "0.7", default-features = false, features = ["opentelemetry_0_31"] } tracing-log = "0.2" opentelemetry = "0.31" -opentelemetry_sdk = { version = "0.31", features = ["rt-tokio-current-thread"] } -opentelemetry-otlp = { version = "0.31", features = ["http-proto", "grpc-tonic"] } +opentelemetry_sdk = { version = "0.31", features = ["metrics", "rt-tokio-current-thread", "spec_unstable_metrics_views"] } +opentelemetry-otlp = { version = "0.31", features = ["http-proto", "grpc-tonic", "metrics"] } +opentelemetry-semantic-conventions = { version = "0.31", features = ["semconv_experimental"] } [features] diff --git a/examples/telemetry/docker-compose.yml b/examples/telemetry/docker-compose.yml index fcc2b776..e4640c06 100644 --- a/examples/telemetry/docker-compose.yml +++ b/examples/telemetry/docker-compose.yml @@ -23,6 +23,7 @@ services: environment: - DATABASE_URL=postgres://sqlpage:sqlpage@postgres:5432/sqlpage - OTEL_EXPORTER_OTLP_ENDPOINT=http://otel-collector:4318 + - OTEL_METRIC_EXPORT_INTERVAL=1000 - OTEL_SERVICE_NAME=sqlpage volumes: - ./website:/var/www @@ -113,8 +114,10 @@ services: - "1514:1514/udp" - "1516:1516/udp" depends_on: - - tempo - - postgres + tempo: + condition: service_started + postgres: + condition: service_started loki: condition: service_healthy diff --git a/examples/telemetry/grafana/sqlpage-home.json b/examples/telemetry/grafana/sqlpage-home.json index 6c19ad36..3cd35cb9 100644 --- a/examples/telemetry/grafana/sqlpage-home.json +++ b/examples/telemetry/grafana/sqlpage-home.json @@ -39,7 +39,7 @@ "showLineNumbers": false, "showMiniMap": false }, - "content": "

Recent SQLPage traces, logs, and PostgreSQL metrics

Open http://localhost and interact with the app. New requests will appear here automatically.

The trace table shows recent requests. Click any trace ID to open the full span waterfall in Grafana. PostgreSQL slow-query explain plans appear in the PostgreSQL Logs panel and link back to the same trace via the extracted trace ID. The metrics panels come from the OpenTelemetry PostgreSQL receiver via Prometheus.

", + "content": "

SQLPage Observability

Open http://localhost and interact with the app. New requests will appear here automatically.

This dashboard shows traces, logs, and application metrics exported by SQLPage. Trace waterfalls link to PostgreSQL logs via trace IDs. Metrics include HTTP durations, DB query latencies, and connection pool states.

", "mode": "html" }, "pluginVersion": "12.4.0", @@ -54,7 +54,39 @@ "fieldConfig": { "defaults": { "color": { - "mode": "thresholds" + "mode": "palette-classic" + }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 10, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 2, + "pointSize": 4, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "never", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } }, "mappings": [], "thresholds": { @@ -63,38 +95,31 @@ { "color": "green", "value": null - }, - { - "color": "orange", - "value": 10 } ] }, - "unit": "none" + "unit": "s" }, "overrides": [] }, "gridPos": { - "h": 4, + "h": 8, "w": 12, "x": 0, "y": 4 }, - "id": 4, + "id": 10, "options": { - "colorMode": "value", - "graphMode": "none", - "justifyMode": "auto", - "orientation": "auto", - "percentChangeColorMode": "standard", - "reduceOptions": { - "calcs": ["lastNotNull"], - "fields": "", - "values": false + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": true }, - "showPercentChange": false, - "textMode": "auto", - "wideLayout": true + "tooltip": { + "mode": "single", + "sort": "none" + } }, "pluginVersion": "12.4.0", "targets": [ @@ -103,12 +128,22 @@ "type": "prometheus", "uid": "prometheus" }, - "expr": "sum(postgresql_backends)", + "expr": "histogram_quantile(0.95, sum(rate(http_server_request_duration_seconds_bucket[5m])) by (le, http_route))", + "legendFormat": "HTTP P95 {{http_route}}", "refId": "A" + }, + { + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, + "expr": "histogram_quantile(0.95, sum(rate(db_client_operation_duration_seconds_bucket[5m])) by (le, db_operation_name))", + "legendFormat": "DB P95 {{db_operation_name}}", + "refId": "B" } ], - "title": "PostgreSQL Backends", - "type": "stat" + "title": "Request & Query Latency (P95)", + "type": "timeseries" }, { "datasource": { @@ -162,17 +197,17 @@ } ] }, - "unit": "bytes" + "unit": "none" }, "overrides": [] }, "gridPos": { - "h": 4, + "h": 8, "w": 12, "x": 12, "y": 4 }, - "id": 5, + "id": 11, "options": { "legend": { "calcs": [], @@ -192,12 +227,12 @@ "type": "prometheus", "uid": "prometheus" }, - "expr": "sum(postgresql_db_size) by (postgresql_database_name)", - "legendFormat": "{{postgresql_database_name}}", + "expr": "sum(db_client_connection_count) by (db_client_connection_state)", + "legendFormat": "{{db_client_connection_state}}", "refId": "A" } ], - "title": "PostgreSQL Database Size", + "title": "SQLPage DB Connection Pool", "type": "timeseries" }, { @@ -223,8 +258,8 @@ }, "properties": [ { - "id": "custom.width", - "value": 300 + "id": "custom.hidden", + "value": true } ] }, @@ -248,7 +283,35 @@ "properties": [ { "id": "custom.width", - "value": 140 + "value": 120 + } + ] + }, + { + "matcher": { + "id": "byName", + "options": "traceName" + }, + "properties": [ + { + "id": "custom.width", + "value": 520 + }, + { + "id": "custom.cellOptions", + "value": { + "type": "data-links" + } + }, + { + "id": "links", + "value": [ + { + "targetBlank": false, + "title": "${__value.text}", + "url": "/a/grafana-exploretraces-app/explore?traceId=${__data.fields.traceID}" + } + ] } ] }, @@ -279,10 +342,10 @@ ] }, "gridPos": { - "h": 12, + "h": 8, "w": 24, "x": 0, - "y": 8 + "y": 12 }, "id": 2, "options": { @@ -335,7 +398,7 @@ "renameByName": { "startTime": "Start time", "traceDuration": "Duration", - "traceID": "Trace ID", + "traceID": "Trace", "traceName": "Route", "traceService": "Service" } @@ -344,6 +407,36 @@ ], "type": "table" }, + { + "datasource": { + "type": "tempo", + "uid": "tempo" + }, + "gridPos": { + "h": 10, + "w": 24, + "x": 0, + "y": 30 + }, + "id": 12, + "pluginVersion": "12.4.0", + "targets": [ + { + "datasource": { + "type": "tempo", + "uid": "tempo" + }, + "limit": 20, + "query": "$traceId", + "queryType": "traceql", + "refId": "A", + "tableType": "traces" + } + ], + "timeFrom": "1h", + "title": "Selected Trace", + "type": "traces" + }, { "datasource": { "type": "loki", @@ -428,9 +521,30 @@ "refresh": "5s", "schemaVersion": 41, "style": "dark", - "tags": ["sqlpage", "tracing", "logs"], + "tags": ["sqlpage", "tracing", "logs", "metrics"], "templating": { - "list": [] + "list": [ + { + "current": { + "selected": true, + "text": "", + "value": "" + }, + "hide": 2, + "label": "Trace ID", + "name": "traceId", + "options": [ + { + "selected": true, + "text": "", + "value": "" + } + ], + "query": "", + "skipUrlSync": false, + "type": "textbox" + } + ] }, "time": { "from": "now-1h", @@ -440,5 +554,5 @@ "timezone": "browser", "title": "SQLPage Observability Home", "uid": "sqlpage-tracing-home", - "version": 5 + "version": 6 } diff --git a/examples/telemetry/otel-collector.yaml b/examples/telemetry/otel-collector.yaml index 032d2919..36325199 100644 --- a/examples/telemetry/otel-collector.yaml +++ b/examples/telemetry/otel-collector.yaml @@ -118,22 +118,41 @@ exporters: service: pipelines: traces: - receivers: [otlp] - processors: [batch] - exporters: [otlp_grpc/tempo] + receivers: + - otlp + processors: + - batch + exporters: + - otlp_grpc/tempo metrics: - receivers: [postgresql] - processors: [batch] - exporters: [prometheus] + receivers: + - otlp + - postgresql + processors: + - batch + exporters: + - prometheus logs/sqlpage: - receivers: [syslog/sqlpage] - processors: [transform/sqlpage_logs, batch] - exporters: [otlp_http/loki] + receivers: + - syslog/sqlpage + processors: + - transform/sqlpage_logs + - batch + exporters: + - otlp_http/loki logs/postgresql: - receivers: [filelog/postgresql] - processors: [transform/postgresql_logs, batch] - exporters: [otlp_http/loki] + receivers: + - filelog/postgresql + processors: + - transform/postgresql_logs + - batch + exporters: + - otlp_http/loki logs/nginx: - receivers: [syslog/nginx] - processors: [transform/nginx_logs, batch] - exporters: [otlp_http/loki] + receivers: + - syslog/nginx + processors: + - transform/nginx_logs + - batch + exporters: + - otlp_http/loki diff --git a/src/lib.rs b/src/lib.rs index a4babf3c..d0c13b3f 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -77,6 +77,7 @@ pub mod file_cache; pub mod filesystem; pub mod render; pub mod telemetry; +pub mod telemetry_metrics; pub mod template_helpers; pub mod templates; pub mod utils; @@ -89,6 +90,7 @@ use crate::webserver::oidc::OidcState; use file_cache::FileCache; use std::path::{Path, PathBuf}; use std::sync::Arc; +use telemetry_metrics::TelemetryMetrics; use templates::AllTemplates; use webserver::Database; @@ -108,6 +110,7 @@ pub struct AppState { file_system: FileSystem, config: AppConfig, pub oidc_state: Option>, + pub telemetry_metrics: TelemetryMetrics, } impl AppState { @@ -133,6 +136,8 @@ impl AppState { ); let oidc_state = crate::webserver::oidc::initialize_oidc_state(config).await?; + let telemetry_metrics = + TelemetryMetrics::new(&db.connection, db.info.database_type.otel_name()); Ok(AppState { db, @@ -141,6 +146,7 @@ impl AppState { file_system, config: config.clone(), oidc_state, + telemetry_metrics, }) } } diff --git a/src/telemetry.rs b/src/telemetry.rs index 2c11921d..46a47757 100644 --- a/src/telemetry.rs +++ b/src/telemetry.rs @@ -9,9 +9,11 @@ use std::env; use std::sync::OnceLock; +use opentelemetry_sdk::metrics::SdkMeterProvider; use opentelemetry_sdk::trace::SdkTracerProvider; static TRACER_PROVIDER: OnceLock = OnceLock::new(); +static METER_PROVIDER: OnceLock = OnceLock::new(); /// Initializes logging / tracing. Returns `true` if `OTel` was activated. #[must_use] @@ -35,6 +37,11 @@ pub fn shutdown_telemetry() { eprintln!("Error shutting down tracer provider: {e}"); } } + if let Some(provider) = METER_PROVIDER.get() { + if let Err(e) = provider.shutdown() { + eprintln!("Error shutting down meter provider: {e}"); + } + } } /// Tracing subscriber without `OTel` export — logfmt output only. @@ -71,6 +78,39 @@ fn init_otel_tracing() { global::set_tracer_provider(provider.clone()); let _ = TRACER_PROVIDER.set(provider); + // OTLP Metric exporter + let metric_exporter = opentelemetry_otlp::MetricExporter::builder() + .with_http() + .build() + .expect("Failed to build OTLP metric exporter"); + + let reader = opentelemetry_sdk::metrics::PeriodicReader::builder(metric_exporter).build(); + let meter_provider = SdkMeterProvider::builder() + .with_reader(reader) + .with_view(|instrument: &opentelemetry_sdk::metrics::Instrument| { + if instrument.kind() == opentelemetry_sdk::metrics::InstrumentKind::Histogram { + Some( + opentelemetry_sdk::metrics::Stream::builder() + .with_aggregation( + opentelemetry_sdk::metrics::Aggregation::ExplicitBucketHistogram { + boundaries: vec![ + 0.001, 0.005, 0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5, 0.75, + 1.0, 2.5, 5.0, 7.5, 10.0, + ], + record_min_max: true, + }, + ) + .build() + .expect("Failed to build metrics stream"), + ) + } else { + None + } + }) + .build(); + global::set_meter_provider(meter_provider.clone()); + let _ = METER_PROVIDER.set(meter_provider.clone()); + let otel_layer = tracing_opentelemetry::layer() .with_tracer(tracer) .with_location(false); @@ -78,7 +118,8 @@ fn init_otel_tracing() { let subscriber = tracing_subscriber::registry() .with(default_env_filter()) .with(logfmt::LogfmtLayer::new()) - .with(otel_layer); + .with(otel_layer) + .with(tracing_opentelemetry::MetricsLayer::new(meter_provider)); set_global_subscriber(subscriber); } @@ -153,14 +194,15 @@ mod logfmt { } } + use opentelemetry_semantic_conventions::attribute as otel; /// Fields we pick from spans, in display order. /// (`span_field_name`, `logfmt_key`) const SPAN_FIELDS: &[(&str, &str)] = &[ - ("http.method", "method"), - ("http.target", "path"), - ("http.status_code", "status"), + (otel::HTTP_REQUEST_METHOD, "method"), + (otel::URL_PATH, "path"), + (otel::HTTP_RESPONSE_STATUS_CODE, "status"), ("sqlpage.file", "file"), - ("http.client_ip", "client_ip"), + (otel::CLIENT_ADDRESS, "client_ip"), ]; /// All-zeros trace ID means no real trace context. @@ -467,8 +509,8 @@ mod logfmt { fn debug_logs_include_unmapped_span_fields() { let mut buf = String::new(); let span_fields = HashMap::from([ - ("http.method", "GET".to_string()), - ("http.route", "/users/:id".to_string()), + (otel::HTTP_REQUEST_METHOD, "GET".to_string()), + (otel::HTTP_ROUTE, "/users/:id".to_string()), ("otel.kind", "server".to_string()), ]); @@ -481,8 +523,8 @@ mod logfmt { fn info_logs_keep_only_mapped_span_fields_when_not_in_debug_mode() { let mut buf = String::new(); let span_fields = HashMap::from([ - ("http.method", "GET".to_string()), - ("http.route", "/users/:id".to_string()), + (otel::HTTP_REQUEST_METHOD, "GET".to_string()), + (otel::HTTP_ROUTE, "/users/:id".to_string()), ("otel.kind", "server".to_string()), ]); diff --git a/src/telemetry_metrics.rs b/src/telemetry_metrics.rs new file mode 100644 index 00000000..89974438 --- /dev/null +++ b/src/telemetry_metrics.rs @@ -0,0 +1,97 @@ +use opentelemetry::global; +use opentelemetry::metrics::{Histogram, ObservableGauge}; +use opentelemetry_semantic_conventions::attribute as otel; +use opentelemetry_semantic_conventions::metric as otel_metric; +use sqlx::AnyPool; + +pub struct TelemetryMetrics { + pub http_request_duration: Histogram, + pub db_query_duration: Histogram, + _pool_connection_count: ObservableGauge, +} + +impl Default for TelemetryMetrics { + fn default() -> Self { + let meter = global::meter("sqlpage"); + let http_request_duration = meter + .f64_histogram(otel_metric::HTTP_SERVER_REQUEST_DURATION) + .with_unit("s") + .with_description("Duration of HTTP requests processed by the server.") + .build(); + let db_query_duration = meter + .f64_histogram(otel_metric::DB_CLIENT_OPERATION_DURATION) + .with_unit("s") + .with_description("Duration of executing SQL queries.") + .build(); + // This default is only used in tests that don't touch pool metrics. + let pool_connection_count = meter + .i64_observable_gauge(otel_metric::DB_CLIENT_CONNECTION_COUNT) + .with_unit("{connection}") + .with_description("Number of connections in the database pool.") + .with_callback(|_| {}) + .build(); + + Self { + http_request_duration, + db_query_duration, + _pool_connection_count: pool_connection_count, + } + } +} + +impl TelemetryMetrics { + #[must_use] + pub fn new(pool: &AnyPool, db_system_name: &'static str) -> Self { + let meter = global::meter("sqlpage"); + let http_request_duration = meter + .f64_histogram(otel_metric::HTTP_SERVER_REQUEST_DURATION) + .with_unit("s") + .with_description("Duration of HTTP requests processed by the server.") + .build(); + let db_query_duration = meter + .f64_histogram(otel_metric::DB_CLIENT_OPERATION_DURATION) + .with_unit("s") + .with_description("Duration of executing SQL queries.") + .build(); + let pool_ref = pool.clone(); + let pool_connection_count = meter + .i64_observable_gauge(otel_metric::DB_CLIENT_CONNECTION_COUNT) + .with_unit("{connection}") + .with_description("Number of connections in the database pool.") + .with_callback(move |observer| { + let size = pool_ref.size(); + let idle_u32 = u32::try_from(pool_ref.num_idle()).unwrap_or(u32::MAX); + let used = i64::from(size.saturating_sub(idle_u32)); + let idle = i64::from(idle_u32); + observer.observe( + used, + &[ + opentelemetry::KeyValue::new(otel::DB_SYSTEM_NAME, db_system_name), + opentelemetry::KeyValue::new( + otel::DB_CLIENT_CONNECTION_POOL_NAME, + "sqlpage", + ), + opentelemetry::KeyValue::new(otel::DB_CLIENT_CONNECTION_STATE, "used"), + ], + ); + observer.observe( + idle, + &[ + opentelemetry::KeyValue::new(otel::DB_SYSTEM_NAME, db_system_name), + opentelemetry::KeyValue::new( + otel::DB_CLIENT_CONNECTION_POOL_NAME, + "sqlpage", + ), + opentelemetry::KeyValue::new(otel::DB_CLIENT_CONNECTION_STATE, "idle"), + ], + ); + }) + .build(); + + Self { + http_request_duration, + db_query_duration, + _pool_connection_count: pool_connection_count, + } + } +} diff --git a/src/webserver/database/connect.rs b/src/webserver/database/connect.rs index 573df2fa..dfb49654 100644 --- a/src/webserver/database/connect.rs +++ b/src/webserver/database/connect.rs @@ -10,7 +10,7 @@ use anyhow::Context; use futures_util::future::BoxFuture; use sqlx::odbc::OdbcConnectOptions; use sqlx::{ - any::{Any, AnyConnectOptions, AnyKind}, + any::{Any, AnyConnectOptions, AnyConnection, AnyKind}, pool::PoolOptions, sqlite::{Function, SqliteConnectOptions, SqliteFunctionCtx}, ConnectOptions, Connection, Executor, @@ -38,12 +38,9 @@ impl Database { set_custom_connect_options(&mut connect_options, config); log::debug!("Connecting to database: {database_url}"); let mut retries = config.database_connection_retries; - let db_kind = connect_options.kind(); - let pool = loop { - match Self::create_pool_options(config, db_kind) - .connect_with(connect_options.clone()) - .await - { + + let mut conn: AnyConnection = loop { + match AnyConnection::connect_with(&connect_options).await { Ok(c) => break c, Err(e) => { if retries == 0 { @@ -56,8 +53,16 @@ impl Database { } } }; - let dbms_name: String = pool.acquire().await?.dbms_name().await?; + let dbms_name: String = conn.dbms_name().await?; let database_type = SupportedDatabase::from_dbms_name(&dbms_name); + drop(conn); + + let db_kind = connect_options.kind(); + let pool = Self::create_pool_options(config, db_kind) + .connect_with(connect_options) + .await + .with_context(|| format!("Unable to open connection pool to {database_url}"))?; + log::debug!("Initialized {dbms_name:?} database pool: {pool:#?}"); Ok(Database { connection: pool, @@ -101,32 +106,43 @@ impl Database { fn add_on_return_to_pool(config: &AppConfig, pool_options: PoolOptions) -> PoolOptions { let on_disconnect_file = config.configuration_directory.join(ON_RESET_FILE); - if !on_disconnect_file.exists() { + let sql = if on_disconnect_file.exists() { + log::info!( + "Creating a custom SQL connection cleanup handler from {}", + on_disconnect_file.display() + ); + match std::fs::read_to_string(&on_disconnect_file) { + Ok(sql) => { + log::trace!("The custom SQL connection cleanup handler is:\n{sql}"); + Some(std::sync::Arc::new(sql)) + } + Err(e) => { + log::error!( + "Unable to read the file {}: {}", + on_disconnect_file.display(), + e + ); + None + } + } + } else { log::debug!( "Not creating a custom SQL connection cleanup handler because {} does not exist", on_disconnect_file.display() ); - return pool_options; - } - log::info!( - "Creating a custom SQL connection cleanup handler from {}", - on_disconnect_file.display() - ); - let sql = match std::fs::read_to_string(&on_disconnect_file) { - Ok(sql) => std::sync::Arc::new(sql), - Err(e) => { - log::error!( - "Unable to read the file {}: {}", - on_disconnect_file.display(), - e - ); - return pool_options; - } + None }; - log::trace!("The custom SQL connection cleanup handler is:\n{sql}"); - let sql = sql.clone(); - pool_options - .after_release(move |conn, meta| on_return_to_pool(conn, meta, std::sync::Arc::clone(&sql))) + + pool_options.after_release(move |conn, meta| { + let sql = sql.clone(); + Box::pin(async move { + if let Some(sql) = sql { + on_return_to_pool(conn, meta, sql).await + } else { + Ok(true) + } + }) + }) } fn on_return_to_pool( @@ -154,35 +170,43 @@ fn add_on_connection_handler( pool_options: PoolOptions, ) -> PoolOptions { let on_connect_file = config.configuration_directory.join(ON_CONNECT_FILE); - if !on_connect_file.exists() { + let on_connect_file_display = on_connect_file.display().to_string(); + let sql = if on_connect_file.exists() { + log::info!( + "Creating a custom SQL database connection handler from {}", + on_connect_file.display() + ); + match std::fs::read_to_string(&on_connect_file) { + Ok(sql) => { + log::trace!("The custom SQL database connection handler is:\n{sql}"); + Some(std::sync::Arc::new(sql)) + } + Err(e) => { + log::error!( + "Unable to read the file {}: {}", + on_connect_file.display(), + e + ); + None + } + } + } else { log::debug!( "Not creating a custom SQL database connection handler because {} does not exist", on_connect_file.display() ); - return pool_options; - } - log::info!( - "Creating a custom SQL database connection handler from {}", - on_connect_file.display() - ); - let sql = match std::fs::read_to_string(&on_connect_file) { - Ok(sql) => std::sync::Arc::new(sql), - Err(e) => { - log::error!( - "Unable to read the file {}: {}", - on_connect_file.display(), - e - ); - return pool_options; - } + None }; - log::trace!("The custom SQL database connection handler is:\n{sql}"); - pool_options.after_connect(move |conn, _metadata| { - log::debug!("Running {} on new connection", on_connect_file.display()); - let sql = std::sync::Arc::clone(&sql); + + pool_options.after_connect(move |conn, _| { + let sql = sql.clone(); + let on_connect_file_display = on_connect_file_display.clone(); Box::pin(async move { - let r = conn.execute(sql.as_str()).await?; - log::debug!("Finished running connection handler on new connection: {r:?}"); + if let Some(sql) = sql { + log::debug!("Running {on_connect_file_display} on new connection"); + let r = conn.execute(sql.as_str()).await?; + log::debug!("Finished running connection handler on new connection: {r:?}"); + } Ok(()) }) }) diff --git a/src/webserver/database/execute_queries.rs b/src/webserver/database/execute_queries.rs index fed185f5..2e29b4e8 100644 --- a/src/webserver/database/execute_queries.rs +++ b/src/webserver/database/execute_queries.rs @@ -30,10 +30,17 @@ use sqlx::{ pub type DbConn = Option>; +fn source_line_number(line: usize) -> i64 { + i64::try_from(line).unwrap_or(i64::MAX) +} + +use crate::telemetry_metrics::TelemetryMetrics; +use opentelemetry_semantic_conventions::attribute as otel; + fn record_query_params(span: &tracing::Span, params: &[Option]) { use tracing_opentelemetry::OpenTelemetrySpanExt; for (idx, value) in params.iter().enumerate() { - let key = opentelemetry::Key::new(format!("db.query.parameter.{idx}")); + let key = opentelemetry::Key::new(format!("{}.{idx}", otel::DB_QUERY_PARAMETER)); let otel_value = match value { Some(v) => opentelemetry::Value::String(v.clone().into()), None => opentelemetry::Value::String("NULL".into()), @@ -42,20 +49,88 @@ fn record_query_params(span: &tracing::Span, params: &[Option]) { } } -fn source_line_number(line: usize) -> i64 { - i64::try_from(line).unwrap_or(i64::MAX) +struct DbQueryMetricsContext<'a> { + span: tracing::Span, + duration: std::time::Duration, + db_system_name: &'static str, + operation_name: String, + metrics: &'a TelemetryMetrics, } -fn record_db_query_success(span: &tracing::Span, returned_rows: i64) { - span.record("db.response.returned_rows", returned_rows); - span.record("otel.status_code", "OK"); +impl<'a> DbQueryMetricsContext<'a> { + fn new( + span: tracing::Span, + operation_name: String, + db_system_name: &'static str, + metrics: &'a TelemetryMetrics, + ) -> Self { + Self { + span, + duration: std::time::Duration::ZERO, + db_system_name, + operation_name, + metrics, + } + } + + fn add_duration(&mut self, duration: std::time::Duration) { + self.duration += duration; + } + + fn record_success(&self, returned_rows: i64) { + self.span + .record(otel::DB_RESPONSE_RETURNED_ROWS, returned_rows); + self.span.record(otel::OTEL_STATUS_CODE, "OK"); + let attributes = [ + opentelemetry::KeyValue::new(otel::DB_SYSTEM_NAME, self.db_system_name), + opentelemetry::KeyValue::new(otel::DB_OPERATION_NAME, self.operation_name.clone()), + opentelemetry::KeyValue::new(otel::OTEL_STATUS_CODE, "OK"), + ]; + self.metrics + .db_query_duration + .record(self.duration.as_secs_f64(), &attributes); + } + + fn record_error(&self, returned_rows: i64, error: &anyhow::Error) { + self.span + .record(otel::DB_RESPONSE_RETURNED_ROWS, returned_rows); + self.span.record(otel::OTEL_STATUS_CODE, "ERROR"); + self.span + .record(otel::EXCEPTION_MESSAGE, tracing::field::display(error)); + self.span + .record("exception.details", tracing::field::debug(error)); + let attributes = [ + opentelemetry::KeyValue::new(otel::DB_SYSTEM_NAME, self.db_system_name), + opentelemetry::KeyValue::new(otel::DB_OPERATION_NAME, self.operation_name.clone()), + opentelemetry::KeyValue::new(otel::OTEL_STATUS_CODE, "ERROR"), + opentelemetry::KeyValue::new(otel::ERROR_TYPE, error.to_string()), + ]; + self.metrics + .db_query_duration + .record(self.duration.as_secs_f64(), &attributes); + } } -fn record_db_query_error(span: &tracing::Span, returned_rows: i64, error: &anyhow::Error) { - span.record("db.response.returned_rows", returned_rows); - span.record("otel.status_code", "ERROR"); - span.record("exception.message", tracing::field::display(error)); - span.record("exception.details", tracing::field::debug(error)); +fn create_db_query_span( + sql: &str, + source_file: &Path, + line: usize, + db_system_name: &'static str, +) -> (tracing::Span, String) { + let operation_name = sql.split_whitespace().next().unwrap_or("").to_uppercase(); + let span = tracing::info_span!( + "db.query", + { otel::DB_QUERY_TEXT } = sql, + { otel::DB_SYSTEM_NAME } = db_system_name, + { otel::DB_OPERATION_NAME } = operation_name, + { otel::CODE_FILE_PATH } = %source_file.display(), + { otel::CODE_LINE_NUMBER } = source_line_number(line), + { otel::OTEL_STATUS_CODE } = tracing::field::Empty, + { otel::EXCEPTION_MESSAGE } = tracing::field::Empty, + "exception.details" = tracing::field::Empty, + { otel::DB_RESPONSE_RETURNED_ROWS } = tracing::field::Empty, + ); + (span, operation_name) } impl Database { @@ -93,22 +168,29 @@ pub fn stream_query_results_with_conn<'a>( request.server_timing.record("bind_params"); let connection = take_connection(&request.app_state.db, db_connection, request).await?; log::trace!("Executing query {:?}", query.sql); - let query_span = tracing::info_span!( - "db.query", - db.query.text = query.sql, - db.system.name = request.app_state.db.info.database_type.otel_name(), - code.file.path = %source_file.display(), - code.line.number = source_line_number(stmt.query_position.start.line), - otel.status_code = tracing::field::Empty, - exception.message = tracing::field::Empty, - exception.details = tracing::field::Empty, - db.response.returned_rows = tracing::field::Empty, + let db_system_name = request.app_state.db.info.database_type.otel_name(); + let (query_span, operation_name) = create_db_query_span( + query.sql, + source_file, + stmt.query_position.start.line, + db_system_name, ); - record_query_params(&query_span, &query.param_values); + let mut query_metrics = DbQueryMetricsContext::new( + query_span.clone(), + operation_name, + db_system_name, + &request.app_state.telemetry_metrics, + ); + record_query_params(&query_metrics.span, &query.param_values); let mut stream = connection.fetch_many(query); let mut error = None; let mut returned_rows: i64 = 0; - while let Some(elem) = stream.next().instrument(query_span.clone()).await { + loop { + let start_next = std::time::Instant::now(); + let next_elem = stream.next().instrument(query_span.clone()).await; + query_metrics.add_duration(start_next.elapsed()); + let Some(elem) = next_elem else { break; }; + let mut query_result = parse_single_sql_result(source_file, stmt, elem); if let DbItem::Error(e) = query_result { error = Some(e); @@ -131,11 +213,11 @@ pub fn stream_query_results_with_conn<'a>( } drop(stream); if let Some(error) = error { - record_db_query_error(&query_span, returned_rows, &error); + query_metrics.record_error(returned_rows, &error); try_rollback_transaction(connection).await; yield DbItem::Error(error); } else { - record_db_query_success(&query_span, returned_rows); + query_metrics.record_success(returned_rows); } }, ParsedStatement::SetVariable { variable, value} => { @@ -271,35 +353,41 @@ async fn execute_set_variable_query<'a>( query.sql ); - let query_span = tracing::info_span!( - "db.query", - db.query.text = query.sql, - db.system.name = request.app_state.db.info.database_type.otel_name(), - code.file.path = %source_file.display(), - code.line.number = source_line_number(statement.query_position.start.line), - otel.status_code = tracing::field::Empty, - exception.message = tracing::field::Empty, - exception.details = tracing::field::Empty, - db.response.returned_rows = tracing::field::Empty, + let db_system_name = request.app_state.db.info.database_type.otel_name(); + let (query_span, operation_name) = create_db_query_span( + query.sql, + source_file, + statement.query_position.start.line, + db_system_name, + ); + let mut query_metrics = DbQueryMetricsContext::new( + query_span.clone(), + operation_name, + db_system_name, + &request.app_state.telemetry_metrics, ); - record_query_params(&query_span, &query.param_values); + record_query_params(&query_metrics.span, &query.param_values); + let start_time = std::time::Instant::now(); let value = match connection .fetch_optional(query) .instrument(query_span.clone()) .await { Ok(Some(row)) => { - record_db_query_success(&query_span, 1_i64); + query_metrics.add_duration(start_time.elapsed()); + query_metrics.record_success(1_i64); row_to_string(&row) } Ok(None) => { - record_db_query_success(&query_span, 0_i64); + query_metrics.add_duration(start_time.elapsed()); + query_metrics.record_success(0_i64); None } Err(e) => { + query_metrics.add_duration(start_time.elapsed()); try_rollback_transaction(connection).await; let err = display_stmt_db_error(source_file, statement, e); - record_db_query_error(&query_span, 0_i64, &err); + query_metrics.record_error(0_i64, &err); return Err(err); } }; @@ -798,13 +886,16 @@ mod tests { exception.details = tracing::field::Empty, db.response.returned_rows = tracing::field::Empty, ); - record_db_query_success(&span, 3); + let metrics = crate::telemetry_metrics::TelemetryMetrics::default(); + let query_metrics = + DbQueryMetricsContext::new(span.clone(), "SELECT".to_string(), "sqlite", &metrics); + query_metrics.record_success(3); drop(span); }); - assert_eq!(fields["otel.status_code"], "OK"); - assert_eq!(fields["db.response.returned_rows"], "3"); - assert!(!fields.contains_key("exception.message")); + assert_eq!(fields[otel::OTEL_STATUS_CODE], "OK"); + assert_eq!(fields[otel::DB_RESPONSE_RETURNED_ROWS], "3"); + assert!(!fields.contains_key(otel::EXCEPTION_MESSAGE)); assert!(!fields.contains_key("exception.details")); } @@ -819,13 +910,16 @@ mod tests { db.response.returned_rows = tracing::field::Empty, ); let error = anyhow!("query failed").context("while executing SELECT 1"); - record_db_query_error(&span, 2, &error); + let metrics = crate::telemetry_metrics::TelemetryMetrics::default(); + let query_metrics = + DbQueryMetricsContext::new(span.clone(), "SELECT".to_string(), "sqlite", &metrics); + query_metrics.record_error(2, &error); drop(span); }); - assert_eq!(fields["otel.status_code"], "ERROR"); - assert_eq!(fields["db.response.returned_rows"], "2"); - assert!(fields["exception.message"].contains("while executing SELECT 1")); + assert_eq!(fields[otel::OTEL_STATUS_CODE], "ERROR"); + assert_eq!(fields[otel::DB_RESPONSE_RETURNED_ROWS], "2"); + assert!(fields[otel::EXCEPTION_MESSAGE].contains("while executing SELECT 1")); assert!(fields["exception.details"].contains("query failed")); } } diff --git a/src/webserver/database/mod.rs b/src/webserver/database/mod.rs index 950096db..2849c37a 100644 --- a/src/webserver/database/mod.rs +++ b/src/webserver/database/mod.rs @@ -63,7 +63,12 @@ impl SupportedDatabase { /// See #[must_use] pub fn otel_name(self) -> &'static str { - match self { + Self::otel_name_from_kind(self) + } + + #[must_use] + pub fn otel_name_from_kind(kind: impl Into) -> &'static str { + match kind.into() { Self::Sqlite => "sqlite", Self::Duckdb => "duckdb", Self::Oracle => "oracle.db", @@ -76,6 +81,18 @@ impl SupportedDatabase { } } +impl From for SupportedDatabase { + fn from(kind: AnyKind) -> Self { + match kind { + AnyKind::Postgres => Self::Postgres, + AnyKind::MySql => Self::MySql, + AnyKind::Sqlite => Self::Sqlite, + AnyKind::Mssql => Self::Mssql, + AnyKind::Odbc => Self::Generic, + } + } +} + pub struct Database { pub connection: sqlx::AnyPool, pub info: DbInfo, diff --git a/src/webserver/database/sqlpage_functions/functions.rs b/src/webserver/database/sqlpage_functions/functions.rs index c3dbd218..e756b0d9 100644 --- a/src/webserver/database/sqlpage_functions/functions.rs +++ b/src/webserver/database/sqlpage_functions/functions.rs @@ -13,6 +13,7 @@ use crate::webserver::{ use anyhow::{anyhow, Context}; use futures_util::StreamExt; use mime_guess::mime; +use opentelemetry_semantic_conventions::attribute as otel; use std::fmt::Write; use std::{borrow::Cow, ffi::OsStr, str::FromStr}; use tracing::Instrument; @@ -216,11 +217,11 @@ async fn fetch( let method = http_request.method.as_deref().unwrap_or("GET"); let fetch_span = tracing::info_span!( "http.client", - otel.name = format!("{method}"), - http.request.method = method, - url.full = %http_request.url, - http.request.body.size = tracing::field::Empty, - http.response.status_code = tracing::field::Empty, + "otel.name" = format!("{method}"), + { otel::HTTP_REQUEST_METHOD } = method, + { otel::URL_FULL } = %http_request.url, + { otel::HTTP_REQUEST_BODY_SIZE } = tracing::field::Empty, + { otel::HTTP_RESPONSE_STATUS_CODE } = tracing::field::Empty, ); async { @@ -232,7 +233,7 @@ async fn fetch( let mut response = if let Some(body) = &http_request.body { let (body, req) = prepare_request_body(body, req)?; tracing::Span::current().record( - "http.request.body.size", + otel::HTTP_REQUEST_BODY_SIZE, i64::try_from(body.len()).unwrap_or(i64::MAX), ); req.send_body(body) @@ -243,7 +244,7 @@ async fn fetch( .map_err(|e| anyhow!("Unable to fetch {}: {e}", http_request.url))?; tracing::Span::current().record( - "http.response.status_code", + otel::HTTP_RESPONSE_STATUS_CODE, i64::from(response.status().as_u16()), ); @@ -319,11 +320,11 @@ async fn fetch_with_meta( let method = http_request.method.as_deref().unwrap_or("GET"); let fetch_span = tracing::info_span!( "http.client", - otel.name = format!("{method}"), - http.request.method = method, - url.full = %http_request.url, - http.request.body.size = tracing::field::Empty, - http.response.status_code = tracing::field::Empty, + "otel.name" = format!("{method}"), + { otel::HTTP_REQUEST_METHOD } = method, + { otel::URL_FULL } = %http_request.url, + { otel::HTTP_REQUEST_BODY_SIZE } = tracing::field::Empty, + { otel::HTTP_RESPONSE_STATUS_CODE } = tracing::field::Empty, ); async { @@ -335,7 +336,7 @@ async fn fetch_with_meta( let response_result = if let Some(body) = &http_request.body { let (body, req) = prepare_request_body(body, req)?; tracing::Span::current().record( - "http.request.body.size", + otel::HTTP_REQUEST_BODY_SIZE, i64::try_from(body.len()).unwrap_or(i64::MAX), ); req.send_body(body).await @@ -350,7 +351,7 @@ async fn fetch_with_meta( Ok(mut response) => { let status = response.status(); tracing::Span::current() - .record("http.response.status_code", i64::from(status.as_u16())); + .record(otel::HTTP_RESPONSE_STATUS_CODE, i64::from(status.as_u16())); obj.serialize_entry("status", &status.as_u16())?; let mut has_error = false; if status.is_server_error() { diff --git a/src/webserver/http.rs b/src/webserver/http.rs index 9098b550..1d0e31e7 100644 --- a/src/webserver/http.rs +++ b/src/webserver/http.rs @@ -17,6 +17,7 @@ use actix_web::http::header::{ContentType, Header, HttpDate, IfModifiedSince, La use actix_web::http::{header, StatusCode}; use actix_web::web::PayloadConfig; use actix_web::{dev::ServiceResponse, middleware, web, App, Error, HttpResponse, HttpServer}; +use opentelemetry_semantic_conventions::attribute as otel; use tracing::{Instrument, Span}; use tracing_actix_web::{DefaultRootSpanBuilder, RootSpanBuilder, TracingLogger}; @@ -258,7 +259,7 @@ async fn render_sql( let exec_span = tracing::info_span!( "sqlpage.file", otel.name = %sql_execution_span_name(&source_path), - code.file.path = %source_path.display(), + { otel::CODE_FILE_PATH } = %source_path.display(), ); actix_web::rt::spawn(tracing::Instrument::instrument( async move { @@ -342,24 +343,23 @@ impl RootSpanBuilder for SqlPageRootSpanBuilder { let span = tracing::span!( tracing::Level::INFO, "HTTP request", - http.method = %http_method, - http.route = %http_route, - http.flavor = %tracing_actix_web::root_span_macro::private::http_flavor(request.version()), - http.scheme = %tracing_actix_web::root_span_macro::private::http_scheme(connection_info.scheme()), - http.host = %connection_info.host(), - http.client_ip = %request.connection_info().realip_remote_addr().unwrap_or(""), - http.user_agent = %user_agent, - http.target = %request - .uri() - .path_and_query() - .map_or("", actix_web::http::uri::PathAndQuery::as_str), - http.status_code = tracing::field::Empty, - otel.name = %otel_name, - otel.kind = "server", - otel.status_code = tracing::field::Empty, + { otel::HTTP_REQUEST_METHOD } = %http_method, + { otel::HTTP_ROUTE } = %http_route, + { otel::NETWORK_PROTOCOL_NAME } = "http", + { otel::NETWORK_PROTOCOL_VERSION } = %tracing_actix_web::root_span_macro::private::http_flavor(request.version()), + { otel::URL_SCHEME } = %tracing_actix_web::root_span_macro::private::http_scheme(connection_info.scheme()), + { otel::SERVER_ADDRESS } = %connection_info.host(), + { otel::CLIENT_ADDRESS } = %request.connection_info().realip_remote_addr().unwrap_or(""), + { otel::USER_AGENT_ORIGINAL } = %user_agent, + { otel::URL_PATH } = %request.path(), + { otel::URL_QUERY } = %request.query_string(), + { otel::HTTP_RESPONSE_STATUS_CODE } = tracing::field::Empty, + "otel.name" = %otel_name, + "otel.kind" = "server", + { otel::OTEL_STATUS_CODE } = tracing::field::Empty, request_id = %request_id, - exception.message = tracing::field::Empty, - exception.details = tracing::field::Empty, + { otel::EXCEPTION_MESSAGE } = tracing::field::Empty, + "exception.details" = tracing::field::Empty, ); std::mem::drop(connection_info); tracing_actix_web::root_span_macro::private::set_otel_parent(request, &span); @@ -392,7 +392,7 @@ async fn process_sql_request( let sql_file = { let span = tracing::info_span!( "sqlpage.file.load", - code.file.path = %sql_path.display(), + { otel::CODE_FILE_PATH } = %sql_path.display(), ); app_state .sql_file_cache @@ -553,6 +553,7 @@ pub fn create_app( // when receiving a request outside of the prefix, redirect to the prefix .default_service(fn_service(default_prefix_redirect)) .wrap(OidcMiddleware::new(&app_state)) + .wrap(super::http_metrics::HttpMetrics) .wrap(TracingLogger::::new()) .wrap(default_headers()) .wrap(middleware::Condition::new( diff --git a/src/webserver/http_metrics.rs b/src/webserver/http_metrics.rs new file mode 100644 index 00000000..ea8bb4b8 --- /dev/null +++ b/src/webserver/http_metrics.rs @@ -0,0 +1,90 @@ +use std::future::{ready, Ready}; +use std::time::Instant; + +use actix_web::{ + dev::{forward_ready, Service, ServiceRequest, ServiceResponse, Transform}, + web, Error, +}; +use futures_util::future::LocalBoxFuture; +use opentelemetry::KeyValue; +use opentelemetry_semantic_conventions::attribute as otel; +use tracing_actix_web::root_span_macro::private::{http_method_str, http_scheme}; + +use crate::AppState; + +pub struct HttpMetrics; + +impl Transform for HttpMetrics +where + S: Service, Error = Error> + 'static, + S::Future: 'static, +{ + type Response = ServiceResponse; + type Error = Error; + type Transform = HttpMetricsMiddleware; + type InitError = (); + type Future = Ready>; + + fn new_transform(&self, service: S) -> Self::Future { + ready(Ok(HttpMetricsMiddleware { service })) + } +} + +pub struct HttpMetricsMiddleware { + service: S, +} + +impl Service for HttpMetricsMiddleware +where + S: Service, Error = Error> + 'static, + S::Future: 'static, +{ + type Response = ServiceResponse; + type Error = Error; + type Future = LocalBoxFuture<'static, Result>; + + forward_ready!(service); + + fn call(&self, req: ServiceRequest) -> Self::Future { + let start_time = Instant::now(); + let method = http_method_str(req.method()).to_string(); + let connection_info = req.connection_info(); + let scheme = http_scheme(connection_info.scheme()).to_string(); + let host = connection_info.host().to_string(); + drop(connection_info); + + // We get the route pattern. In Actix, req.match_pattern() returns the matched route + let route = req + .match_pattern() + .unwrap_or_else(|| req.path().to_string()); + + let fut = self.service.call(req); + + Box::pin(async move { + let res = fut.await?; + let duration = start_time.elapsed().as_secs_f64(); + let status = res.status().as_u16(); + + let mut attributes = vec![ + KeyValue::new(otel::HTTP_REQUEST_METHOD, method), + KeyValue::new(otel::HTTP_RESPONSE_STATUS_CODE, status.to_string()), + KeyValue::new(otel::HTTP_ROUTE, route), + KeyValue::new(otel::URL_SCHEME, scheme), + KeyValue::new(otel::SERVER_ADDRESS, host), + ]; + + if status >= 500 { + attributes.push(KeyValue::new(otel::ERROR_TYPE, status.to_string())); + } + + if let Some(app_state) = res.request().app_data::>() { + app_state + .telemetry_metrics + .http_request_duration + .record(duration, &attributes); + } + + Ok(res) + }) + } +} diff --git a/src/webserver/mod.rs b/src/webserver/mod.rs index c640970c..a692616d 100644 --- a/src/webserver/mod.rs +++ b/src/webserver/mod.rs @@ -35,6 +35,7 @@ mod error; pub mod error_with_status; pub mod http; pub mod http_client; +pub mod http_metrics; pub mod http_request_info; mod https; pub mod request_variables; diff --git a/src/webserver/oidc.rs b/src/webserver/oidc.rs index f1ce269c..8e26ad43 100644 --- a/src/webserver/oidc.rs +++ b/src/webserver/oidc.rs @@ -34,6 +34,7 @@ use openidconnect::{ EmptyExtraTokenFields, IdTokenFields, IdTokenVerifier, StandardErrorResponse, StandardTokenResponse, }; +use opentelemetry_semantic_conventions::attribute as otel; use serde::{Deserialize, Serialize}; use tracing::Instrument; @@ -727,8 +728,8 @@ async fn exchange_code_for_token( ) -> anyhow::Result { let span = tracing::info_span!( "http.client", - otel.name = "POST token_endpoint", - http.request.method = "POST", + "otel.name" = "POST token_endpoint", + { otel::HTTP_REQUEST_METHOD } = "POST", ); let token_response = oidc_client .exchange_code(openidconnect::AuthorizationCode::new(