diff --git a/kuyaseg-controller/src/reporting/mod.rs b/kuyaseg-controller/src/reporting/mod.rs index 5c234dc..240a9c4 100644 --- a/kuyaseg-controller/src/reporting/mod.rs +++ b/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 { +pub fn init(k8s_client: kube::Client, ct: CancellationToken) -> anyhow::Result { let config = ReportingConfig::builder() .env() .load() @@ -59,9 +60,14 @@ pub fn init(k8s_client: kube::Client) -> anyhow::Result { 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)); + .then(|| init_k8s_events(k8s_client, &mut reporting_jobs, ct.clone())); + let pro_tx = config + .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 { }) } -fn init_k8s_events(client: kube::Client, joinset: &mut JoinSet<()>) -> Sender> { +fn init_k8s_events(client: kube::Client, joinset: &mut JoinSet<()>, ct: CancellationToken) -> Sender> { use kube::runtime::events::{ Recorder, Reporter, @@ -81,7 +87,10 @@ fn init_k8s_events(client: kube::Client, joinset: &mut JoinSet<()>) -> Sender>(128); + const WARN_THRESHOLD: usize = 10; + const MAX_CAPACITY: usize = 100; + + let (tx, mut rx) = broadcast::channel::>(512); let reporter = Reporter { controller: env!("CARGO_PKG_NAME").into(), instance: std::env::var("CONTROLLER_POD_NAME").ok(), @@ -91,36 +100,65 @@ fn init_k8s_events(client: kube::Client, joinset: &mut JoinSet<()>) -> Sender> = 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 { - 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; + tokio::select! { + 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) => { + 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" + "K8s Event reporting from tracing stopped" ); }); tx } -fn init_pro_metrics(joinset: &mut JoinSet<()>) -> Sender> { +fn init_pro_metrics(joinset: &mut JoinSet<()>, ct: CancellationToken) -> Sender> { use prometheus_client::registry::Registry; use crate::reporting::prometheus::ProMetrics; - let (tx, mut rx) = broadcast::channel::>(128); + let (tx, mut rx) = broadcast::channel::>(512); let metrics = ProMetrics::new(); let registration_cancellation = CancellationToken::new(); @@ -128,28 +166,37 @@ fn init_pro_metrics(joinset: &mut JoinSet<()>) -> Sender> { 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 { - 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; + 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"); + 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" + "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> { 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,48 +219,52 @@ fn init_pro_metrics(joinset: &mut JoinSet<()>) -> Sender> { tx } -fn init_otel(joinset: &mut JoinSet<()>) -> Sender> { +fn init_otel(joinset: &mut JoinSet<()>, ct: CancellationToken) -> anyhow::Result>> { 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::>(128); + let (tx, mut rx) = broadcast::channel::>(512); 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; + tokio::select! { + biased; + _ = ct.cancelled() => break, + msg = rx.recv() => { + match msg { + Ok(obs) => { + task_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" - ); + tracing::error!(no_otel = true, "OpenTelemetry reporting from tracing stopped"); }); - tx + Ok(tx) } diff --git a/kuyaseg-controller/src/reporting/otel.rs b/kuyaseg-controller/src/reporting/otel.rs index e95841e..f850d4c 100644 --- a/kuyaseg-controller/src/reporting/otel.rs +++ b/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(); + } +} diff --git a/kuyaseg-controller/src/reporting/tracing_evt.rs b/kuyaseg-controller/src/reporting/tracing_evt.rs index 71ee571..0e1c8fa 100644 --- a/kuyaseg-controller/src/reporting/tracing_evt.rs +++ b/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. -/// Channel consumer must batch/dedup to avoid API errors. +/// - k8s events are rate-limited to 10 events/min pub struct TracingK8sEventLayer { - k8s_tx: RwLock>>>, - pro_tx: RwLock>>>, - otel_tx: RwLock>>>, + k8s_tx: Option>>, + pro_tx: Option>>, + otel_tx: Option>>, } impl TracingK8sEventLayer { pub fn new(config: TracingK8sEventLayerConfig) -> Self { Self { - k8s_tx: RwLock::new(config.k8s_events), - pro_tx: RwLock::new(config.prometheus), - otel_tx: RwLock::new(config.otel), + k8s_tx: config.k8s_events, + pro_tx: config.prometheus, + otel_tx: config.otel, } } - - pub fn reload_k8s(&self, tx: Option>>) { - *self.k8s_tx.write() = tx; - } - - pub fn reload_pro(&self, tx: Option>>) { - *self.pro_tx.write() = tx; - } - - pub fn reload_otel(&self, tx: Option>>) { - *self.otel_tx.write() = tx; - } } impl Layer 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 {