chore(controller/reporting): init, setup prometheus basics, setup tracing event propagation

kuyaseg-controller/src/bin/generate_crds.rs

@@ -0,0 +1 @@
+fn main() {}

kuyaseg-controller/src/lib.rs

@@ -0,0 +1,4 @@
+#![feature(sync_nonpoison, nonpoison_rwlock)]
+
+pub mod reporting;
+pub mod utils;

kuyaseg-controller/src/main.rs

@@ -0,0 +1,7 @@
+#[global_allocator]
+static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    todo!("WIP")
+}

kuyaseg-controller/src/reporting/mod.rs

@@ -0,0 +1,2 @@
+pub mod prometheus;
+pub mod tracing_evt;

kuyaseg-controller/src/reporting/prometheus.rs

@@ -0,0 +1,44 @@
+use prometheus_client::encoding::{
+    EncodeLabelSet,
+    EncodeLabelValue,
+};
+use prometheus_client::metrics::counter::Counter;
+use prometheus_client::metrics::family::Family;
+
+#[derive(Default, Clone)]
+pub struct ProMetrics {
+    k8s_metrics: Family<ProK8sLabel, Counter>,
+}
+
+impl ProMetrics {
+    pub fn measure(&self, label: &ProLabel) {
+        match label {
+            ProLabel::K8s(label) => {
+                self.k8s_metrics.get_or_create(label).inc();
+            }
+        }
+    }
+}
+
+pub enum ProLabel {
+    K8s(ProK8sLabel),
+}
+
+#[derive(Clone, Hash, PartialEq, Eq, EncodeLabelSet)]
+pub struct ProK8sLabel {
+    pub severity: ProK8sSeverity,
+    pub reason: String,
+    pub name: String,
+    pub kind: String,
+    pub namespace: Option<String>,
+}
+
+#[derive(Clone, Hash, PartialEq, Eq, EncodeLabelValue)]
+#[allow(non_camel_case_types)] // verbatim label value encoding must match k8s standard
+pub enum ProK8sSeverity {
+    critical,
+    error,
+    warning,
+    info,
+    debug,
+}

kuyaseg-controller/src/reporting/tracing_evt.rs

