From 8d407ef7b509679a5babc67bed734890889cb7b3 Mon Sep 17 00:00:00 2001 From: spushk Date: Sat, 16 May 2026 04:22:06 +0500 Subject: [PATCH] feat(controller/reporting): implement reporting initialization & otel & minor fixes --- kuyaseg-controller/src/reporting/k8s_event.rs | 36 +++ kuyaseg-controller/src/reporting/mod.rs | 193 ++++++++++++++- kuyaseg-controller/src/reporting/otel.rs | 100 ++++++++ .../src/reporting/prometheus.rs | 10 + .../src/reporting/tracing_evt.rs | 222 ++++++++++++------ 5 files changed, 488 insertions(+), 73 deletions(-) create mode 100644 kuyaseg-controller/src/reporting/k8s_event.rs create mode 100644 kuyaseg-controller/src/reporting/otel.rs diff --git a/kuyaseg-controller/src/reporting/k8s_event.rs b/kuyaseg-controller/src/reporting/k8s_event.rs new file mode 100644 index 0000000..5bd1294 --- /dev/null +++ b/kuyaseg-controller/src/reporting/k8s_event.rs @@ -0,0 +1,36 @@ +use std::fmt::Debug; + +use k8s_openapi::api::core::v1::ObjectReference; +use kube::runtime::events::{ + Event, + Recorder, +}; + +pub struct K8sEvent { + pub object: ObjectReference, + pub event: Event, +} + +pub struct K8sEventRecorder { + recorder: Recorder, +} + +impl K8sEventRecorder { + pub fn new(recorder: Recorder) -> Self { + Self { recorder } + } + + pub async fn record(&self, event: &K8sEvent) { + _ = self + .recorder + .publish(&event.event, &event.object) + .await + .inspect_err(|e| tracing::error!(no_event=true, error=?e, "Failed to record K8s event")); + } +} + +impl Debug for K8sEvent { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("K8sEvent").finish() + } +} diff --git a/kuyaseg-controller/src/reporting/mod.rs b/kuyaseg-controller/src/reporting/mod.rs index b853132..5c234dc 100644 --- a/kuyaseg-controller/src/reporting/mod.rs +++ b/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 { 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 { + 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> { + use kube::runtime::events::{ + Recorder, + Reporter, + }; + + use crate::reporting::k8s_event::K8sEventRecorder; + + let (tx, mut rx) = broadcast::channel::>(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> { + use prometheus_client::registry::Registry; + + use crate::reporting::prometheus::ProMetrics; + + let (tx, mut rx) = broadcast::channel::>(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> { + 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::>(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 +} diff --git a/kuyaseg-controller/src/reporting/otel.rs b/kuyaseg-controller/src/reporting/otel.rs new file mode 100644 index 0000000..e95841e --- /dev/null +++ b/kuyaseg-controller/src/reporting/otel.rs @@ -0,0 +1,100 @@ +use std::fmt::Debug; + +use opentelemetry::logs::{ + AnyValue, + LogRecord, + Logger, + LoggerProvider, + Severity, +}; +use opentelemetry_sdk::logs::SdkLoggerProvider; + +pub struct OtelLogFields { + pub severity: Severity, + pub message: String, + pub reason: String, + pub action: String, + pub k8s_name: String, + pub k8s_kind: String, + pub k8s_ns: Option, + pub k8s_api: Option, + pub k8s_uid: Option, + pub trace_id: Option, +} + +pub enum OtelObservation { + Log { fields: OtelLogFields }, + Metric {}, +} + +impl Debug for OtelObservation { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + let variant = match self { + Self::Log { .. } => "Log", + Self::Metric { .. } => "Metric", + }; + write!(f, "OtelObservation::{}", variant) + } +} + +pub struct OtelAggregator { + provider: SdkLoggerProvider, + logger: ::Logger, +} + +impl OtelAggregator { + pub fn new(provider: SdkLoggerProvider) -> Self { + let logger = provider.logger("kuyaseg-controller"); + Self { provider, logger } + } + + pub fn observe(&self, obs: &OtelObservation) { + match obs { + OtelObservation::Log { fields } => self.observe_log(fields), + OtelObservation::Metric {} => { /*Not implemented */ } + } + } + + pub fn force_flush(&self) { + if let Err(e) = self.provider.force_flush() { + tracing::warn!(error = ?e, "otel export force_flush failed"); + } + } + + fn observe_log(&self, fields: &OtelLogFields) { + let record = { + let mut record = self.logger.create_log_record(); + record.set_severity_number(fields.severity); + record.set_severity_text(fields.severity.name()); + record.set_body(AnyValue::from(fields.message.to_string())); + record.set_target("kuyaseg-controller"); + + record.add_attribute("reason", fields.reason.to_string()); + record.add_attribute("action", fields.action.to_string()); + record.add_attribute("k8s.resource.name", fields.k8s_name.to_string()); + record.add_attribute("k8s.resource.kind", fields.k8s_kind.to_string()); + + if let Some(ns) = fields.k8s_ns.as_ref() { + record.add_attribute("k8s.namespace.name", ns.to_string()); + } + if let Some(api) = fields.k8s_api.as_ref() { + record.add_attribute("k8s.resource.api_version", api.to_string()); + } + if let Some(uid) = fields.k8s_uid.as_ref() { + record.add_attribute("k8s.resource.uid", uid.to_string()); + } + if let Some(trace_id) = fields.trace_id.as_ref() { + record.add_attribute("trace_id", trace_id.to_string()); + } + + record + }; + + self.logger.emit(record); + } + + #[deprecated(note = "Not implemented")] + fn observe_metric(&self, _metric: ()) { + todo!("Metrics are not implemented and should not be used for now") + } +} diff --git a/kuyaseg-controller/src/reporting/prometheus.rs b/kuyaseg-controller/src/reporting/prometheus.rs index af013d0..4d0d7ba 100644 --- a/kuyaseg-controller/src/reporting/prometheus.rs +++ b/kuyaseg-controller/src/reporting/prometheus.rs @@ -1,3 +1,4 @@ +use std::fmt::Debug; use std::time::{ Instant, SystemTime, @@ -113,6 +114,15 @@ pub enum ProLabel { }, } +impl Debug for ProLabel { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + let variant = match self { + Self::Generic { .. } => "Generic", + }; + write!(f, "ProLabel::{}", variant) + } +} + /// General purpose label #[derive(Debug, Clone, Hash, PartialEq, Eq, EncodeLabelSet)] pub struct ProK8sLabel { diff --git a/kuyaseg-controller/src/reporting/tracing_evt.rs b/kuyaseg-controller/src/reporting/tracing_evt.rs index 12acb60..71ee571 100644 --- a/kuyaseg-controller/src/reporting/tracing_evt.rs +++ b/kuyaseg-controller/src/reporting/tracing_evt.rs @@ -1,3 +1,4 @@ +use std::sync::Arc; use std::sync::nonpoison::RwLock; use k8s_openapi::api::core::v1::ObjectReference; @@ -5,7 +6,8 @@ use kube::runtime::events::{ Event, EventType, }; -use tokio::sync::mpsc; +use opentelemetry::logs::Severity; +use tokio::sync::broadcast::Sender; use tracing::field::Visit; use tracing::span::Attributes; use tracing::{ @@ -18,30 +20,26 @@ use tracing_subscriber::Layer; use tracing_subscriber::layer::Context; use tracing_subscriber::registry::LookupSpan; +use crate::reporting::k8s_event::K8sEvent; +use crate::reporting::otel::{ + OtelLogFields, + OtelObservation, +}; use crate::reporting::prometheus::{ ProK8sExemplar, ProK8sLabel, ProK8sSeverity, ProLabel, - ProMetrics, }; use crate::reporting::trace_id_str; -/// - k8s events are not automatically recorded -/// - prometheus labels are recorded if metrics are set for layer -pub enum TracingK8sEvent { - K8sEvent { - reference: ObjectReference, - event: Box, - }, - ProLabel { - label: ProLabel, - }, -} - pub struct TracingK8sEventLayerConfig { - pub enable_k8s_events: bool, - pub enable_prometheus_metrics: bool, + /// Set to enable k8s events from tracing + pub k8s_events: Option>>, + /// set to enable prometheus metrics from tracing + pub prometheus: Option>>, + /// set to enable otel observations from tracing + pub otel: Option>>, } /// Watches for tracing events, containing information necessary to construst k8s events and write prometheus metrics. @@ -51,30 +49,30 @@ pub struct TracingK8sEventLayerConfig { /// - k8s events are NOT rate-limited. K8s API rejects >10 events/object/min. /// Channel consumer must batch/dedup to avoid API errors. pub struct TracingK8sEventLayer { - config: TracingK8sEventLayerConfig, - tx: RwLock>, - metrics: RwLock>, + k8s_tx: RwLock>>>, + pro_tx: RwLock>>>, + otel_tx: RwLock>>>, } impl TracingK8sEventLayer { - pub fn new( - config: TracingK8sEventLayerConfig, - tx: mpsc::UnboundedSender, - metrics: Option, - ) -> Self { + pub fn new(config: TracingK8sEventLayerConfig) -> Self { Self { - config, - tx: RwLock::new(tx), - metrics: RwLock::new(metrics), + k8s_tx: RwLock::new(config.k8s_events), + pro_tx: RwLock::new(config.prometheus), + otel_tx: RwLock::new(config.otel), } } - pub fn reload_tx(&self, tx: mpsc::UnboundedSender) { - *self.tx.write() = tx; + pub fn reload_k8s(&self, tx: Option>>) { + *self.k8s_tx.write() = tx; } - pub fn reload_metrics(&self, metrics: Option) { - *self.metrics.write() = metrics; + pub fn reload_pro(&self, tx: Option>>) { + *self.pro_tx.write() = tx; + } + + pub fn reload_otel(&self, tx: Option>>) { + *self.otel_tx.write() = tx; } } @@ -101,11 +99,29 @@ where } fn on_event(&self, event: &TracingEvent<'_>, ctx: Context<'_, S>) { + // Level guard: do not process DEBUG, TRACE + if *event.metadata().level() > tracing::Level::INFO { + return; + } + let scope = match ctx.event_scope(event) { Some(s) => s, - None => return, + None => { + /* In this layer we've got nothing to do without data */ + return; + } }; + // construct & check pro fields + let event_fields = { + let mut fields = EventFields::default(); + event.record(&mut fields); + fields + }; + if !event_fields.check_fields() { + return; + } + // construct & check event fields let span_fields = { let mut fields = SpanFields::default(); @@ -120,67 +136,100 @@ where return; } - // construct & check pro fields - let event_fields = { - let mut fields = EventFields::default(); - event.record(&mut fields); - fields - }; - if !event_fields.check_fields() { - return; - } - - // calculate event types - let (k8s_event_type, pro_severity) = match *event.metadata().level() { - tracing::Level::ERROR => ( - EventType::Warning, - if event_fields.critical.is_some_and(|v| v) { - ProK8sSeverity::critical - } else { - ProK8sSeverity::error - }, - ), - tracing::Level::WARN => (EventType::Warning, ProK8sSeverity::warning), - tracing::Level::INFO => (EventType::Normal, ProK8sSeverity::info), - _ => (EventType::Normal, ProK8sSeverity::debug), - }; // gather trace_id let otel_trace_id = ctx .event_span(event) .and_then(|span| span.extensions().get::().and_then(|dat| dat.trace_id())); let str_trace_id = otel_trace_id.as_ref().and_then(trace_id_str); - let tx = self.tx.read(); - // write k8s event - if self.config.enable_k8s_events + let k8s_event_type = match *event.metadata().level() { + tracing::Level::ERROR => EventType::Warning, + tracing::Level::WARN => EventType::Warning, + tracing::Level::INFO => EventType::Normal, + _ => EventType::Normal, + }; + 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() { _ = tx - .send(TracingK8sEvent::K8sEvent { - reference: span_fields.to_object(), - event: Box::new(event), - }) + .send(Arc::new(K8sEvent { + object: span_fields.to_object(), + event, + })) .inspect_err(|e| tracing::error!(debug=?e, "Failed to issue event on k8s event from tracing")); } // write/measure pro event - if self.config.enable_prometheus_metrics - && let Some(label) = try_create_pro_label(&span_fields, &event_fields, pro_severity, str_trace_id) - { - if let Some(metrics) = self.metrics.read().as_ref() { - metrics.measure(&label); + let pro_severity = match *event.metadata().level() { + tracing::Level::ERROR => { + if event_fields.critical.is_some_and(|v| v) { + ProK8sSeverity::critical + } else { + ProK8sSeverity::error + } } - - _ = tx.send(TracingK8sEvent::ProLabel { label }).inspect_err( + tracing::Level::WARN => ProK8sSeverity::warning, + tracing::Level::INFO => ProK8sSeverity::info, + _ => ProK8sSeverity::debug, + }; + 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() + { + _ = tx.send(Arc::new(label)).inspect_err( |e| tracing::error!(debug=?e, "Failed to issue event on prometheus label measure from tracing"), ); } + + // write otel event + let otel_severity = match *event.metadata().level() { + tracing::Level::ERROR => { + if event_fields.critical.is_some_and(|v| v) { + Severity::Fatal + } else { + Severity::Error + } + } + tracing::Level::WARN => Severity::Warn, + tracing::Level::INFO => Severity::Info, + _ => return, + }; + 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() + { + _ = tx + .send(Arc::new(OtelObservation::Log { + fields: OtelLogFields { + severity: otel_severity, + message: event_fields.message.clone().unwrap_or_default(), + reason: event_fields.reason.clone().unwrap_or_default(), + action: event_fields.action.clone().unwrap_or_default(), + k8s_name: span_fields.name.clone().unwrap_or_default(), + k8s_kind: span_fields.kind.clone().unwrap_or_default(), + k8s_ns: span_fields.ns.clone(), + k8s_api: span_fields.api.clone(), + k8s_uid: span_fields.uid.clone(), + trace_id: str_trace_id.clone(), + }, + })) + .inspect_err(|e| tracing::error!(debug=?e, "Failed to issue otel observation from tracing")); + } } } #[derive(Clone, Default)] struct SpanFields { + /// Setting this to true will block entire span from emitting k8s events + no_event: Option, + /// Setting this to true will block entire span from emitting prometheus metric labels + no_pro: Option, + /// Setting this to true will block entire span from emitting opentelemetry logs, + no_otel: Option, name: Option, ns: Option, kind: Option, @@ -200,6 +249,9 @@ impl SpanFields { || self.uid.is_some() || self.version.is_some() || self.field_path.is_some() + || self.no_event.is_some() + || self.no_otel.is_some() + || self.no_pro.is_some() } /// scope-independent (k8s/prometheus) field validation @@ -229,6 +281,9 @@ impl SpanFields { other.uid.as_ref().map(|v| self.uid.replace(v.clone())); other.version.as_ref().map(|v| self.version.replace(v.clone())); other.field_path.as_ref().map(|v| self.field_path.replace(v.clone())); + other.no_event.as_ref().map(|v| self.no_event.replace(*v)); + other.no_otel.as_ref().map(|v| self.no_otel.replace(*v)); + other.no_pro.as_ref().map(|v| self.no_pro.replace(*v)); } fn under_owned(&mut self, other: Self) { @@ -239,6 +294,9 @@ impl SpanFields { other.uid.map(|v| self.uid.replace(v)); other.version.map(|v| self.version.replace(v)); other.field_path.map(|v| self.field_path.replace(v)); + other.no_event.map(|v| self.no_event.replace(v)); + other.no_otel.map(|v| self.no_otel.replace(v)); + other.no_pro.map(|v| self.no_pro.replace(v)); } } @@ -256,13 +314,29 @@ impl Visit for SpanFields { } } + fn record_bool(&mut self, field: &tracing::field::Field, value: bool) { + match field.name() { + "no_event" => self.no_event = Some(value), + "no_pro" => self.no_pro = Some(value), + "no_otel" => self.no_otel = Some(value), + _ => {} + } + } + fn record_debug(&mut self, _field: &tracing::field::Field, _value: &dyn std::fmt::Debug) {} } #[derive(Default, Clone)] struct EventFields { + /// Setting this to true will block entire span from emitting k8s events + no_event: Option, + /// Setting this to true will block entire span from emitting prometheus metric labels + no_pro: Option, + /// Setting this to true will block entire span from emitting opentelemetry logs, + no_otel: Option, /// Only for: /// - prometheus label: force event severity to "critical". Only triggers on `ERROR` tracing level + /// - opentelemetry: make event severify to "Fatal". Only triggers on `ERROR` tracing level critical: Option, reason: Option, action: Option, @@ -299,8 +373,12 @@ impl Visit for EventFields { } fn record_bool(&mut self, field: &tracing::field::Field, value: bool) { - if field.name() == "critical" { - self.critical = Some(value) + match field.name() { + "critical" => self.critical = Some(value), + "no_event" => self.no_event = Some(value), + "no_pro" => self.no_pro = Some(value), + "no_otel" => self.no_otel = Some(value), + _ => {} } }