Add prometheus metrics

Author: barthalion

Cargo.lock

@@ -50,3 +50,4 @@ tokio = { version = "1.18", features = ["time", "macros", "rt", "rt-multi-thread
 walkdir = "2"
 futures = "0.3"
 pin-project-lite = "0.2.17"
+prometheus-client = "0.23"

Cargo.toml

@@ -63,9 +63,10 @@ pub async fn review_check(
 
     let check = db.get_check_by_job_id(params.id).await?;
     web::block(move || {
-        let mut conn =
-            db.0.get()
-                .map_err(|err| JobError::InternalError(err.to_string()))?;
+        let mut conn = db
+            .pool
+            .get()
+            .map_err(|err| JobError::InternalError(err.to_string()))?;
         update_build_status_after_check(check.build_id, &mut conn)
     })
     .await

src/api/build.rs

@@ -0,0 +1,11 @@
+use actix_web::web::Data;
+use actix_web::{HttpResponse, Responder};
+use std::sync::Arc;
+
+use crate::metrics::Metrics;
+
+pub async fn metrics(metrics: Data<Arc<Metrics>>) -> impl Responder {
+    HttpResponse::Ok()
+        .content_type("application/openmetrics-text; version=1.0.0; charset=utf-8")
+        .body(metrics.encode())
+}

src/api/metrics.rs

@@ -1,4 +1,5 @@
 pub mod build;
+pub mod metrics;
 pub mod prune;
 pub mod repo;
 pub mod status;

src/api/mod.rs

@@ -16,6 +16,7 @@ use crate::db::Db;
 use crate::deltas::DeltaGenerator;
 use crate::jobs::JobQueue;
 use crate::logger::Logger;
+use crate::metrics::Metrics;
 use crate::tokens::TokenParser;
 use crate::Pool;
 
@@ -72,6 +73,7 @@ pub fn create_app(
     config: &Arc<Config>,
     job_queue: Addr<JobQueue>,
     delta_generator: Addr<DeltaGenerator>,
+    metrics: Arc<Metrics>,
 ) -> Server {
     let c = config.clone();
     let secret = config.secret.clone();
@@ -81,14 +83,18 @@ pub fn create_app(
         .unwrap_or_else(|| config.secret.as_ref())
         .clone();
 
-    let db = Db(pool);
+    let db = Db {
+        pool,
+        metrics: Some(metrics.clone()),
+    };
 
     let http_server = HttpServer::new(move || {
         App::new()
             .app_data(Data::new(job_queue.clone()))
             .app_data(Data::new(delta_generator.clone()))
             .app_data(Data::new((*c).clone()))
             .app_data(Data::new(db.clone()))
+            .app_data(Data::new(metrics.clone()))
             .wrap(Logger::default())
             .wrap(middleware::Compress::default())
             .service(
@@ -212,6 +218,7 @@ pub fn create_app(
                     .route(web::head().to(api::repo::handle_build_repo))
                     .to(HttpResponse::MethodNotAllowed),
             )
+            .service(web::resource("/metrics").route(web::get().to(api::metrics::metrics)))
             .service(web::resource("/status").route(web::get().to(api::status::status)))
             .service(web::resource("/status/{id}").route(web::get().to(api::status::job_status)))
     });

src/app.rs

@@ -2,16 +2,33 @@ use actix_web::*;
 use chrono::Utc;
 use diesel::prelude::*;
 use serde_json::json;
+use std::sync::Arc;
 
 use crate::errors::ApiError;
+use crate::metrics::{kind_label, repo_label, JobLabels, Metrics};
 use crate::models::*;
 use crate::schema;
 use crate::Pool;
 
 #[derive(Clone)]
-pub struct Db(pub Pool);
+pub struct Db {
+    pub pool: Pool,
+    pub metrics: Option<Arc<Metrics>>,
+}
 
 impl Db {
+    fn increment_queued_job_metric(&self, job: &Job) {
+        if let Some(metrics) = &self.metrics {
+            metrics
+                .jobs_queued
+                .get_or_create(&JobLabels {
+                    kind: kind_label(job.kind),
+                    repo: repo_label(&job.repo),
+                })
+                .inc();
+        }
+    }
+
     async fn run<Func, T>(&self, func: Func) -> Result<T, ApiError>
     where
         Func: FnOnce(
@@ -20,7 +37,7 @@ impl Db {
         Func: Send + 'static,
         T: Send + 'static,
     {
-        let p = self.0.clone();
+        let p = self.pool.clone();
         web::block(move || {
             let mut conn = p.get()?;
             func(&mut conn)
@@ -139,6 +156,10 @@ impl Db {
             Ok(job)
         })
         .await
+        .map(|job| {
+            self.increment_queued_job_metric(&job);
+            job
+        })
     }
 
     pub async fn start_publish_job(&self, build_id: i32, repo: String) -> Result<Job, ApiError> {
@@ -237,6 +258,10 @@ impl Db {
             Ok(job)
         })
         .await
+        .map(|job| {
+            self.increment_queued_job_metric(&job);
+            job
+        })
     }
 
     pub async fn start_republish_job(
@@ -264,6 +289,10 @@ impl Db {
             Ok(job)
         })
         .await
+        .map(|job| {
+            self.increment_queued_job_metric(&job);
+            job
+        })
     }
 
     pub async fn start_prune_job(&self, repo: String) -> Result<Job, ApiError> {
@@ -279,6 +308,10 @@ impl Db {
                 .map_err(ApiError::from)
         })
         .await
+        .map(|job| {
+            self.increment_queued_job_metric(&job);
+            job
+        })
     }
 
     /* Checks */

src/db.rs

@@ -11,6 +11,7 @@ use std::str;
 
 use crate::config::{Config, RepoConfig};
 use crate::errors::{JobError, JobResult};
+use crate::metrics::{kind_label, repo_label, JobLabels};
 use crate::models::{
     self, Check, CheckJob, CheckStatus, CommitJob, Job, JobKind, NewJob, RepoState,
 };
@@ -231,6 +232,17 @@ impl JobInstance for CommitJobInstance {
                             )
                             .get_results::<Job>(conn)?;
 
+                        for job in &check_jobs {
+                            executor
+                                .metrics
+                                .jobs_queued
+                                .get_or_create(&JobLabels {
+                                    kind: kind_label(job.kind),
+                                    repo: repo_label(&job.repo),
+                                })
+                                .inc();
+                        }
+
                         // Create a check row for each new check job. This row ties the job to the build and records its status.
                         let (pending_status, pending_status_msg) = CheckStatus::Pending.to_db();
                         let checks = repoconfig