@@ -0,0 +1,293 @@
+use std::sync::nonpoison::RwLock;
+
+use k8s_openapi::api::core::v1::ObjectReference;
+use kube::runtime::events::{
+    Event,
+    EventType,
+};
+use tokio::sync::mpsc;
+use tracing::field::Visit;
+use tracing::span::Attributes;
+use tracing::{
+    Event as TracingEvent,
+    Id,
+    Subscriber,
+};
+use tracing_subscriber::Layer;
+use tracing_subscriber::layer::Context;
+use tracing_subscriber::registry::LookupSpan;
+
+use crate::reporting::prometheus::{
+    ProK8sLabel,
+    ProK8sSeverity,
+    ProLabel,
+    ProMetrics,
+};
+
+/// - k8s events are not automatically recorded
+/// - prometheus labels are recorded if metrics are set for layer
+pub enum TracingK8sEvent {
+    K8sEvent { reference: ObjectReference, event: Event },
+    ProLabel { label: ProLabel },
+}
+
+/// 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.
+/// - prometheus metrics are written automatically (if metrics set), but labels are also send with `tx`.
+/// - 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 {
+    tx: RwLock<mpsc::UnboundedSender<TracingK8sEvent>>,
+    metrics: RwLock<Option<ProMetrics>>,
+}
+
+impl TracingK8sEventLayer {
+    pub fn new(tx: mpsc::UnboundedSender<TracingK8sEvent>, metrics: Option<ProMetrics>) -> Self {
+        Self {
+            tx: RwLock::new(tx),
+            metrics: RwLock::new(metrics),
+        }
+    }
+
+    pub fn reload_tx(&self, tx: mpsc::UnboundedSender<TracingK8sEvent>) {
+        *self.tx.write() = tx;
+    }
+
+    pub fn reload_metrics(&self, metrics: Option<ProMetrics>) {
+        *self.metrics.write() = metrics;
+    }
+}
+
+impl<S> Layer<S> for TracingK8sEventLayer
+where
+    S: Subscriber + for<'a> LookupSpan<'a>,
+{
+    fn on_new_span(&self, attrs: &Attributes<'_>, id: &Id, ctx: Context<'_, S>) {
+        let mut fields = SpanFields::default();
+        attrs.record(&mut fields);
+
+        if fields.has_any_fields()
+            && let Some(span) = ctx.span(id)
+        {
+            if let Some(prev) = span.extensions_mut().get_mut::<SpanFields>() {
+                prev.under_owned(fields);
+            } else {
+                span.extensions_mut().insert(fields);
+            }
+        }
+    }
+
+    fn on_event(&self, event: &TracingEvent<'_>, ctx: Context<'_, S>) {
+        let scope = match ctx.event_scope(event) {
+            Some(s) => s,
+            None => return,
+        };
+
+        // construct & check event fields
+        let mut span_fields = SpanFields::default();
+        for span in scope.from_root() {
+            if let Some(fields) = span.extensions().get::<SpanFields>() {
+                span_fields.under(fields);
+            }
+        }
+        if !span_fields.check_fields() {
+            return;
+        }
+
+        // construct & check pro fields
+        let mut event_fields = EventFields::default();
+        event.record(&mut event_fields);
+        if !event_fields.check_fields() {
+            return;
+        }
+
+        // calculate event types
+        let (k8s_event_type, pro_severity) = match *event.metadata().level() {
+            tracing::Level::ERROR => (
+                EventType::Warning,
+                if event_fields.critical.is_some_and(|v| v) {
+                    ProK8sSeverity::critical
+                } else {
+                    ProK8sSeverity::error
+                },
+            ),
+            tracing::Level::WARN => (EventType::Warning, ProK8sSeverity::warning),
+            tracing::Level::INFO => (EventType::Normal, ProK8sSeverity::info),
+            _ => (EventType::Normal, ProK8sSeverity::debug),
+        };
+
+        let tx = self.tx.read();
+
+        // write k8s event
+        if let Some(event) = event_fields.try_to_event(k8s_event_type) {
+            _ = tx
+                .send(TracingK8sEvent::K8sEvent {
+                    reference: span_fields.to_object(),
+                    event,
+                })
+                .inspect_err(|e| tracing::error!(debug=?e, "Failed to issue event on k8s event from tracing"));
+        }
+
+        // write/measure pro event
+        if let Some(label) = try_create_pro_label(&span_fields, &event_fields, pro_severity) {
+            self.metrics.read().as_ref().map(|metrics| {
+                metrics.measure(&label);
+            });
+
+            _ = tx.send(TracingK8sEvent::ProLabel { label }).inspect_err(
+                |e| tracing::error!(debug=?e, "Failed to issue event on prometheus label measure from tracing"),
+            );
+        }
+    }
+}
+
+#[derive(Clone, Default)]
+struct SpanFields {
+    name: Option<String>,
+    ns: Option<String>,
+    kind: Option<String>,
+    api: Option<String>,
+    uid: Option<String>,
+    version: Option<String>,
+    field_path: Option<String>,
+}
+
+impl SpanFields {
+    #[inline]
+    fn has_any_fields(&self) -> bool {
+        self.name.is_some()
+            || self.ns.is_some()
+            || self.kind.is_some()
+            || self.api.is_some()
+            || self.uid.is_some()
+            || self.version.is_some()
+            || self.field_path.is_some()
+    }
+
+    /// scope-independent (k8s/prometheus) field validation
+    fn check_fields(&self) -> bool {
+        self.name.is_some() && self.kind.is_some()
+    }
+
+    /// Never fails to `None` since no hard-requirements.
+    /// Custom field validation is at `check_fields`
+    fn to_object(&self) -> ObjectReference {
+        ObjectReference {
+            name: self.name.clone(),
+            kind: self.kind.clone(),
+            api_version: self.api.clone(),
+            namespace: self.ns.clone(),
+            resource_version: self.version.clone(),
+            uid: self.uid.clone(),
+            field_path: self.field_path.clone(),
+        }
+    }
+
+    fn under(&mut self, other: &Self) {
+        other.name.as_ref().map(|v| self.name.replace(v.clone()));
+        other.ns.as_ref().map(|v| self.ns.replace(v.clone()));
+        other.kind.as_ref().map(|v| self.kind.replace(v.clone()));
+        other.api.as_ref().map(|v| self.api.replace(v.clone()));
+        other.uid.as_ref().map(|v| self.uid.replace(v.clone()));
+        other.version.as_ref().map(|v| self.version.replace(v.clone()));
+        other.field_path.as_ref().map(|v| self.field_path.replace(v.clone()));
+    }
+
+    fn under_owned(&mut self, other: Self) {
+        other.name.map(|v| self.name.replace(v));
+        other.ns.map(|v| self.ns.replace(v));
+        other.kind.map(|v| self.kind.replace(v));
+        other.api.map(|v| self.api.replace(v));
+        other.uid.map(|v| self.uid.replace(v));
+        other.version.map(|v| self.version.replace(v));
+        other.field_path.map(|v| self.field_path.replace(v));
+    }
+}
+
+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()),
+            _ => {}
+        }
+    }
+
+    fn record_debug(&mut self, _field: &tracing::field::Field, _value: &dyn std::fmt::Debug) {}
+}
+
+#[derive(Default, Clone)]
+struct EventFields {
+    /// Only for:
+    /// - prometheus label: force event severity to "critical". Only triggers on `ERROR` tracing level
+    critical: Option<bool>,
+    reason: Option<String>,
+    action: Option<String>,
+    message: Option<String>,
+}
+
+impl EventFields {
+    /// Scope independend custom field validation
+    fn check_fields(&self) -> bool {
+        self.reason.is_some() && self.action.is_some()
+    }
+
+    /// Return `None` if hard-requirements on fields not met.
+    /// Custom field validation is `check_fields`
+    fn try_to_event(&self, type_: EventType) -> Option<Event> {
+        Some(Event {
+            type_,
+            reason: self.reason.clone()?,
+            action: self.action.clone()?,
+            note: self.message.clone(),
+            secondary: None,
+        })
+    }
+}
+
+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()),
+            _ => {}
+        }
+    }
+
+    fn record_bool(&mut self, field: &tracing::field::Field, value: bool) {
+        match field.name() {
+            "critical" => self.critical = Some(value),
+            _ => {}
+        }
+    }
+
+    fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
+        if field.name() == "message" {
+            self.message = Some(format!("{value:?}"));
+        }
+    }
+}
+
+/// Try to create ProK8sLabel from provided fields.
+/// Returns `None` if hard requirements on fields are not satisfied.
+/// Custom field validation is done on `SpanFields`/`EventFields` level with `check_fields`
+fn try_create_pro_label(
+    span_fields: &SpanFields,
+    event_fields: &EventFields,
+    severity: ProK8sSeverity,
+) -> Option<ProLabel> {
+    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(),
+    }))
+}

kuyaseg-controller/src/utils/mod.rs

@@ -0,0 +1 @@
+