diff --git a/Cargo.lock b/Cargo.lock index 17cb3be..5e85717 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -430,6 +430,28 @@ dependencies = [ "crossbeam-utils", ] +[[package]] +name = "confique" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "06b4f5ec222421e22bb0a8cbaa36b1d2b50fd45cdd30c915ded34108da78b29f" +dependencies = [ + "confique-macro", + "serde", +] + +[[package]] +name = "confique-macro" +version = "0.0.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e4d1754680cd218e7bcb4c960cc9bae3444b5197d64563dccccfdf83cab9e1a7" +dependencies = [ + "heck", + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "constant_time_eq" version = "0.4.2" @@ -1548,7 +1570,7 @@ dependencies = [ ] [[package]] -name = "kuyaseg-operator" +name = "kuyaseg-controller" version = "0.0.1" dependencies = [ "anyhow", @@ -1556,6 +1578,7 @@ dependencies = [ "axum", "base64", "blake3", + "confique", "jiff", "k8s-openapi", "kube", diff --git a/Cargo.toml b/Cargo.toml index 84b3230..a4a25f3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -34,6 +34,7 @@ opentelemetry_sdk.features = ["rt-tokio"] jiff.version = "^0.2" scopeguard.version = "^1" smallvec.version = "^1" +confique.version = "^0.4" ## Marshal schemars.version = "^1" base64.version = "^0.22" diff --git a/kuyaseg-controller/Cargo.toml b/kuyaseg-controller/Cargo.toml index f11a980..511573e 100644 --- a/kuyaseg-controller/Cargo.toml +++ b/kuyaseg-controller/Cargo.toml @@ -32,6 +32,7 @@ prometheus-client.workspace = true jiff.workspace = true scopeguard.workspace = true smallvec.workspace = true +confique.workspace = true ## Marshal schemars.workspace = true base64.workspace = true diff --git a/kuyaseg-controller/src/reporting/mod.rs b/kuyaseg-controller/src/reporting/mod.rs index 6cd30e1..b853132 100644 --- a/kuyaseg-controller/src/reporting/mod.rs +++ b/kuyaseg-controller/src/reporting/mod.rs @@ -1,2 +1,24 @@ +use confique::Config; + pub mod prometheus; pub mod tracing_evt; + +#[derive(Config)] +pub struct ReportingConfig { + #[config(env = "KUYASEG_REPORT_K8S")] + #[config(default = true)] + pub k8s_events: bool, + #[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()), + } +} diff --git a/kuyaseg-controller/src/reporting/prometheus.rs b/kuyaseg-controller/src/reporting/prometheus.rs index 8ce5da0..af013d0 100644 --- a/kuyaseg-controller/src/reporting/prometheus.rs +++ b/kuyaseg-controller/src/reporting/prometheus.rs @@ -1,30 +1,120 @@ +use std::time::{ + Instant, + SystemTime, +}; + +use opentelemetry::trace::TraceContextExt; use prometheus_client::encoding::{ EncodeLabelSet, EncodeLabelValue, }; use prometheus_client::metrics::counter::Counter; +use prometheus_client::metrics::exemplar::{ + CounterWithExemplar, + HistogramWithExemplars, +}; use prometheus_client::metrics::family::Family; +use prometheus_client::registry::Registry; +use tracing_opentelemetry::OpenTelemetrySpanExt; + +use crate::reporting::trace_id_str; #[derive(Default, Clone)] pub struct ProMetrics { - k8s_metrics: Family, + k8s_metrics: Family>, + reconcile_metrics: ProReconcileMetrics, } +/// Will not register unless told to with `register` impl ProMetrics { + pub fn new() -> Self { + Self { + k8s_metrics: Default::default(), + reconcile_metrics: Default::default(), + } + } + + /// Measure labelled evt pub fn measure(&self, label: &ProLabel) { match label { - ProLabel::K8s(label) => { - self.k8s_metrics.get_or_create(label).inc(); + ProLabel::Generic { label, exemplar } => { + self.k8s_metrics + .get_or_create(label) + .inc_by(1, exemplar.clone(), Some(SystemTime::now())); } } } + + /// Start measuring reconcile duration. Registers on `TimedMeasurer` drop + pub fn measure_reconcilation(&self) -> TimedMeasurer { + self.reconcile_metrics.measure() + } + + pub fn register(&self, registry: &mut Registry) { + registry.register("events", "General event counters", self.k8s_metrics.clone()); + registry.register( + "reconcile_duration", + "Reconcile durations", + self.reconcile_metrics.duration.clone(), + ); + registry.register("reconcile_runs", "Reconcilations", self.reconcile_metrics.runs.clone()); + } +} + +#[derive(Clone)] +pub struct ProReconcileMetrics { + runs: Counter, + duration: HistogramWithExemplars, +} + +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 }); + + TimedMeasurer { + label, + start: Instant::now(), + metric: self.duration.clone(), + } + } } -pub enum ProLabel { - K8s(ProK8sLabel), +impl Default for ProReconcileMetrics { + fn default() -> Self { + Self { + runs: Default::default(), + duration: HistogramWithExemplars::new([0.01, 0.1, 0.25, 0.5, 1., 5., 15., 60.].into_iter()), + } + } } -#[derive(Clone, Hash, PartialEq, Eq, EncodeLabelSet)] +/// Reports timing on drop +pub struct TimedMeasurer { + start: Instant, + label: Option, + metric: HistogramWithExemplars, +} + +impl Drop for TimedMeasurer { + fn drop(&mut self) { + let duration_f = self.start.elapsed().as_secs_f64(); + let label = self.label.take(); + self.metric.observe(duration_f, label, Some(SystemTime::now())); + } +} + +pub enum ProLabel { + /// Generic event to be registered as labels + Generic { + label: ProK8sLabel, + exemplar: Option, + }, +} + +/// General purpose label +#[derive(Debug, Clone, Hash, PartialEq, Eq, EncodeLabelSet)] pub struct ProK8sLabel { pub severity: ProK8sSeverity, pub reason: String, @@ -33,7 +123,7 @@ pub struct ProK8sLabel { pub namespace: Option, } -#[derive(Clone, Hash, PartialEq, Eq, EncodeLabelValue)] +#[derive(Debug, Clone, Hash, PartialEq, Eq, EncodeLabelValue)] #[allow(non_camel_case_types)] // verbatim label value encoding must match k8s standard pub enum ProK8sSeverity { critical, @@ -42,3 +132,15 @@ pub enum ProK8sSeverity { info, debug, } + +/// General purpose counter exemplar +#[derive(Default, Debug, Clone, Hash, PartialEq, Eq, EncodeLabelSet)] +pub struct ProK8sExemplar { + pub description: Option, + pub trace_id: Option, +} + +#[derive(Default, Debug, Clone, Hash, PartialEq, Eq, EncodeLabelSet)] +pub struct TraceLabel { + pub trace_id: String, +} diff --git a/kuyaseg-controller/src/reporting/tracing_evt.rs b/kuyaseg-controller/src/reporting/tracing_evt.rs index 7bd4e7b..94b5389 100644 --- a/kuyaseg-controller/src/reporting/tracing_evt.rs +++ b/kuyaseg-controller/src/reporting/tracing_evt.rs @@ -13,16 +13,19 @@ 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::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 @@ -36,6 +39,11 @@ pub enum TracingK8sEvent { }, } +pub struct TracingK8sEventLayerConfig { + pub enable_k8s_events: bool, + pub enable_prometheus_metrics: bool, +} + /// Watches for tracing events, containing information necessary to construst k8s events and write prometheus metrics. /// /// - k8s native events you must record yourself - thats what `tx` is for. @@ -43,13 +51,19 @@ pub enum TracingK8sEvent { /// - 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>, } impl TracingK8sEventLayer { - pub fn new(tx: mpsc::UnboundedSender, metrics: Option) -> Self { + pub fn new( + config: TracingK8sEventLayerConfig, + tx: mpsc::UnboundedSender, + metrics: Option, + ) -> Self { Self { + config, tx: RwLock::new(tx), metrics: RwLock::new(metrics), } @@ -121,11 +135,18 @@ where 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 let Some(event) = event_fields.try_to_event(k8s_event_type) { + if self.config.enable_k8s_events + && let Some(event) = event_fields.try_to_event(k8s_event_type) + { _ = tx .send(TracingK8sEvent::K8sEvent { reference: span_fields.to_object(), @@ -135,7 +156,9 @@ where } // write/measure pro event - if let Some(label) = try_create_pro_label(&span_fields, &event_fields, pro_severity) { + 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); } @@ -286,12 +309,19 @@ fn try_create_pro_label( span_fields: &SpanFields, event_fields: &EventFields, severity: ProK8sSeverity, + trace_id: Option, ) -> Option { - Some(ProLabel::K8s(ProK8sLabel { - severity, - reason: event_fields.reason.clone()?, - name: span_fields.name.clone()?, - kind: span_fields.kind.clone()?, - namespace: span_fields.ns.clone(), - })) + Some(ProLabel::Generic { + label: ProK8sLabel { + severity, + reason: event_fields.reason.clone()?, + name: span_fields.name.clone()?, + kind: span_fields.kind.clone()?, + namespace: span_fields.ns.clone(), + }, + exemplar: Some(ProK8sExemplar { + description: event_fields.message.clone(), + trace_id, + }), + }) }