diff --git a/k8s/deployment-grid-rs.yaml b/k8s/deployment-grid-rs.yaml index 48cdc415..7822b3ee 100644 --- a/k8s/deployment-grid-rs.yaml +++ b/k8s/deployment-grid-rs.yaml @@ -99,6 +99,12 @@ spec: value: "3" - name: METRICS_ADDR value: "0.0.0.0:9100" + # OTLP/gRPC endpoint for the cluster OTel Collector. Unset + # disables span export entirely (fmt layer still logs to + # stdout). See vntx-infra/observability/ for the collector + # + Tempo backend. + - name: OTEL_EXPORTER_OTLP_ENDPOINT + value: "http://otel-collector.observability.svc.cluster.local:4317" envFrom: - secretRef: name: prop-secrets diff --git a/k8s/deployment-hrrr-point-rs.yaml b/k8s/deployment-hrrr-point-rs.yaml index 65cd73da..379265bb 100644 --- a/k8s/deployment-hrrr-point-rs.yaml +++ b/k8s/deployment-hrrr-point-rs.yaml @@ -59,6 +59,8 @@ spec: value: "info" - name: PROP_GRID_RS_PG_CONNS value: "4" + - name: OTEL_EXPORTER_OTLP_ENDPOINT + value: "http://otel-collector.observability.svc.cluster.local:4317" envFrom: - secretRef: name: prop-secrets diff --git a/rust/prop_grid_rs/Cargo.lock b/rust/prop_grid_rs/Cargo.lock index 71fc0918..812587c7 100644 --- a/rust/prop_grid_rs/Cargo.lock +++ b/rust/prop_grid_rs/Cargo.lock @@ -622,6 +622,25 @@ dependencies = [ "wasip3", ] +[[package]] +name = "h2" +version = "0.4.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2f44da3a8150a6703ed5d34e164b875fd14c2cdab9af1252a9a1020bde2bdc54" +dependencies = [ + "atomic-waker", + "bytes", + "fnv", + "futures-core", + "futures-sink", + "http", + "indexmap", + "slab", + "tokio", + "tokio-util", + "tracing", +] + [[package]] name = "hashbrown" version = "0.15.5" @@ -748,6 +767,7 @@ dependencies = [ "bytes", "futures-channel", "futures-core", + "h2", "http", "http-body", "httparse", @@ -775,6 +795,19 @@ dependencies = [ "webpki-roots 1.0.7", ] +[[package]] +name = "hyper-timeout" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2b90d566bffbce6a75bd8b09a05aa8c2cb1fabb6cb348f8840c9e4c90a0d83b0" +dependencies = [ + "hyper", + "hyper-util", + "pin-project-lite", + "tokio", + "tower-service", +] + [[package]] name = "hyper-util" version = "0.1.20" @@ -959,6 +992,15 @@ dependencies = [ "serde", ] +[[package]] +name = "itertools" +version = "0.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2b192c782037fadd9cfa75548310488aabdbf3d2da73885b31bd0abd03351285" +dependencies = [ + "either", +] + [[package]] name = "itoa" version = "1.0.18" @@ -1188,6 +1230,66 @@ version = "1.21.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" +[[package]] +name = "opentelemetry" +version = "0.31.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b84bcd6ae87133e903af7ef497404dda70c60d0ea14895fc8a5e6722754fc2a0" +dependencies = [ + "futures-core", + "futures-sink", + "js-sys", + "pin-project-lite", + "thiserror 2.0.18", + "tracing", +] + +[[package]] +name = "opentelemetry-otlp" +version = "0.31.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1f69cd6acbb9af919df949cd1ec9e5e7fdc2ef15d234b6b795aaa525cc02f71f" +dependencies = [ + "http", + "opentelemetry", + "opentelemetry-proto", + "opentelemetry_sdk", + "prost", + "thiserror 2.0.18", + "tokio", + "tonic", +] + +[[package]] +name = "opentelemetry-proto" +version = "0.31.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7175df06de5eaee9909d4805a3d07e28bb752c34cab57fa9cff549da596b30f" +dependencies = [ + "opentelemetry", + "opentelemetry_sdk", + "prost", + "tonic", + "tonic-prost", +] + +[[package]] +name = "opentelemetry_sdk" +version = "0.31.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e14ae4f5991976fd48df6d843de219ca6d31b01daaab2dad5af2badeded372bd" +dependencies = [ + "futures-channel", + "futures-executor", + "futures-util", + "opentelemetry", + "percent-encoding", + "rand 0.9.4", + "thiserror 2.0.18", + "tokio", + "tokio-stream", +] + [[package]] name = "parking" version = "2.2.1" @@ -1232,6 +1334,26 @@ version = "2.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220" +[[package]] +name = "pin-project" +version = "1.1.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f1749c7ed4bcaf4c3d0a3efc28538844fb29bcdd7d2b67b2be7e20ba861ff517" +dependencies = [ + "pin-project-internal", +] + +[[package]] +name = "pin-project-internal" +version = "1.1.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d9b20ed30f105399776b9c883e68e536ef602a16ae6f596d2c473591d6ad64c6" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "pin-project-lite" version = "0.2.17" @@ -1358,6 +1480,9 @@ dependencies = [ "flate2", "futures", "num_cpus", + "opentelemetry", + "opentelemetry-otlp", + "opentelemetry_sdk", "png", "pretty_assertions", "prometheus", @@ -1372,10 +1497,34 @@ dependencies = [ "thiserror 2.0.18", "tokio", "tracing", + "tracing-opentelemetry", "tracing-subscriber", "uuid", ] +[[package]] +name = "prost" +version = "0.14.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d2ea70524a2f82d518bce41317d0fae74151505651af45faf1ffbd6fd33f0568" +dependencies = [ + "bytes", + "prost-derive", +] + +[[package]] +name = "prost-derive" +version = "0.14.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "27c6023962132f4b30eb4c172c91ce92d933da334c59c23cddee82358ddafb0b" +dependencies = [ + "anyhow", + "itertools", + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "quinn" version = "0.11.9" @@ -2322,6 +2471,43 @@ dependencies = [ "tokio", ] +[[package]] +name = "tonic" +version = "0.14.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fec7c61a0695dc1887c1b53952990f3ad2e3a31453e1f49f10e75424943a93ec" +dependencies = [ + "async-trait", + "base64", + "bytes", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-timeout", + "hyper-util", + "percent-encoding", + "pin-project", + "sync_wrapper", + "tokio", + "tokio-stream", + "tower", + "tower-layer", + "tower-service", + "tracing", +] + +[[package]] +name = "tonic-prost" +version = "0.14.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a55376a0bbaa4975a3f10d009ad763d8f4108f067c7c2e74f3001fb49778d309" +dependencies = [ + "bytes", + "prost", + "tonic", +] + [[package]] name = "tower" version = "0.5.3" @@ -2330,11 +2516,15 @@ checksum = "ebe5ef63511595f1344e2d5cfa636d973292adc0eec1f0ad45fae9f0851ab1d4" dependencies = [ "futures-core", "futures-util", + "indexmap", "pin-project-lite", + "slab", "sync_wrapper", "tokio", + "tokio-util", "tower-layer", "tower-service", + "tracing", ] [[package]] @@ -2411,6 +2601,22 @@ dependencies = [ "tracing-core", ] +[[package]] +name = "tracing-opentelemetry" +version = "0.32.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ac28f2d093c6c477eaa76b23525478f38de514fa9aeb1285738d4b97a9552fc" +dependencies = [ + "js-sys", + "opentelemetry", + "smallvec", + "tracing", + "tracing-core", + "tracing-log", + "tracing-subscriber", + "web-time", +] + [[package]] name = "tracing-serde" version = "0.2.0" diff --git a/rust/prop_grid_rs/Cargo.toml b/rust/prop_grid_rs/Cargo.toml index 692479cb..8fb0229d 100644 --- a/rust/prop_grid_rs/Cargo.toml +++ b/rust/prop_grid_rs/Cargo.toml @@ -25,6 +25,10 @@ anyhow = "1" thiserror = "2" tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] } +tracing-opentelemetry = "0.32" +opentelemetry = "0.31" +opentelemetry_sdk = { version = "0.31", features = ["rt-tokio"] } +opentelemetry-otlp = { version = "0.31", default-features = false, features = ["grpc-tonic", "trace"] } chrono = { version = "0.4", features = ["serde"] } uuid = { version = "1", features = ["v4", "serde"] } serde = { version = "1", features = ["derive"] } diff --git a/rust/prop_grid_rs/src/bin/hrrr_point_worker.rs b/rust/prop_grid_rs/src/bin/hrrr_point_worker.rs index 2dba04ae..b9fd213c 100644 --- a/rust/prop_grid_rs/src/bin/hrrr_point_worker.rs +++ b/rust/prop_grid_rs/src/bin/hrrr_point_worker.rs @@ -9,18 +9,14 @@ use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use std::time::{Duration, Instant}; -use prop_grid_rs::{db, fetcher::HrrrClient, hrrr_points}; +use prop_grid_rs::{db, fetcher::HrrrClient, hrrr_points, telemetry}; use tracing::{error, info, warn}; -use tracing_subscriber::EnvFilter; const IDLE_SLEEP: Duration = Duration::from_secs(10); #[tokio::main] async fn main() -> Result<(), Box> { - tracing_subscriber::fmt() - .json() - .with_env_filter(EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"))) - .init(); + let _telemetry = telemetry::init("hrrr-point-worker"); let database_url = std::env::var("DATABASE_URL") .map_err(|_| "DATABASE_URL environment variable is required")?; diff --git a/rust/prop_grid_rs/src/bin/worker.rs b/rust/prop_grid_rs/src/bin/worker.rs index edfa40f5..792b58cc 100644 --- a/rust/prop_grid_rs/src/bin/worker.rs +++ b/rust/prop_grid_rs/src/bin/worker.rs @@ -16,10 +16,9 @@ use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use std::time::{Duration, Instant}; -use prop_grid_rs::{db, fetcher::HrrrClient, metrics, pipeline}; +use prop_grid_rs::{db, fetcher::HrrrClient, metrics, pipeline, telemetry}; use tokio::task::JoinSet; use tracing::{error, info, warn}; -use tracing_subscriber::EnvFilter; const IDLE_SLEEP: Duration = Duration::from_secs(5); @@ -30,10 +29,7 @@ enum ChainOutcome { #[tokio::main] async fn main() -> Result<(), Box> { - tracing_subscriber::fmt() - .json() - .with_env_filter(EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"))) - .init(); + let _telemetry = telemetry::init("prop-grid-rs"); let database_url = std::env::var("DATABASE_URL") .map_err(|_| "DATABASE_URL environment variable is required")?; diff --git a/rust/prop_grid_rs/src/hrrr_points.rs b/rust/prop_grid_rs/src/hrrr_points.rs index 46126a74..cb762674 100644 --- a/rust/prop_grid_rs/src/hrrr_points.rs +++ b/rust/prop_grid_rs/src/hrrr_points.rs @@ -37,6 +37,14 @@ pub struct PointFetchStats { } /// Fetch + decode + persist for one `hrrr_fetch_tasks` batch. +#[tracing::instrument( + name = "hrrr_points.process_batch", + skip(client, pool, _tmp_dir), + fields( + valid_time = %valid_time, + points = points.len(), + ) +)] pub async fn process_batch( client: &HrrrClient, pool: &PgPool, diff --git a/rust/prop_grid_rs/src/lib.rs b/rust/prop_grid_rs/src/lib.rs index 49178ddd..54ee763e 100644 --- a/rust/prop_grid_rs/src/lib.rs +++ b/rust/prop_grid_rs/src/lib.rs @@ -35,3 +35,4 @@ pub mod region; pub mod scores_file; pub mod scorer; pub mod sounding_params; +pub mod telemetry; diff --git a/rust/prop_grid_rs/src/pipeline.rs b/rust/prop_grid_rs/src/pipeline.rs index 50a7aa90..e50a331f 100644 --- a/rust/prop_grid_rs/src/pipeline.rs +++ b/rust/prop_grid_rs/src/pipeline.rs @@ -64,6 +64,14 @@ pub struct ChainStepStats { pub band_count: u32, } +#[tracing::instrument( + name = "pipeline.run_chain_step", + skip(client, scores_dir), + fields( + run_time = %step.run_time, + forecast_hour = step.forecast_hour, + ) +)] pub async fn run_chain_step( client: &HrrrClient, scores_dir: &Path, @@ -368,6 +376,11 @@ pub struct AnalysisStepStats { pub commercial_cells_boosted: u32, } +#[tracing::instrument( + name = "pipeline.run_analysis_step", + skip(client, pool, scores_dir), + fields(run_time = %step.run_time) +)] pub async fn run_analysis_step( client: &HrrrClient, pool: &sqlx::PgPool, diff --git a/rust/prop_grid_rs/src/telemetry.rs b/rust/prop_grid_rs/src/telemetry.rs new file mode 100644 index 00000000..eb2204a3 --- /dev/null +++ b/rust/prop_grid_rs/src/telemetry.rs @@ -0,0 +1,108 @@ +//! Tracing init with optional OTLP export. +//! +//! Both bin targets (`worker`, `hrrr_point_worker`) call `init(svc)` at +//! startup. The local fmt layer (JSON to stdout) is always on so kubectl +//! logs keeps working. When `OTEL_EXPORTER_OTLP_ENDPOINT` is set we +//! additionally install a `tracing-opentelemetry` layer that ships +//! spans over OTLP/gRPC to the cluster's OTel Collector — see +//! `k8s/deployment-grid-rs.yaml` + `deployment-hrrr-point-rs.yaml` +//! for the env wiring. +//! +//! Returning the `TracerProvider` from `init` is deliberate: the caller +//! must hold it until shutdown and call `.shutdown()` so queued spans +//! flush before the process exits. Dropping it mid-run cancels the +//! background exporter and loses recent spans. +//! +//! Service identity is set via OpenTelemetry resource attributes +//! (`service.name`, `service.namespace`) so Tempo groups spans per +//! binary. `service.namespace=prop` matches the k8s namespace. + +use opentelemetry::{KeyValue, global, trace::TracerProvider as _}; +use opentelemetry_otlp::WithExportConfig; +use opentelemetry_sdk::{Resource, trace::SdkTracerProvider}; +use tracing_subscriber::{EnvFilter, layer::SubscriberExt, util::SubscriberInitExt}; + +/// Opaque handle the caller must hold until process shutdown; dropping +/// it stops the OTLP exporter. Returns `None` when OTLP is disabled +/// (no endpoint env var) so callers don't need to care which mode is +/// active. +pub struct TelemetryGuard { + provider: Option, +} + +impl TelemetryGuard { + /// Flush queued spans and shut down the exporter. Safe to call + /// multiple times; subsequent calls are no-ops. + pub fn shutdown(mut self) { + if let Some(provider) = self.provider.take() { + let _ = provider.shutdown(); + } + } +} + +impl Drop for TelemetryGuard { + fn drop(&mut self) { + if let Some(provider) = self.provider.take() { + let _ = provider.shutdown(); + } + } +} + +/// Initialize tracing for a binary. `service_name` becomes the +/// `service.name` attribute in exported spans and the log target +/// prefix. +pub fn init(service_name: &'static str) -> TelemetryGuard { + let env_filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")); + + let fmt_layer = tracing_subscriber::fmt::layer().json(); + + // If the collector endpoint isn't set, skip the OTLP layer entirely + // — useful for local cargo runs where there's no collector to ship + // to. The fmt layer alone keeps dev parity with prod log output. + let otlp_endpoint = std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").ok(); + + match otlp_endpoint { + Some(endpoint) if !endpoint.is_empty() => { + let exporter = opentelemetry_otlp::SpanExporter::builder() + .with_tonic() + .with_endpoint(endpoint) + .build() + .expect("build OTLP span exporter"); + + let resource = Resource::builder() + .with_attributes([ + KeyValue::new("service.name", service_name), + KeyValue::new("service.namespace", "prop"), + ]) + .build(); + + let provider = SdkTracerProvider::builder() + .with_batch_exporter(exporter) + .with_resource(resource) + .build(); + + let tracer = provider.tracer(service_name); + global::set_tracer_provider(provider.clone()); + + let otel_layer = tracing_opentelemetry::layer().with_tracer(tracer); + + tracing_subscriber::registry() + .with(env_filter) + .with(fmt_layer) + .with(otel_layer) + .init(); + + TelemetryGuard { + provider: Some(provider), + } + } + _ => { + tracing_subscriber::registry() + .with(env_filter) + .with(fmt_layer) + .init(); + + TelemetryGuard { provider: None } + } + } +}