From e7f6ddf741d8526b0f0db85075ec41cc46a66690 Mon Sep 17 00:00:00 2001 From: spushk Date: Sat, 30 May 2026 10:19:53 +0500 Subject: [PATCH] fix(reporting)!: drop OTEL (for now) & rename layer field names & fix prometheus registry --- .../src/reporting/{k8s_event.rs => k8s.rs} | 2 +- kuyaseg-controller/src/reporting/mod.rs | 134 +++++------------- kuyaseg-controller/src/reporting/otel.rs | 106 -------------- .../src/reporting/prometheus.rs | 9 +- .../src/reporting/tracing_evt.rs | 114 +++++---------- 5 files changed, 76 insertions(+), 289 deletions(-) rename kuyaseg-controller/src/reporting/{k8s_event.rs => k8s.rs} (86%) delete mode 100644 kuyaseg-controller/src/reporting/otel.rs diff --git a/kuyaseg-controller/src/reporting/k8s_event.rs b/kuyaseg-controller/src/reporting/k8s.rs similarity index 86% rename from kuyaseg-controller/src/reporting/k8s_event.rs rename to kuyaseg-controller/src/reporting/k8s.rs index 80ce3b8..8254852 100644 --- a/kuyaseg-controller/src/reporting/k8s_event.rs +++ b/kuyaseg-controller/src/reporting/k8s.rs @@ -25,7 +25,7 @@ impl K8sEventRecorder { .recorder .publish(&event.event, &event.object) .await - .inspect_err(|e| tracing::error!(k8s_no_event=true, error=?e, "Failed to record K8s event")); + .inspect_err(|e| tracing::error!(k13t.no_report=true, debug=?e, "Failed to record K8s event")); } } diff --git a/kuyaseg-controller/src/reporting/mod.rs b/kuyaseg-controller/src/reporting/mod.rs index d24cbfc..7e352a8 100644 --- a/kuyaseg-controller/src/reporting/mod.rs +++ b/kuyaseg-controller/src/reporting/mod.rs @@ -4,6 +4,7 @@ use std::time::Duration; use anyhow::Context; use confique::Config; +use opentelemetry::TraceId; use tokio::sync::broadcast::{ self, Sender, @@ -11,19 +12,27 @@ use tokio::sync::broadcast::{ use tokio::task::JoinSet; use tokio_util::sync::CancellationToken; -use crate::reporting::k8s_event::K8sEvent; -use crate::reporting::otel::OtelObservation; +use crate::reporting::k8s::K8sEvent; use crate::reporting::prometheus::ProLabel; use crate::reporting::tracing_evt::{ TracingK8sEventLayer, TracingK8sEventLayerConfig, }; -pub mod k8s_event; -pub mod otel; +pub mod k8s; pub mod prometheus; pub mod tracing_evt; +pub fn span_trace_id() -> Option { + use opentelemetry::trace::TraceContextExt; + use tracing_opentelemetry::OpenTelemetrySpanExt; + + match tracing::Span::current().context().span().span_context().trace_id() { + TraceId::INVALID => None, + other => Some(other), + } +} + #[derive(Config)] pub struct ReportingConfig { #[config(env = "KUYASEG_REPORT_K8S")] @@ -32,22 +41,13 @@ pub struct ReportingConfig { #[config(env = "KUYASEG_REPORT_PROMETHEUS")] #[config(default = false)] pub prometheus: bool, - #[config(env = "KUYASEG_REPORT_OTEL")] - #[config(default = false)] - pub opentelemetry: bool, -} - -pub fn trace_id_str(trace_id: &opentelemetry::TraceId) -> Option { - match trace_id { - &opentelemetry::TraceId::INVALID => None, - x => Some(x.to_string()), - } } /// You should await on handle somewhere pub struct ReportingInit { pub tracing_layer: TracingK8sEventLayer, pub handle: JoinSet<()>, + pub pro_registry: Option, } pub fn init(k8s_client: kube::Client, ct: CancellationToken) -> anyhow::Result { @@ -61,21 +61,18 @@ pub fn init(k8s_client: kube::Client, ct: CancellationToken) -> anyhow::Result, ct: Cancella Reporter, }; - use crate::reporting::k8s_event::K8sEventRecorder; + use crate::reporting::k8s::K8sEventRecorder; const WARN_THRESHOLD: usize = 10; const MAX_CAPACITY: usize = 100; @@ -126,7 +123,7 @@ fn init_k8s_events(client: kube::Client, joinset: &mut JoinSet<()>, ct: Cancella } queue.push_back(ev); if queue.len() > WARN_THRESHOLD && !warned { - tracing::warn!(k8s_no_event=true, queue_len=queue.len(), "K8s event queue exceeded threshold"); + tracing::warn!(k13t.no_report=true, debug=?queue.len(), "K8s event queue exceeded threshold"); warned = true; } else if queue.len() <= WARN_THRESHOLD { warned = false; @@ -135,7 +132,7 @@ fn init_k8s_events(client: kube::Client, joinset: &mut JoinSet<()>, ct: Cancella rec_limit = 4; } Err(e) => { - tracing::error!(k8s_no_event=true, debug=?e, "Failed to receive k8s event from tracing"); + tracing::error!(k13t.no_report=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; @@ -145,7 +142,7 @@ fn init_k8s_events(client: kube::Client, joinset: &mut JoinSet<()>, ct: Cancella } } tracing::error!( - no_event = true, + k13t.no_report = true, "K8s Event reporting from tracing stopped" ); }); @@ -153,7 +150,10 @@ fn init_k8s_events(client: kube::Client, joinset: &mut JoinSet<()>, ct: Cancella tx } -fn init_pro_metrics(joinset: &mut JoinSet<()>, ct: CancellationToken) -> Sender> { +fn init_pro_metrics( + joinset: &mut JoinSet<()>, + ct: CancellationToken, +) -> (Sender>, prometheus_client::registry::Registry) { use prometheus_client::registry::Registry; use crate::reporting::prometheus::ProMetrics; @@ -161,10 +161,7 @@ fn init_pro_metrics(joinset: &mut JoinSet<()>, ct: CancellationToken) -> Sender< let (tx, mut rx) = broadcast::channel::>(512); let metrics = ProMetrics::new(); - let registration_cancellation = CancellationToken::new(); - let rtx = tx.clone(); - let ctk = registration_cancellation.clone(); let promt = metrics.clone(); let consumer_ct = ct.clone(); joinset.spawn(async move { @@ -180,7 +177,7 @@ fn init_pro_metrics(joinset: &mut JoinSet<()>, ct: CancellationToken) -> Sender< rec_limit = 4; } Err(e) => { - tracing::error!(k8s_no_pro=true, debug=?e, "Failed to receive prometheus measurement from tracing"); + tracing::error!(k13t.no_report=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; @@ -189,82 +186,17 @@ fn init_pro_metrics(joinset: &mut JoinSet<()>, ct: CancellationToken) -> Sender< } } } - ctk.cancel(); tracing::error!( - no_pro = true, + k13t.no_report=true, "Prometheus measurement from tracing stopped" ); }); - let ctk = ct.clone(); - joinset.spawn(async move { + let registry = { let mut registry = Registry::default(); - loop { - tokio::select! { - _ = tokio::time::sleep(Duration::from_secs(10)) => { - metrics.register(&mut registry); - }, - _ = registration_cancellation.cancelled() => { - tracing::warn!(k8s_no_pro=true, "Stopped publishing prometheus metrics: consumer job cancelled"); - break; - } - _ = ctk.cancelled() => { - tracing::warn!(k8s_no_pro=true, "Stopped publishing prometheus metrics: shutdown"); - break; - } - } - } - }); + metrics.register(&mut registry); + registry + }; - tx -} - -fn init_otel(joinset: &mut JoinSet<()>, ct: CancellationToken) -> anyhow::Result>> { - use crate::reporting::otel::OtelAggregator; - - let exporter = opentelemetry_otlp::LogExporter::builder() - .with_tonic() - .build() - .context("Failed to build OTLP log exporter")?; - - let provider = opentelemetry_sdk::logs::SdkLoggerProvider::builder() - .with_batch_exporter(exporter) - .build(); - - let aggregator = Arc::new(OtelAggregator::new(provider)); - let task_aggregator = aggregator.clone(); - - let (tx, mut rx) = broadcast::channel::>(512); - - let rtx = tx.clone(); - joinset.spawn(async move { - let mut rec_limit = 4; - while rec_limit > 0 { - tokio::select! { - biased; - _ = ct.cancelled() => break, - msg = rx.recv() => { - match msg { - Ok(obs) => { - task_aggregator.observe(&obs); - rec_limit = 4; - } - Err(e) => { - tracing::error!( - k8s_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!(k8s_no_otel = true, "OpenTelemetry reporting from tracing stopped"); - }); - - Ok(tx) + (tx, registry) } diff --git a/kuyaseg-controller/src/reporting/otel.rs b/kuyaseg-controller/src/reporting/otel.rs deleted file mode 100644 index f850d4c..0000000 --- a/kuyaseg-controller/src/reporting/otel.rs +++ /dev/null @@ -1,106 +0,0 @@ -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") - } -} - -impl Drop for OtelAggregator { - fn drop(&mut self) { - self.force_flush(); - } -} diff --git a/kuyaseg-controller/src/reporting/prometheus.rs b/kuyaseg-controller/src/reporting/prometheus.rs index c648ffb..dfcc3b1 100644 --- a/kuyaseg-controller/src/reporting/prometheus.rs +++ b/kuyaseg-controller/src/reporting/prometheus.rs @@ -4,7 +4,6 @@ use std::time::{ SystemTime, }; -use opentelemetry::trace::TraceContextExt; use prometheus_client::encoding::{ EncodeLabelSet, EncodeLabelValue, @@ -16,9 +15,8 @@ use prometheus_client::metrics::exemplar::{ }; use prometheus_client::metrics::family::Family; use prometheus_client::registry::Registry; -use tracing_opentelemetry::OpenTelemetrySpanExt; -use crate::reporting::trace_id_str; +use crate::reporting::span_trace_id; #[derive(Default, Clone)] pub struct ProMetrics { @@ -72,8 +70,9 @@ pub struct ProReconcileMetrics { impl ProReconcileMetrics { pub fn measure(&self) -> TimedMeasurer { self.runs.inc(); - let trace_id = tracing::Span::current().context().span().span_context().trace_id(); - let label = trace_id_str(&trace_id).map(|trace_id| TraceLabel { trace_id }); + let label = span_trace_id().map(|id| TraceLabel { + trace_id: id.to_string(), + }); TimedMeasurer { label, diff --git a/kuyaseg-controller/src/reporting/tracing_evt.rs b/kuyaseg-controller/src/reporting/tracing_evt.rs index 7663f3a..f94b612 100644 --- a/kuyaseg-controller/src/reporting/tracing_evt.rs +++ b/kuyaseg-controller/src/reporting/tracing_evt.rs @@ -5,7 +5,6 @@ use kube::runtime::events::{ Event, EventType, }; -use opentelemetry::logs::Severity; use tokio::sync::broadcast::Sender; use tracing::field::Visit; use tracing::span::Attributes; @@ -14,31 +13,23 @@ use tracing::{ Id, Subscriber, }; -use tracing_opentelemetry::OtelData; 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::k8s::K8sEvent; use crate::reporting::prometheus::{ ProK8sExemplar, ProK8sLabel, ProK8sSeverity, ProLabel, }; -use crate::reporting::trace_id_str; pub struct TracingK8sEventLayerConfig { /// 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. @@ -46,10 +37,11 @@ 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 rate-limited to 10 events/min +/// +/// Recognizes fields with `k13t` (KubernatesEvent) prefix. pub struct TracingK8sEventLayer { k8s_tx: Option>>, pro_tx: Option>>, - otel_tx: Option>>, } impl TracingK8sEventLayer { @@ -57,7 +49,6 @@ impl TracingK8sEventLayer { Self { k8s_tx: config.k8s_events, pro_tx: config.prometheus, - otel_tx: config.otel, } } } @@ -90,6 +81,7 @@ where return; } + let span = ctx.event_span(event); let scope = match ctx.event_scope(event) { Some(s) => s, None => { @@ -122,11 +114,13 @@ where return; } - // 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 trace_id = span + .and_then(|s| { + s.extensions() + .get::() + .map(|ctx| ctx.trace_id()) + }) + .map(|id| id.to_string()); // write k8s event let k8s_event_type = match *event.metadata().level() { @@ -145,7 +139,7 @@ where object: span_fields.to_object(), event, })) - .inspect_err(|e| tracing::error!(debug=?e, "Failed to issue event on k8s event from tracing")); + .inspect_err(|e| tracing::error!(k13t.no_report=true, debug=?e, "Failed to issue event on k8s event from tracing")); } // write/measure pro event @@ -163,48 +157,13 @@ 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(label) = try_create_pro_label(&span_fields, &event_fields, pro_severity, trace_id) && 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"), + |e| tracing::error!(k13t.no_report=true, 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.as_ref() - { - _ = 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")); - } } } @@ -289,22 +248,22 @@ impl SpanFields { impl Visit for SpanFields { fn record_str(&mut self, field: &tracing::field::Field, value: &str) { match field.name() { - "k8s_name" => self.name = Some(value.to_string()), - "k8s_ns" => self.ns = Some(value.to_string()), - "k8s_kind" => self.kind = Some(value.to_string()), - "k8s_api" => self.api = Some(value.to_string()), - "k8s_uid" => self.uid = Some(value.to_string()), - "k8s_version" => self.version = Some(value.to_string()), - "k8s_field_path" => self.field_path = Some(value.to_string()), + "k13t.name" => self.name = Some(value.to_string()), + "k13t.ns" => self.ns = Some(value.to_string()), + "k13t.kind" => self.kind = Some(value.to_string()), + "k13t.api" => self.api = Some(value.to_string()), + "k13t.uid" => self.uid = Some(value.to_string()), + "k13t.version" => self.version = Some(value.to_string()), + "k13t.field_path" => self.field_path = Some(value.to_string()), _ => {} } } fn record_bool(&mut self, field: &tracing::field::Field, value: bool) { match field.name() { - "k8s_no_event" => self.no_event = Some(value), - "k8s_no_pro" => self.no_pro = Some(value), - "k8s_no_otel" => self.no_otel = Some(value), + "k13t.no_event" => self.no_event = Some(value), + "k13t.no_pro" => self.no_pro = Some(value), + "k13t.no_otel" => self.no_otel = Some(value), _ => {} } } @@ -314,12 +273,14 @@ impl Visit for SpanFields { #[derive(Default, Clone)] struct EventFields { - /// Setting this to true will block entire span from emitting k8s events + /// Setting this to true will prevent from emitting k8s events no_event: Option, - /// Setting this to true will block entire span from emitting prometheus metric labels + /// Setting this to true will prevent from emitting prometheus metric labels no_pro: Option, - /// Setting this to true will block entire span from emitting opentelemetry logs, + /// Setting this to true will prevent from emitting opentelemetry logs, no_otel: Option, + /// Setting this to true will prevent from emitting any pro/otel/k8s events + no_report: 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 @@ -332,7 +293,7 @@ struct EventFields { impl EventFields { /// Scope independend custom field validation fn check_fields(&self) -> bool { - self.reason.is_some() && self.action.is_some() + !self.no_report.is_some_and(|v| v) && self.reason.is_some() && self.action.is_some() } /// Return `None` if hard-requirements on fields not met. @@ -351,19 +312,20 @@ impl EventFields { impl Visit for EventFields { fn record_str(&mut self, field: &tracing::field::Field, value: &str) { match field.name() { - "reason" => self.reason = Some(value.to_string()), - "action" => self.action = Some(value.to_string()), - "message" => self.message = Some(value.to_string()), + "k13t.reason" => self.reason = Some(value.to_string()), + "k13t.action" => self.action = Some(value.to_string()), + "k13t.message" => self.message = Some(value.to_string()), _ => {} } } fn record_bool(&mut self, field: &tracing::field::Field, value: bool) { match field.name() { - "critical" => self.critical = Some(value), - "k8s_no_event" => self.no_event = Some(value), - "k8s_no_pro" => self.no_pro = Some(value), - "k8s_no_otel" => self.no_otel = Some(value), + "k13t.critical" => self.critical = Some(value), + "k13t.no_event" => self.no_event = Some(value), + "k13t.no_pro" => self.no_pro = Some(value), + "k13t.no_otel" => self.no_otel = Some(value), + "k13t.no_report" => self.no_report = Some(value), _ => {} } }