feat(controller/reporting): reconcile-rspecific measurements & prometheus metric retains tracing messages

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

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"

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

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<String> {
+    match trace_id {
+        &opentelemetry::TraceId::INVALID => None,
+        x => Some(x.to_string()),
+    }
+}

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<ProK8sLabel, Counter>,
+    k8s_metrics: Family<ProK8sLabel, CounterWithExemplar<ProK8sExemplar>>,
+    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) => {
+            ProLabel::Generic { label, exemplar } => {
-                self.k8s_metrics.get_or_create(label).inc();
+                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<TraceLabel>,
+}
+
+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 {
+impl Default for ProReconcileMetrics {
-    K8s(ProK8sLabel),
+    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<TraceLabel>,
+    metric: HistogramWithExemplars<TraceLabel>,
+}
+
+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<ProK8sExemplar>,
+    },
+}
+
+/// 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<String>,
 }
 
-#[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<String>,
+    pub trace_id: Option<String>,
+}
+
+#[derive(Default, Debug, Clone, Hash, PartialEq, Eq, EncodeLabelSet)]
+pub struct TraceLabel {
+    pub trace_id: String,
+}

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<mpsc::UnboundedSender<TracingK8sEvent>>,
     metrics: RwLock<Option<ProMetrics>>,
 }
 
 impl TracingK8sEventLayer {
-    pub fn new(tx: mpsc::UnboundedSender<TracingK8sEvent>, metrics: Option<ProMetrics>) -> Self {
+    pub fn new(
+        config: TracingK8sEventLayerConfig,
+        tx: mpsc::UnboundedSender<TracingK8sEvent>,
+        metrics: Option<ProMetrics>,
+    ) -> 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::<OtelData>().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<String>,
 ) -> Option<ProLabel> {
-    Some(ProLabel::K8s(ProK8sLabel {
+    Some(ProLabel::Generic {
-        severity,
+        label: ProK8sLabel {
-        reason: event_fields.reason.clone()?,
+            severity,
-        name: span_fields.name.clone()?,
+            reason: event_fields.reason.clone()?,
-        kind: span_fields.kind.clone()?,
+            name: span_fields.name.clone()?,
-        namespace: span_fields.ns.clone(),
+            kind: span_fields.kind.clone()?,
-    }))
+            namespace: span_fields.ns.clone(),
+        },
+        exemplar: Some(ProK8sExemplar {
+            description: event_fields.message.clone(),
+            trace_id,
+        }),
+    })
 }