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(); } }