Implement tracing and metrics for Boxer (#43)

* Implement tracing and metrics for Boxer

* Fix formatting issues

Author: s-vitaliy

Cargo.lock

@@ -35,12 +35,10 @@ base64 = "0.22.1"
 futures-util = "0.3.31"
 
 # Open Telemetry dependencies
-opentelemetry = "0.24.0"
+opentelemetry = "0.30.0"
 opentelemetry-appender-log = "0.30.0"
 opentelemetry-otlp = { version = "0.30.0", features = ["grpc-tonic", "trace", "metrics", "logs"] }
-opentelemetry-resource-detectors = "0.3.0"
 opentelemetry_sdk = { version = "0.30.0", features = ["rt-tokio"] }
-opentelemetry-stdout = "0.30.0"
 opentelemetry-appender-tracing = "0.30.1"
 
 [dev-dependencies]

Cargo.toml

@@ -28,3 +28,15 @@ where
 
     fn stop(&self) -> anyhow::Result<()>;
 }
+
+#[async_trait]
+pub trait KubernetesResourceWatcherRunner<H, R>: Sized
+where
+    R: Resource<Scope = NamespaceResourceScope> + Clone + Debug + Serialize + DeserializeOwned + Send + Sync,
+    R::DynamicType: Hash + Eq + Clone + Default,
+    H: ResourceUpdateHandler<R> + Send + Sync + 'static,
+{
+    async fn start(&mut self, config: KubernetesResourceManagerConfig) -> anyhow::Result<()>;
+
+    fn stop(&self) -> anyhow::Result<()>;
+}

src/services/backends/kubernetes/kubernetes_resource_watcher.rs

@@ -1 +1,3 @@
 pub mod logging;
+pub mod metrics;
+pub mod tracing;

src/services/observability/open_telemetry.rs

@@ -0,0 +1,27 @@
+use anyhow::Context;
+use opentelemetry::global;
+use opentelemetry::metrics::Counter;
+
+pub fn init_metrics() -> anyhow::Result<()> {
+    let exporter = opentelemetry_otlp::MetricExporter::builder()
+        .with_tonic()
+        .with_temporality(opentelemetry_sdk::metrics::Temporality::Delta)
+        .build()
+        .with_context(|| "creating metric exporter")?;
+
+    let meter_provider = opentelemetry_sdk::metrics::SdkMeterProvider::builder()
+        .with_reader(opentelemetry_sdk::metrics::PeriodicReader::builder(exporter).build())
+        .build();
+
+    global::set_meter_provider(meter_provider);
+    Ok(())
+}
+
+pub fn token_succeeded(app_name: &'static str) -> Counter<u64> {
+    let meter = global::meter(app_name);
+    meter
+        .u64_counter(format!("{}.{}", app_name, "token_succeeded"))
+        .with_description("Count of successfully processed tokens")
+        .with_unit("tokens")
+        .build()
+}

src/services/observability/open_telemetry/metrics.rs

@@ -0,0 +1,56 @@
+pub mod tracing_facade;
+
+use opentelemetry::trace::{Status, TraceContextExt, Tracer};
+use opentelemetry::{Context, global};
+use opentelemetry_sdk::propagation::TraceContextPropagator;
+use std::fmt::Display;
+
+/// Initialize OpenTelemetry tracing with OTLP exporter
+/// Should be called once at the start of the application
+pub fn init_tracer() -> anyhow::Result<()> {
+    global::set_text_map_propagator(TraceContextPropagator::new());
+
+    let exporter = opentelemetry_otlp::SpanExporter::builder().with_tonic().build()?;
+
+    let tracer_provider = opentelemetry_sdk::trace::SdkTracerProvider::builder()
+        .with_batch_exporter(exporter)
+        .build();
+
+    global::set_tracer_provider(tracer_provider);
+    Ok(())
+}
+
+/// Start a new trace span with the given name
+/// Returns a Context containing the new span
+/// The caller is responsible for ending the span
+pub fn start_trace(span_name: &str) -> Context {
+    let tracer = global::tracer("");
+    let span = tracer
+        .span_builder(span_name.to_string())
+        .with_kind(opentelemetry::trace::SpanKind::Internal)
+        .start(&tracer);
+    Context::current().with_span(span)
+}
+
+/// Extension trait for Result to stop tracing and set span status
+pub trait ErrorExt<T, E> {
+    fn stop_trace(self, ctx: Context) -> Self;
+}
+
+/// Implementation of ErrorExt for Result
+impl<T, E> ErrorExt<T, E> for Result<T, E>
+where
+    E: Display,
+{
+    fn stop_trace(self, ctx: Context) -> Self {
+        if let Err(err) = &self {
+            ctx.span().set_status(Status::Error {
+                description: err.to_string().into(),
+            });
+        } else {
+            ctx.span().set_status(Status::Ok);
+        }
+        ctx.span().end();
+        self
+    }
+}

src/services/observability/open_telemetry/tracing.rs

@@ -0,0 +1,59 @@
+use crate::services::base::upsert_repository::ReadOnlyRepository;
+use crate::services::observability::open_telemetry::tracing::{ErrorExt, start_trace};
+use async_trait::async_trait;
+use opentelemetry::context::FutureExt;
+use std::marker::PhantomData;
+use std::sync::Arc;
+
+pub trait WithTracingFacade<Key, Value> {
+    fn with_tracing(
+        self: Arc<Self>,
+        span_name: String,
+    ) -> Arc<dyn ReadOnlyRepository<Key, Value, ReadError = anyhow::Error>>
+    where
+        Self: Sized + Send + Sync + 'static;
+}
+
+impl<Repo, Key, Value> WithTracingFacade<Key, Value> for Repo
+where
+    Repo: ReadOnlyRepository<Key, Value, ReadError = anyhow::Error> + Send + Sync + 'static,
+    Key: Send + Sync + 'static,
+    Value: Send + Sync + 'static,
+{
+    fn with_tracing(
+        self: Arc<Self>,
+        span_name: String,
+    ) -> Arc<dyn ReadOnlyRepository<Key, Value, ReadError = anyhow::Error>> {
+        Arc::new(TracingFacade {
+            span_name,
+            underlying: self,
+            _p: Default::default(),
+        })
+    }
+}
+
+struct TracingFacade<Repo, Key, Value>
+where
+    Repo: ReadOnlyRepository<Key, Value, ReadError = anyhow::Error>,
+    Key: Send + Sync + 'static,
+    Value: Send + Sync + 'static,
+{
+    span_name: String,
+    underlying: Arc<Repo>,
+    _p: PhantomData<(Key, Value)>,
+}
+
+#[async_trait]
+impl<Repo, Key, Value> ReadOnlyRepository<Key, Value> for TracingFacade<Repo, Key, Value>
+where
+    Repo: ReadOnlyRepository<Key, Value, ReadError = anyhow::Error>,
+    Key: Send + Sync + 'static,
+    Value: Send + Sync + 'static,
+{
+    type ReadError = anyhow::Error;
+
+    async fn get(&self, key: Key) -> Result<Value, Self::ReadError> {
+        let cx = start_trace(&self.span_name);
+        self.underlying.get(key).with_context(cx.clone()).await.stop_trace(cx)
+    }
+}