feat(rust): OTLP trace export to cluster OTel Collector
New telemetry module wires tracing-subscriber to both a local JSON fmt layer (keeps kubectl-logs output identical) and an OpenTelemetry OTLP/gRPC exporter, activated only when OTEL_EXPORTER_OTLP_ENDPOINT is set. The returned TelemetryGuard holds the SdkTracerProvider until process shutdown so queued spans flush before exit. Both bin targets (worker, hrrr_point_worker) now call telemetry::init(service_name) at startup; service_name becomes the OTel service.name attribute so Tempo groups spans per binary. Tracing instrumentation on the three main work units: - pipeline::run_chain_step (forecast f01..f18) - pipeline::run_analysis_step (analysis f00) - hrrr_points::process_batch (per-QSO point drain) k8s manifests set OTEL_EXPORTER_OTLP_ENDPOINT to the cluster collector at otel-collector.observability.svc.cluster.local:4317. Backend wiring lives in the vntx-infra repo.
This commit is contained in:
parent
a52d2e2b93
commit
65a1b1edb3
10 changed files with 352 additions and 12 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
206
rust/prop_grid_rs/Cargo.lock
generated
206
rust/prop_grid_rs/Cargo.lock
generated
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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"] }
|
||||
|
|
|
|||
|
|
@ -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<dyn std::error::Error>> {
|
||||
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")?;
|
||||
|
|
|
|||
|
|
@ -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<dyn std::error::Error>> {
|
||||
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")?;
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -35,3 +35,4 @@ pub mod region;
|
|||
pub mod scores_file;
|
||||
pub mod scorer;
|
||||
pub mod sounding_params;
|
||||
pub mod telemetry;
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
108
rust/prop_grid_rs/src/telemetry.rs
Normal file
108
rust/prop_grid_rs/src/telemetry.rs
Normal file
|
|
@ -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<SdkTracerProvider>,
|
||||
}
|
||||
|
||||
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 }
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
Add table
Reference in a new issue