Try out offficial prometheus_client for metrics (#72)

* Try out official `prometheus_client` crate

Signed-off-by: clux <[email protected]>

* fix metric setting

Signed-off-by: clux <[email protected]>

* make metrics work with registry

Signed-off-by: clux <[email protected]>

* fix prefix

Signed-off-by: clux <[email protected]>

* make exemplars work

Signed-off-by: clux <[email protected]>

* openmetrics response header

Signed-off-by: clux <[email protected]>

* simplify

Signed-off-by: clux <[email protected]>

---------

Signed-off-by: clux <[email protected]>

Author: clux

Cargo.lock

@@ -34,7 +34,6 @@ schemars = { version = "0.8.12", features = ["chrono"] }
 serde = { version = "1.0.185", features = ["derive"] }
 serde_json = "1.0.105"
 serde_yaml = "0.9.25"
-prometheus = "0.13.3"
 chrono = { version = "0.4.26", features = ["serde"] }
 tracing = "0.1.37"
 tracing-subscriber = { version = "0.3.17", features = ["json", "env-filter"] }
@@ -44,6 +43,7 @@ opentelemetry-otlp = { version = "0.13.0", features = ["tokio"], optional = true
 tonic = { version = "0.9", optional = true }
 thiserror = "1.0.47"
 anyhow = "1.0.75"
+prometheus-client = "0.22.2"
 
 [dev-dependencies]
 assert-json-diff = "2.0.2"

Cargo.toml

@@ -12,7 +12,7 @@ generate:
 
 # run with opentelemetry
 run-telemetry:
-  OPENTELEMETRY_ENDPOINT_URL=http://127.0.0.1:55680 RUST_LOG=info,kube=trace,controller=debug cargo run --features=telemetry
+  OPENTELEMETRY_ENDPOINT_URL=http://127.0.0.1:55680 RUST_LOG=info,kube=debug,controller=debug cargo run --features=telemetry
 
 # run without opentelemetry
 run:

justfile

@@ -53,14 +53,14 @@ pub struct Context {
     /// Diagnostics read by the web server
     pub diagnostics: Arc<RwLock<Diagnostics>>,
     /// Prometheus metrics
-    pub metrics: Metrics,
+    pub metrics: Arc<Metrics>,
 }
 
 #[instrument(skip(ctx, doc), fields(trace_id))]
 async fn reconcile(doc: Arc<Document>, ctx: Arc<Context>) -> Result<Action> {
     let trace_id = telemetry::get_trace_id();
     Span::current().record("trace_id", &field::display(&trace_id));
-    let _timer = ctx.metrics.count_and_measure();
+    let _timer = ctx.metrics.reconcile.count_and_measure(&trace_id);
     ctx.diagnostics.write().await.last_event = Utc::now();
     let ns = doc.namespace().unwrap(); // doc is namespace scoped
     let docs: Api<Document> = Api::namespaced(ctx.client.clone(), &ns);
@@ -78,7 +78,7 @@ async fn reconcile(doc: Arc<Document>, ctx: Arc<Context>) -> Result<Action> {
 
 fn error_policy(doc: Arc<Document>, error: &Error, ctx: Arc<Context>) -> Action {
     warn!("reconcile failed: {:?}", error);
-    ctx.metrics.reconcile_failure(&doc, error);
+    ctx.metrics.reconcile.set_failure(&doc, error);
     Action::requeue(Duration::from_secs(5 * 60))
 }
 
@@ -171,15 +171,18 @@ impl Diagnostics {
 pub struct State {
     /// Diagnostics populated by the reconciler
     diagnostics: Arc<RwLock<Diagnostics>>,
-    /// Metrics registry
-    registry: prometheus::Registry,
+    /// Metrics
+    metrics: Arc<Metrics>,
 }
 
 /// State wrapper around the controller outputs for the web server
 impl State {
     /// Metrics getter
-    pub fn metrics(&self) -> Vec<prometheus::proto::MetricFamily> {
-        self.registry.gather()
+    pub fn metrics(&self) -> String {
+        let mut buffer = String::new();
+        let registry = &*self.metrics.registry;
+        prometheus_client::encoding::text::encode(&mut buffer, &registry).unwrap();
+        buffer
     }
 
     /// State getter
@@ -191,7 +194,7 @@ impl State {
     pub fn to_context(&self, client: Client) -> Arc<Context> {
         Arc::new(Context {
             client,
-            metrics: Metrics::default().register(&self.registry).unwrap(),
+            metrics: self.metrics.clone(),
             diagnostics: self.diagnostics.clone(),
         })
     }
@@ -218,12 +221,15 @@ pub async fn run(state: State) {
 #[cfg(test)]
 mod test {
     use super::{error_policy, reconcile, Context, Document};
-    use crate::fixtures::{timeout_after_1s, Scenario};
+    use crate::{
+        fixtures::{timeout_after_1s, Scenario},
+        metrics::ErrorLabels,
+    };
     use std::sync::Arc;
 
     #[tokio::test]
     async fn documents_without_finalizer_gets_a_finalizer() {
-        let (testctx, fakeserver, _) = Context::test();
+        let (testctx, fakeserver) = Context::test();
         let doc = Document::test();
         let mocksrv = fakeserver.run(Scenario::FinalizerCreation(doc.clone()));
         reconcile(Arc::new(doc), testctx).await.expect("reconciler");
@@ -232,7 +238,7 @@ mod test {
 
     #[tokio::test]
     async fn finalized_doc_causes_status_patch() {
-        let (testctx, fakeserver, _) = Context::test();
+        let (testctx, fakeserver) = Context::test();
         let doc = Document::test().finalized();
         let mocksrv = fakeserver.run(Scenario::StatusPatch(doc.clone()));
         reconcile(Arc::new(doc), testctx).await.expect("reconciler");
@@ -241,7 +247,7 @@ mod test {
 
     #[tokio::test]
     async fn finalized_doc_with_hide_causes_event_and_hide_patch() {
-        let (testctx, fakeserver, _) = Context::test();
+        let (testctx, fakeserver) = Context::test();
         let doc = Document::test().finalized().needs_hide();
         let scenario = Scenario::EventPublishThenStatusPatch("HideRequested".into(), doc.clone());
         let mocksrv = fakeserver.run(scenario);
@@ -251,7 +257,7 @@ mod test {
 
     #[tokio::test]
     async fn finalized_doc_with_delete_timestamp_causes_delete() {
-        let (testctx, fakeserver, _) = Context::test();
+        let (testctx, fakeserver) = Context::test();
         let doc = Document::test().finalized().needs_delete();
         let mocksrv = fakeserver.run(Scenario::Cleanup("DeleteRequested".into(), doc.clone()));
         reconcile(Arc::new(doc), testctx).await.expect("reconciler");
@@ -260,7 +266,7 @@ mod test {
 
     #[tokio::test]
     async fn illegal_doc_reconcile_errors_which_bumps_failure_metric() {
-        let (testctx, fakeserver, _registry) = Context::test();
+        let (testctx, fakeserver) = Context::test();
         let doc = Arc::new(Document::illegal().finalized());
         let mocksrv = fakeserver.run(Scenario::RadioSilence);
         let res = reconcile(doc.clone(), testctx.clone()).await;
@@ -270,12 +276,12 @@ mod test {
         assert!(err.to_string().contains("IllegalDocument"));
         // calling error policy with the reconciler error should cause the correct metric to be set
         error_policy(doc.clone(), &err, testctx.clone());
-        //dbg!("actual metrics: {}", registry.gather());
-        let failures = testctx
-            .metrics
-            .failures
-            .with_label_values(&["illegal", "finalizererror(applyfailed(illegaldocument))"])
-            .get();
+        let err_labels = ErrorLabels {
+            instance: "illegal".into(),
+            error: "finalizererror(applyfailed(illegaldocument))".into(),
+        };
+        let metrics = &testctx.metrics.reconcile;
+        let failures = metrics.failures.get_or_create(&err_labels).get();
         assert_eq!(failures, 1);
     }
 

src/controller.rs

@@ -1,9 +1,8 @@
 //! Helper methods only available for tests
-use crate::{Context, Document, DocumentSpec, DocumentStatus, Metrics, Result, DOCUMENT_FINALIZER};
+use crate::{Context, Document, DocumentSpec, DocumentStatus, Result, DOCUMENT_FINALIZER};
 use assert_json_diff::assert_json_include;
 use http::{Request, Response};
 use kube::{client::Body, Client, Resource, ResourceExt};
-use prometheus::Registry;
 use std::sync::Arc;
 
 impl Document {
@@ -208,15 +207,14 @@ impl ApiServerVerifier {
 
 impl Context {
     // Create a test context with a mocked kube client, locally registered metrics and default diagnostics
-    pub fn test() -> (Arc<Self>, ApiServerVerifier, Registry) {
+    pub fn test() -> (Arc<Self>, ApiServerVerifier) {
         let (mock_service, handle) = tower_test::mock::pair::<Request<Body>, Response<Body>>();
         let mock_client = Client::new(mock_service, "default");
-        let registry = Registry::default();
         let ctx = Self {
             client: mock_client,
-            metrics: Metrics::default().register(&registry).unwrap(),
+            metrics: Arc::default(),
             diagnostics: Arc::default(),
         };
-        (Arc::new(ctx), ApiServerVerifier(handle), registry)
+        (Arc::new(ctx), ApiServerVerifier(handle))
     }
 }

src/fixtures.rs

@@ -1,15 +1,13 @@
 #![allow(unused_imports, unused_variables)]
 use actix_web::{get, middleware, web::Data, App, HttpRequest, HttpResponse, HttpServer, Responder};
 pub use controller::{self, telemetry, State};
-use prometheus::{Encoder, TextEncoder};
 
 #[get("/metrics")]
 async fn metrics(c: Data<State>, _req: HttpRequest) -> impl Responder {
     let metrics = c.metrics();
-    let encoder = TextEncoder::new();
-    let mut buffer = vec![];
-    encoder.encode(&metrics, &mut buffer).unwrap();
-    HttpResponse::Ok().body(buffer)
+    HttpResponse::Ok()
+        .content_type("application/openmetrics-text; version=1.0.0; charset=utf-8")
+        .body(metrics)
 }
 
 #[get("/health")]