feat(controller/reporting): implement reporting initialization & otel & minor fixes

kuyaseg-controller/src/reporting/mod.rs

@@ -1,5 +1,25 @@
-use confique::Config;
+use std::sync::Arc;
+use std::time::Duration;
 
+use anyhow::Context;
+use confique::Config;
+use tokio::sync::broadcast::{
+    self,
+    Sender,
+};
+use tokio::task::JoinSet;
+use tokio_util::sync::CancellationToken;
+
+use crate::reporting::k8s_event::K8sEvent;
+use crate::reporting::otel::OtelObservation;
+use crate::reporting::prometheus::ProLabel;
+use crate::reporting::tracing_evt::{
+    TracingK8sEventLayer,
+    TracingK8sEventLayerConfig,
+};
+
+pub mod k8s_event;
+pub mod otel;
 pub mod prometheus;
 pub mod tracing_evt;
 
@@ -22,3 +42,174 @@ pub fn trace_id_str(trace_id: &opentelemetry::TraceId) -> Option<String> {
         x => Some(x.to_string()),
     }
 }
+
+/// You should await on handle somewhere
+pub struct ReportingInit {
+    pub tracing_layer: TracingK8sEventLayer,
+    pub handle: JoinSet<()>,
+}
+
+pub fn init(k8s_client: kube::Client) -> anyhow::Result<ReportingInit> {
+    let config = ReportingConfig::builder()
+        .env()
+        .load()
+        .context("Load Reporting Configuration")?;
+
+    let mut reporting_jobs = JoinSet::new();
+
+    let k8s_tx = config
+        .k8s_events
+        .then(|| init_k8s_events(k8s_client, &mut reporting_jobs));
+    let pro_tx = config.prometheus.then(|| init_pro_metrics(&mut reporting_jobs));
+    let otel_tx = config.opentelemetry.then(|| init_otel(&mut reporting_jobs));
+
+    Ok(ReportingInit {
+        tracing_layer: TracingK8sEventLayer::new(TracingK8sEventLayerConfig {
+            k8s_events: k8s_tx,
+            prometheus: pro_tx,
+            otel: otel_tx,
+        }),
+        handle: reporting_jobs,
+    })
+}
+
+fn init_k8s_events(client: kube::Client, joinset: &mut JoinSet<()>) -> Sender<Arc<K8sEvent>> {
+    use kube::runtime::events::{
+        Recorder,
+        Reporter,
+    };
+
+    use crate::reporting::k8s_event::K8sEventRecorder;
+
+    let (tx, mut rx) = broadcast::channel::<Arc<K8sEvent>>(128);
+    let reporter = Reporter {
+        controller: env!("CARGO_PKG_NAME").into(),
+        instance: std::env::var("CONTROLLER_POD_NAME").ok(),
+    };
+    let recorder = Recorder::new(client, reporter);
+    let recorder = K8sEventRecorder::new(recorder);
+
+    let rtx = tx.clone(); // for dumb recovery
+    joinset.spawn(async move {
+        let mut rec_limit = 4;
+        while rec_limit > 0 {
+            match rx.recv().await {
+                Ok(ev) => {
+                    recorder.record(&ev).await;
+                    rec_limit = 4;
+                }
+                Err(e) => {
+                    tracing::error!(no_event=true, debug=?e, "Failed to receive k8s event from tracing");
+                    rx = rtx.subscribe(); // dumb recover strategy
+                    tokio::time::sleep(Duration::from_secs(3)).await;
+                    rec_limit -= 1;
+                }
+            }
+        }
+        tracing::error!(
+            no_event = true,
+            "K8s Event reporting from tracing stopped from limit reached"
+        );
+    });
+
+    tx
+}
+
+fn init_pro_metrics(joinset: &mut JoinSet<()>) -> Sender<Arc<ProLabel>> {
+    use prometheus_client::registry::Registry;
+
+    use crate::reporting::prometheus::ProMetrics;
+
+    let (tx, mut rx) = broadcast::channel::<Arc<ProLabel>>(128);
+    let metrics = ProMetrics::new();
+
+    let registration_cancellation = CancellationToken::new();
+
+    let rtx = tx.clone();
+    let ctk = registration_cancellation.clone();
+    let promt = metrics.clone();
+    joinset.spawn(async move {
+        let mut rec_limit = 4;
+        while rec_limit > 0 {
+            match rx.recv().await {
+                Ok(met) => {
+                    promt.measure(&met);
+                }
+                Err(e) => {
+                    tracing::error!(no_pro=true, debug=?e, "Failed to receive prometheus measurement from tracing");
+                    rx = rtx.subscribe(); // dumb recover strategy;
+                    tokio::time::sleep(Duration::from_secs(3)).await;
+                    rec_limit -= 1;
+                }
+            }
+        }
+        ctk.cancel();
+        tracing::error!(
+            no_pro = true,
+            "Prometheus measurement from tracing stopped from limit reached"
+        );
+    });
+
+    joinset.spawn(async move {
+        let mut registry = Registry::default();
+        loop {
+            tokio::select! {
+                _ = tokio::time::sleep(Duration::from_secs(10)) => {
+                    metrics.register(&mut registry);
+                },
+                _ = registration_cancellation.cancelled() => {
+                    tracing::warn!(no_pro=true, "Stopped publishing prometheus metrics: job cancelled");
+                    break;
+                }
+            }
+        }
+    });
+
+    tx
+}
+
+fn init_otel(joinset: &mut JoinSet<()>) -> Sender<Arc<OtelObservation>> {
+    use crate::reporting::otel::OtelAggregator;
+
+    let exporter = opentelemetry_otlp::LogExporter::builder()
+        .with_tonic()
+        .build()
+        .expect("Failed to create OTLP log exporter");
+
+    let provider = opentelemetry_sdk::logs::SdkLoggerProvider::builder()
+        .with_batch_exporter(exporter)
+        .build();
+
+    let aggregator = OtelAggregator::new(provider);
+
+    let (tx, mut rx) = broadcast::channel::<Arc<OtelObservation>>(128);
+
+    let rtx = tx.clone();
+    joinset.spawn(async move {
+        let mut rec_limit = 4;
+        while rec_limit > 0 {
+            match rx.recv().await {
+                Ok(obs) => {
+                    aggregator.observe(&obs);
+                    rec_limit = 4;
+                }
+                Err(e) => {
+                    tracing::error!(
+                        no_otel = true,
+                        debug = ?e,
+                        "Failed to receive otel observation from tracing"
+                    );
+                    rx = rtx.subscribe();
+                    tokio::time::sleep(Duration::from_secs(3)).await;
+                    rec_limit -= 1;
+                }
+            }
+        }
+        tracing::error!(
+            no_otel = true,
+            "OpenTelemetry reporting from tracing stopped from limit reached"
+        );
+    });
+
+    tx
+}