kubeworks/kube-yaseg
Watch
1
0
Fork
You've already forked kube-yaseg
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(reporting)!: drop OTEL (for now) & rename layer field names & fix prometheus registry

Browse source
This commit is contained in:
spushk 2026-05-30 10:19:53 +05:00 • committed by spushk
commit e7f6ddf741
Signed by: spushk
GPG key ID: 7797B761E105E8AA
5 changed files with 76 additions and 289 deletions
Download patch file Download diff file Expand all files Collapse all files

36
kuyaseg-controller/src/reporting/k8s.rs Normal file
View file

@ -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!(k13t.no_report=true, debug=?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()
}
}

134
kuyaseg-controller/src/reporting/mod.rs
View file

@ -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<TraceId> {
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<String> {
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<prometheus_client::registry::Registry>,
}
pub fn init(k8s_client: kube::Client, ct: CancellationToken) -> anyhow::Result<ReportingInit> {
@ -61,21 +61,18 @@ pub fn init(k8s_client: kube::Client, ct: CancellationToken) -> anyhow::Result<R
let k8s_tx = config
.k8s_events
.then(|| init_k8s_events(k8s_client, &mut reporting_jobs, ct.clone()));
let pro_tx = config
let (pro_tx, pro_registry) = 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()?;
.then(|| init_pro_metrics(&mut reporting_jobs, ct.clone()))
.unzip();
Ok(ReportingInit {
tracing_layer: TracingK8sEventLayer::new(TracingK8sEventLayerConfig {
k8s_events: k8s_tx,
prometheus: pro_tx,
otel: otel_tx,
}),
handle: reporting_jobs,
pro_registry,
})
}
@ -85,7 +82,7 @@ fn init_k8s_events(client: kube::Client, joinset: &mut JoinSet<()>, 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<Arc<ProLabel>> {
fn init_pro_metrics(
joinset: &mut JoinSet<()>,
ct: CancellationToken,
) -> (Sender<Arc<ProLabel>>, 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::<Arc<ProLabel>>(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<Sender<Arc<OtelObservation>>> {
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::<Arc<OtelObservation>>(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)
}

106
kuyaseg-controller/src/reporting/otel.rs
View file

@ -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<String>,
pub k8s_api: Option<String>,
pub k8s_uid: Option<String>,
pub trace_id: Option<String>,
}
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: <SdkLoggerProvider as LoggerProvider>::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();
}
}

9
kuyaseg-controller/src/reporting/prometheus.rs
View file

@ -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,

114
kuyaseg-controller/src/reporting/tracing_evt.rs
View file

@ -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<Sender<Arc<K8sEvent>>>,
/// set to enable prometheus metrics from tracing
pub prometheus: Option<Sender<Arc<ProLabel>>>,
/// set to enable otel observations from tracing
pub otel: Option<Sender<Arc<OtelObservation>>>,
}
/// 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<Sender<Arc<K8sEvent>>>,
pro_tx: Option<Sender<Arc<ProLabel>>>,
otel_tx: Option<Sender<Arc<OtelObservation>>>,
}
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::<OtelData>().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::<opentelemetry::trace::SpanContext>()
.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<bool>,
/// 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<bool>,
/// 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<bool>,
/// Setting this to true will prevent from emitting any pro/otel/k8s events
no_report: Option<bool>,
/// 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),
_ => {}
}
}