From 39bdedc2436aed22b94be750b996bb1eaa18a497 Mon Sep 17 00:00:00 2001 From: Feng Pan Date: Tue, 20 Jan 2026 10:17:28 +0000 Subject: [PATCH] Add telemetry-rs for exporting otel data into mdm --- scripts/rs/Cargo.toml | 26 +++++++++++ scripts/rs/otel.toml | 9 ++++ scripts/rs/rs.sh | 1 + scripts/rs/src/config.rs | 61 +++++++++++++++++++++++++ scripts/rs/src/lib.rs | 76 +++++++++++++++++++++++++++++++ scripts/rs/src/main.rs | 96 ++++++++++++++++++++++++++++++++++++++++ 6 files changed, 269 insertions(+) create mode 100644 scripts/rs/Cargo.toml create mode 100644 scripts/rs/otel.toml create mode 100644 scripts/rs/rs.sh create mode 100644 scripts/rs/src/config.rs create mode 100644 scripts/rs/src/lib.rs create mode 100644 scripts/rs/src/main.rs diff --git a/scripts/rs/Cargo.toml b/scripts/rs/Cargo.toml new file mode 100644 index 00000000..a2b889ec --- /dev/null +++ b/scripts/rs/Cargo.toml @@ -0,0 +1,26 @@ +[package] +name = "telemetry_rs" +version = "1.0.0" +edition = "2021" +description = "Daemon to export its cpu usage via OTEL" +license = "MIT" + +[dependencies] +opentelemetry = { version = "0.21", features = ["metrics"] } +opentelemetry_sdk = { version = "0.21", features = ["metrics", "rt-tokio"] } +opentelemetry-otlp = { version = "0.14", features = ["tonic", "metrics"] } +sysinfo = "0.30" +tokio = { version = "1", features = ["full"] } + +# config + errors +anyhow = "1" +serde = { version = "1", features = ["derive"] } +toml = "0.8" + +[lib] +name = "otel_lib_rs" +path = "src/lib.rs" + +[[bin]] +name = "telemetry_rs" +path = "src/main.rs" diff --git a/scripts/rs/otel.toml b/scripts/rs/otel.toml new file mode 100644 index 00000000..9af6ee35 --- /dev/null +++ b/scripts/rs/otel.toml @@ -0,0 +1,9 @@ +[otel] +endpoint = "http://localhost:4317" +service_name = "rust-otel-service" +export_interval_secs = 10 +set_global = true + +[otel.resource] +env = "dev" +region = "mel" diff --git a/scripts/rs/rs.sh b/scripts/rs/rs.sh new file mode 100644 index 00000000..fc023f3e --- /dev/null +++ b/scripts/rs/rs.sh @@ -0,0 +1 @@ +OTEL_CONFIG_FILE=./otel.toml cargo run --bin telemetry_rs \ No newline at end of file diff --git a/scripts/rs/src/config.rs b/scripts/rs/src/config.rs new file mode 100644 index 00000000..d9a82e40 --- /dev/null +++ b/scripts/rs/src/config.rs @@ -0,0 +1,61 @@ +use serde::Deserialize; +use std::{collections::HashMap, fs, time::Duration}; + +#[derive(Debug, Deserialize)] +pub struct OtelFileConfig { + pub otel: OtelSection, +} + +#[derive(Debug, Deserialize)] +pub struct OtelSection { + pub endpoint: String, + pub service_name: String, + pub export_interval_secs: u64, + + #[serde(default = "default_true")] + pub set_global: bool, + + #[serde(default)] + pub resource: HashMap, +} + +fn default_true() -> bool { + true +} + +#[derive(Clone, Debug)] +pub struct OtelConfig { + pub endpoint: String, + pub service_name: String, + pub export_interval: Duration, + pub set_global: bool, + pub resource_kvs: Vec<(String, String)>, +} + +impl OtelConfig { + pub fn from_toml_file(path: &str) -> anyhow::Result { + let content = fs::read_to_string(path)?; + let file_cfg: OtelFileConfig = toml::from_str(&content)?; + + Ok(Self { + endpoint: file_cfg.otel.endpoint, + service_name: file_cfg.otel.service_name, + export_interval: Duration::from_secs(file_cfg.otel.export_interval_secs), + set_global: file_cfg.otel.set_global, + resource_kvs: file_cfg.otel.resource.into_iter().collect(), + }) + } + + pub fn validate(&self) -> anyhow::Result<()> { + if self.service_name.trim().is_empty() { + anyhow::bail!("otel.service_name cannot be empty"); + } + if !self.endpoint.starts_with("http://") && !self.endpoint.starts_with("https://") { + anyhow::bail!( + "otel.endpoint must start with http:// or https:// (got: {})", + self.endpoint + ); + } + Ok(()) + } +} diff --git a/scripts/rs/src/lib.rs b/scripts/rs/src/lib.rs new file mode 100644 index 00000000..9d3cf9df --- /dev/null +++ b/scripts/rs/src/lib.rs @@ -0,0 +1,76 @@ +mod config; +pub use config::OtelConfig; + +use opentelemetry::{global, KeyValue}; +use opentelemetry::metrics::{Meter, MeterProvider as _}; +use opentelemetry_otlp::WithExportConfig; + +use opentelemetry_sdk::{ + metrics::{ + reader::{AggregationSelector, TemporalitySelector}, + Aggregation, InstrumentKind, MeterProviderBuilder, PeriodicReader, + }, + runtime::Tokio, + Resource, +}; + +pub struct OtelRuntime { + meter: Meter, + // Keep provider alive (dropping it can stop exports) + _provider: opentelemetry_sdk::metrics::SdkMeterProvider, +} + +struct LastValueCumulative; + +impl AggregationSelector for LastValueCumulative { + fn aggregation(&self, _kind: InstrumentKind) -> Aggregation { + Aggregation::LastValue + } +} + +impl TemporalitySelector for LastValueCumulative { + fn temporality(&self, _kind: InstrumentKind) -> opentelemetry_sdk::metrics::data::Temporality { + opentelemetry_sdk::metrics::data::Temporality::Cumulative + } +} + +impl OtelRuntime { + pub fn init(cfg: OtelConfig) -> anyhow::Result { + cfg.validate()?; + + let exporter = opentelemetry_otlp::new_exporter() + .tonic() + .with_endpoint(cfg.endpoint) + .build_metrics_exporter(Box::new(LastValueCumulative), Box::new(LastValueCumulative)) + .map_err(|e| anyhow::anyhow!("Failed to build OTLP metrics exporter: {e}"))?; + + let reader = PeriodicReader::builder(exporter, Tokio) + .with_interval(cfg.export_interval) + .build(); + + let mut attrs = vec![KeyValue::new("service.name", cfg.service_name)]; + for (k, v) in cfg.resource_kvs { + attrs.push(KeyValue::new(k, v)); + } + + let provider = MeterProviderBuilder::default() + .with_resource(Resource::new(attrs)) + .with_reader(reader) + .build(); + + if cfg.set_global { + global::set_meter_provider(provider.clone()); + } + + let meter = provider.meter("telemetry_rs"); + + Ok(Self { + meter, + _provider: provider, + }) + } + + pub fn meter(&self) -> &Meter { + &self.meter + } +} diff --git a/scripts/rs/src/main.rs b/scripts/rs/src/main.rs new file mode 100644 index 00000000..127c1497 --- /dev/null +++ b/scripts/rs/src/main.rs @@ -0,0 +1,96 @@ +use opentelemetry::KeyValue; +use sysinfo::{CpuRefreshKind, ProcessRefreshKind, RefreshKind, System}; +use tokio::time::{sleep, Duration}; + +fn find_main_dockerd_pid(system: &System) -> Option { + // Prefer dockerd whose parent PID is 1 (your original heuristic) + for (pid, process) in system.processes() { + if process.name() == "dockerd" + && process.parent().map(|pp| pp.as_u32()).unwrap_or(0) == 1 + { + return Some(pid.as_u32()); + } + } + // Fallback: first dockerd found + for (pid, process) in system.processes() { + if process.name() == "dockerd" { + return Some(pid.as_u32()); + } + } + None +} + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + // Config path: env override + default + let cfg_path = + std::env::var("OTEL_CONFIG_FILE").unwrap_or_else(|_| "otel.toml".to_string()); + + let cfg = otel_lib_rs::OtelConfig::from_toml_file(&cfg_path)?; + let otel = otel_lib_rs::OtelRuntime::init(cfg)?; + + let meter = otel.meter(); + + let gauge = meter + .f64_observable_gauge("dockerd.cpu.usage.percent") + .with_description("Total CPU usage of dockerd process and its children (%)") + .init(); + + meter + .register_callback(&[gauge.as_any()], move |observer| { + let mut system = System::new_with_specifics( + RefreshKind::new() + .with_processes(ProcessRefreshKind::everything()) + .with_cpu(CpuRefreshKind::everything()), + ); + + // sysinfo CPU% is based on two samples + system.refresh_processes(); + system.refresh_cpu(); + + std::thread::sleep(std::time::Duration::from_millis(500)); + + system.refresh_processes(); + system.refresh_cpu(); + + let mut total_cpu = 0.0; + + if let Some(main_pid) = find_main_dockerd_pid(&system) { + for (pid, process) in system.processes() { + if process.name() == "dockerd" { + let p = pid.as_u32(); + let parent = process.parent().map(|pp| pp.as_u32()).unwrap_or(0); + if p == main_pid || parent == main_pid { + total_cpu += process.cpu_usage() as f64; + } + } + } + + observer.observe_f64( + &gauge, + total_cpu, + &[ + KeyValue::new("process", "dockerd"), + KeyValue::new("scope", "main+children"), + ], + ); + } else { + observer.observe_f64( + &gauge, + 0.0, + &[ + KeyValue::new("process", "dockerd"), + KeyValue::new("status", "not_found"), + ], + ); + } + }) + .expect("Failed to register callback"); + + println!("Loaded OTEL config from: {}", cfg_path); + println!("Exporting dockerd CPU usage..."); + + loop { + sleep(Duration::from_secs(600)).await; + } +}