Add telemetry-rs for exporting otel data into mdm

Author: FengPan-Frank

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"

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"

scripts/rs/rs.sh

@@ -0,0 +1 @@
+OTEL_CONFIG_FILE=./otel.toml cargo run --bin telemetry_rs

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<String, String>,
+}
+
+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<Self> {
+        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(())
+    }
+}

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<Self> {
+        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
+    }
+}

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<u32> {
+    // 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;
+    }
+}