Merge pull request #423 from dora-rs/metrics-refactoring

Metrics refactoring

apis/python/node/Cargo.toml

@@ -20,7 +20,7 @@ pyo3 = { workspace = true, features = ["eyre", "abi3-py37"] }
 eyre = "0.6"
 serde_yaml = "0.8.23"
 flume = "0.10.14"
-dora-runtime = { workspace = true, features = ["tracing", "python"] }
+dora-runtime = { workspace = true, features = ["tracing", "metrics", "python"] }
 arrow = { workspace = true, features = ["pyarrow"] }
 pythonize = { workspace = true }
 futures = "0.3.28"

binaries/runtime/Cargo.toml

@@ -15,8 +15,6 @@ dora-operator-api-types = { workspace = true }
 dora-core = { workspace = true }
 dora-tracing = { workspace = true, optional = true }
 dora-metrics = { workspace = true, optional = true }
-opentelemetry = { version = "0.21.0", features = ["metrics"], optional = true }
-opentelemetry-system-metrics = { version = "0.1.6", optional = true }
 eyre = "0.6.8"
 futures = "0.3.21"
 futures-concurrency = "7.1.0"
@@ -37,6 +35,6 @@ aligned-vec = "0.5.0"
 [features]
 default = ["tracing", "metrics"]
 tracing = ["dora-tracing"]
-telemetry = ["tracing", "opentelemetry", "tracing-opentelemetry"]
-metrics = ["opentelemetry", "opentelemetry-system-metrics", "dora-metrics"]
+telemetry = ["tracing", "tracing-opentelemetry"]
+metrics = ["dora-metrics"]
 python = ["pyo3", "dora-operator-api-python", "pythonize", "arrow/pyarrow"]

binaries/runtime/src/lib.rs

@@ -5,6 +5,7 @@ use dora_core::{
     daemon_messages::{NodeConfig, RuntimeConfig},
     descriptor::OperatorConfig,
 };
+use dora_metrics::init_meter_provider;
 use dora_node_api::{DoraNode, Event};
 use eyre::{bail, Context, Result};
 use futures::{Stream, StreamExt};
@@ -14,7 +15,6 @@ use operator::{run_operator, OperatorEvent, StopReason};
 #[cfg(feature = "tracing")]
 use dora_tracing::set_up_tracing;
 use std::{
-    borrow::Cow,
     collections::{BTreeMap, BTreeSet, HashMap},
     mem,
 };
@@ -123,19 +123,7 @@ async fn run(
     init_done: oneshot::Receiver<Result<()>>,
 ) -> eyre::Result<()> {
     #[cfg(feature = "metrics")]
-    let _started = {
-        use dora_metrics::init_metrics;
-        use opentelemetry::global;
-        use opentelemetry_system_metrics::init_process_observer;
-
-        let _started = init_metrics().context("Could not create opentelemetry meter")?;
-        let meter = global::meter(Cow::Borrowed(Box::leak(
-            config.node_id.to_string().into_boxed_str(),
-        )));
-        init_process_observer(meter).context("could not initiale system metrics observer")?;
-        _started
-    };
-
+    let _meter_provider = init_meter_provider(config.node_id.to_string());
     init_done
         .await
         .wrap_err("the `init_done` channel was closed unexpectedly")?

libraries/extensions/telemetry/metrics/Cargo.toml

@@ -11,4 +11,6 @@ license.workspace = true
 [dependencies]
 opentelemetry = { version = "0.21", features = ["metrics"] }
 opentelemetry-otlp = { version = "0.14.0", features = ["tonic", "metrics"] }
-opentelemetry_sdk = { version = "0.21.0", features = ["rt-tokio", "metrics"] }
+opentelemetry_sdk = { version = "0.21", features = ["rt-tokio", "metrics"] }
+eyre = "0.6.12"
+opentelemetry-system-metrics = { version = "0.1.6" }

libraries/extensions/telemetry/metrics/src/lib.rs

@@ -10,10 +10,13 @@
 //! [`sysinfo`]: https://github.com/GuillaumeGomez/sysinfo
 //! [`opentelemetry-rust`]: https://github.com/open-telemetry/opentelemetry-rust
 
-use opentelemetry::metrics::{self};
-use opentelemetry_sdk::{metrics::MeterProvider, runtime};
+use std::time::Duration;
 
+use eyre::{Context, Result};
+use opentelemetry::metrics::{self, MeterProvider as _};
 use opentelemetry_otlp::{ExportConfig, WithExportConfig};
+use opentelemetry_sdk::{metrics::MeterProvider, runtime};
+use opentelemetry_system_metrics::init_process_observer;
 /// Init opentelemetry meter
 ///
 /// Use the default Opentelemetry exporter with default config
@@ -34,5 +37,13 @@ pub fn init_metrics() -> metrics::Result<MeterProvider> {
                 .tonic()
                 .with_export_config(export_config),
         )
+        .with_period(Duration::from_secs(10))
         .build()
 }
+
+pub fn init_meter_provider(meter_id: String) -> Result<MeterProvider> {
+    let meter_provider = init_metrics().context("Could not create opentelemetry meter")?;
+    let meter = meter_provider.meter(meter_id);
+    let _ = init_process_observer(meter).context("could not initiale system metrics observer")?;
+    Ok(meter_provider)
+}