fix(controller/reporting): rate limit k8s events & add external cancellation

kuyaseg-controller/src/reporting/mod.rs

@@ -1,3 +1,4 @@
+use std::collections::VecDeque;
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -49,7 +50,7 @@ pub struct ReportingInit {
     pub handle: JoinSet<()>,
 }
 
-pub fn init(k8s_client: kube::Client) -> anyhow::Result<ReportingInit> {
+pub fn init(k8s_client: kube::Client, ct: CancellationToken) -> anyhow::Result<ReportingInit> {
     let config = ReportingConfig::builder()
         .env()
         .load()
@@ -59,9 +60,14 @@ pub fn init(k8s_client: kube::Client) -> anyhow::Result<ReportingInit> {
 
     let k8s_tx = config
         .k8s_events
-        .then(|| init_k8s_events(k8s_client, &mut reporting_jobs));
+        .then(|| init_k8s_events(k8s_client, &mut reporting_jobs, ct.clone()));
-    let pro_tx = config.prometheus.then(|| init_pro_metrics(&mut reporting_jobs));
+    let pro_tx = config
-    let otel_tx = config.opentelemetry.then(|| init_otel(&mut reporting_jobs));
+        .prometheus
+        .then(|| init_pro_metrics(&mut reporting_jobs, ct.clone()));
+    let otel_tx = config
+        .opentelemetry
+        .then(|| init_otel(&mut reporting_jobs, ct.clone()))
+        .transpose()?;
 
     Ok(ReportingInit {
         tracing_layer: TracingK8sEventLayer::new(TracingK8sEventLayerConfig {
@@ -73,7 +79,7 @@ pub fn init(k8s_client: kube::Client) -> anyhow::Result<ReportingInit> {
     })
 }
 
-fn init_k8s_events(client: kube::Client, joinset: &mut JoinSet<()>) -> Sender<Arc<K8sEvent>> {
+fn init_k8s_events(client: kube::Client, joinset: &mut JoinSet<()>, ct: CancellationToken) -> Sender<Arc<K8sEvent>> {
     use kube::runtime::events::{
         Recorder,
         Reporter,
@@ -81,7 +87,10 @@ fn init_k8s_events(client: kube::Client, joinset: &mut JoinSet<()>) -> Sender<Ar
 
     use crate::reporting::k8s_event::K8sEventRecorder;
 
-    let (tx, mut rx) = broadcast::channel::<Arc<K8sEvent>>(128);
+    const WARN_THRESHOLD: usize = 10;
+    const MAX_CAPACITY: usize = 100;
+
+    let (tx, mut rx) = broadcast::channel::<Arc<K8sEvent>>(512);
     let reporter = Reporter {
         controller: env!("CARGO_PKG_NAME").into(),
         instance: std::env::var("CONTROLLER_POD_NAME").ok(),
@@ -91,11 +100,38 @@ fn init_k8s_events(client: kube::Client, joinset: &mut JoinSet<()>) -> Sender<Ar
 
     let rtx = tx.clone(); // for dumb recovery
     joinset.spawn(async move {
+        let mut queue: VecDeque<Arc<K8sEvent>> = VecDeque::new();
+        let mut warned = false;
+        let mut drain_tick = tokio::time::interval(Duration::from_secs(6)); // 10 event/min
         let mut rec_limit = 4;
+
         while rec_limit > 0 {
-            match rx.recv().await {
+            tokio::select! {
-                Ok(ev) => {
+                biased;
+                _ = ct.cancelled() => break,
+                _ = drain_tick.tick() => {
+                    if let Some(ev) = queue.pop_front() {
                         recorder.record(&ev).await;
+                    }
+                },
+                msg = rx.recv() => {
+                    match msg {
+                        Ok(ev) => {
+                            if queue.is_empty() {
+                                recorder.record(&ev).await;
+                                warned = false;
+                            } else {
+                                if queue.len() >= MAX_CAPACITY {
+                                    queue.pop_front();
+                                }
+                                queue.push_back(ev);
+                                if queue.len() > WARN_THRESHOLD && !warned {
+                                    tracing::warn!(no_event=true, queue_len=queue.len(), "K8s event queue exceeded threshold");
+                                    warned = true;
+                                } else if queue.len() <= WARN_THRESHOLD {
+                                    warned = false;
+                                }
+                            }
                             rec_limit = 4;
                         }
                         Err(e) => {
@@ -106,21 +142,23 @@ fn init_k8s_events(client: kube::Client, joinset: &mut JoinSet<()>) -> Sender<Ar
                         }
                     }
                 }
+            }
+        }
         tracing::error!(
             no_event = true,
-            "K8s Event reporting from tracing stopped from limit reached"
+            "K8s Event reporting from tracing stopped"
         );
     });
 
     tx
 }
 
-fn init_pro_metrics(joinset: &mut JoinSet<()>) -> Sender<Arc<ProLabel>> {
+fn init_pro_metrics(joinset: &mut JoinSet<()>, ct: CancellationToken) -> Sender<Arc<ProLabel>> {
     use prometheus_client::registry::Registry;
 
     use crate::reporting::prometheus::ProMetrics;
 
-    let (tx, mut rx) = broadcast::channel::<Arc<ProLabel>>(128);
+    let (tx, mut rx) = broadcast::channel::<Arc<ProLabel>>(512);
     let metrics = ProMetrics::new();
 
     let registration_cancellation = CancellationToken::new();
@@ -128,12 +166,18 @@ fn init_pro_metrics(joinset: &mut JoinSet<()>) -> Sender<Arc<ProLabel>> {
     let rtx = tx.clone();
     let ctk = registration_cancellation.clone();
     let promt = metrics.clone();
+    let consumer_ct = ct.clone();
     joinset.spawn(async move {
         let mut rec_limit = 4;
         while rec_limit > 0 {
-            match rx.recv().await {
+            tokio::select! {
+                biased;
+                _ = consumer_ct.cancelled() => break,
+                msg = rx.recv() => {
+                    match msg {
                         Ok(met) => {
                             promt.measure(&met);
+                            rec_limit = 4;
                         }
                         Err(e) => {
                             tracing::error!(no_pro=true, debug=?e, "Failed to receive prometheus measurement from tracing");
@@ -143,13 +187,16 @@ fn init_pro_metrics(joinset: &mut JoinSet<()>) -> Sender<Arc<ProLabel>> {
                         }
                     }
                 }
+            }
+        }
         ctk.cancel();
         tracing::error!(
             no_pro = true,
-            "Prometheus measurement from tracing stopped from limit reached"
+            "Prometheus measurement from tracing stopped"
         );
     });
 
+    let ctk = ct.clone();
     joinset.spawn(async move {
         let mut registry = Registry::default();
         loop {
@@ -158,7 +205,11 @@ fn init_pro_metrics(joinset: &mut JoinSet<()>) -> Sender<Arc<ProLabel>> {
                     metrics.register(&mut registry);
                 },
                 _ = registration_cancellation.cancelled() => {
-                    tracing::warn!(no_pro=true, "Stopped publishing prometheus metrics: job cancelled");
+                    tracing::warn!(no_pro=true, "Stopped publishing prometheus metrics: consumer job cancelled");
+                    break;
+                }
+                _ = ctk.cancelled() => {
+                    tracing::warn!(no_pro=true, "Stopped publishing prometheus metrics: shutdown");
                     break;
                 }
             }
@@ -168,29 +219,34 @@ fn init_pro_metrics(joinset: &mut JoinSet<()>) -> Sender<Arc<ProLabel>> {
     tx
 }
 
-fn init_otel(joinset: &mut JoinSet<()>) -> Sender<Arc<OtelObservation>> {
+fn init_otel(joinset: &mut JoinSet<()>, ct: CancellationToken) -> anyhow::Result<Sender<Arc<OtelObservation>>> {
     use crate::reporting::otel::OtelAggregator;
 
     let exporter = opentelemetry_otlp::LogExporter::builder()
         .with_tonic()
         .build()
-        .expect("Failed to create OTLP log exporter");
+        .context("Failed to build OTLP log exporter")?;
 
     let provider = opentelemetry_sdk::logs::SdkLoggerProvider::builder()
         .with_batch_exporter(exporter)
         .build();
 
-    let aggregator = OtelAggregator::new(provider);
+    let aggregator = Arc::new(OtelAggregator::new(provider));
+    let task_aggregator = aggregator.clone();
 
-    let (tx, mut rx) = broadcast::channel::<Arc<OtelObservation>>(128);
+    let (tx, mut rx) = broadcast::channel::<Arc<OtelObservation>>(512);
 
     let rtx = tx.clone();
     joinset.spawn(async move {
         let mut rec_limit = 4;
         while rec_limit > 0 {
-            match rx.recv().await {
+            tokio::select! {
+                biased;
+                _ = ct.cancelled() => break,
+                msg = rx.recv() => {
+                    match msg {
                         Ok(obs) => {
-                    aggregator.observe(&obs);
+                            task_aggregator.observe(&obs);
                             rec_limit = 4;
                         }
                         Err(e) => {
@@ -205,11 +261,10 @@ fn init_otel(joinset: &mut JoinSet<()>) -> Sender<Arc<OtelObservation>> {
                         }
                     }
                 }
-        tracing::error!(
+            }
-            no_otel = true,
+        }
-            "OpenTelemetry reporting from tracing stopped from limit reached"
+        tracing::error!(no_otel = true, "OpenTelemetry reporting from tracing stopped");
-        );
     });
 
-    tx
+    Ok(tx)
 }

kuyaseg-controller/src/reporting/otel.rs

@@ -98,3 +98,9 @@ impl OtelAggregator {
         todo!("Metrics are not implemented and should not be used for now")
     }
 }
+
+impl Drop for OtelAggregator {
+    fn drop(&mut self) {
+        self.force_flush();
+    }
+}

kuyaseg-controller/src/reporting/tracing_evt.rs

@@ -1,5 +1,4 @@
 use std::sync::Arc;
-use std::sync::nonpoison::RwLock;
 
 use k8s_openapi::api::core::v1::ObjectReference;
 use kube::runtime::events::{
@@ -46,34 +45,21 @@ pub struct TracingK8sEventLayerConfig {
 ///
 /// - k8s native events you must record yourself - thats what `tx` is for.
 /// - prometheus metrics are written automatically (if metrics set), but labels are also send with `tx`.
-/// - k8s events are NOT rate-limited. K8s API rejects >10 events/object/min.
+/// - k8s events are rate-limited to 10 events/min
-///   Channel consumer must batch/dedup to avoid API errors.
 pub struct TracingK8sEventLayer {
-    k8s_tx: RwLock<Option<Sender<Arc<K8sEvent>>>>,
+    k8s_tx: Option<Sender<Arc<K8sEvent>>>,
-    pro_tx: RwLock<Option<Sender<Arc<ProLabel>>>>,
+    pro_tx: Option<Sender<Arc<ProLabel>>>,
-    otel_tx: RwLock<Option<Sender<Arc<OtelObservation>>>>,
+    otel_tx: Option<Sender<Arc<OtelObservation>>>,
 }
 
 impl TracingK8sEventLayer {
     pub fn new(config: TracingK8sEventLayerConfig) -> Self {
         Self {
-            k8s_tx: RwLock::new(config.k8s_events),
+            k8s_tx: config.k8s_events,
-            pro_tx: RwLock::new(config.prometheus),
+            pro_tx: config.prometheus,
-            otel_tx: RwLock::new(config.otel),
+            otel_tx: config.otel,
         }
     }
-
-    pub fn reload_k8s(&self, tx: Option<Sender<Arc<K8sEvent>>>) {
-        *self.k8s_tx.write() = tx;
-    }
-
-    pub fn reload_pro(&self, tx: Option<Sender<Arc<ProLabel>>>) {
-        *self.pro_tx.write() = tx;
-    }
-
-    pub fn reload_otel(&self, tx: Option<Sender<Arc<OtelObservation>>>) {
-        *self.otel_tx.write() = tx;
-    }
 }
 
 impl<S> Layer<S> for TracingK8sEventLayer
@@ -152,7 +138,7 @@ where
         if event_fields.no_event.is_none_or(|v| !v)
             && span_fields.no_event.is_none_or(|v| !v)
             && let Some(event) = event_fields.try_to_event(k8s_event_type)
-            && let Some(tx) = self.k8s_tx.read().clone()
+            && let Some(tx) = self.k8s_tx.as_ref()
         {
             _ = tx
                 .send(Arc::new(K8sEvent {
@@ -178,7 +164,7 @@ where
         if event_fields.no_pro.is_none_or(|v| !v)
             && span_fields.no_pro.is_none_or(|v| !v)
             && let Some(label) = try_create_pro_label(&span_fields, &event_fields, pro_severity, str_trace_id.clone())
-            && let Some(tx) = self.pro_tx.read().clone()
+            && let Some(tx) = self.pro_tx.as_ref()
         {
             _ = tx.send(Arc::new(label)).inspect_err(
                 |e| tracing::error!(debug=?e, "Failed to issue event on prometheus label measure from tracing"),
@@ -200,7 +186,7 @@ where
         };
         if event_fields.no_otel.is_none_or(|v| !v)
             && span_fields.no_otel.is_none_or(|v| !v)
-            && let Some(tx) = self.otel_tx.read().clone()
+            && let Some(tx) = self.otel_tx.as_ref()
         {
             _ = tx
                 .send(Arc::new(OtelObservation::Log {