builder metrics

ofborg/Cargo.toml

@@ -28,3 +28,6 @@ separator = "0.4.1"
 
 async-std = "=1.5.0"
 lapin = "1.0.0-rc6"
+lazy_static = "1.4.0"
+prometheus = "0.8"
+tide = "0.8.0"

ofborg/src/bin/builder.rs

@@ -6,8 +6,7 @@ use async_std::task;
 use tracing::{info, warn};
 
 use ofborg::easyamqp::{self, ChannelExt, ConsumerExt};
-use ofborg::easylapin;
-use ofborg::{checkout, config, tasks};
+use ofborg::{checkout, config, easylapin, metrics, tasks};
 
 // FIXME: remove with rust/cargo update
 #[allow(clippy::cognitive_complexity)]
@@ -93,6 +92,8 @@ fn main() -> Result<(), Box<dyn Error>> {
         },
     )?;
 
+    metrics::spawn_server();
+
     info!("Fetching jobs from {}", &queue_name);
     task::block_on(handle);
 

ofborg/src/easylapin.rs

@@ -1,14 +1,5 @@
 use std::pin::Pin;
 
-use crate::config::RabbitMQConfig;
-use crate::easyamqp::{
-    BindQueueConfig, ChannelExt, ConsumeConfig, ConsumerExt, ExchangeConfig, ExchangeType,
-    QueueConfig,
-};
-use crate::notifyworker::{NotificationReceiver, SimpleNotifyWorker};
-use crate::ofborg;
-use crate::worker::{Action, SimpleWorker};
-
 use async_std::future::Future;
 use async_std::stream::StreamExt;
 use async_std::task;
@@ -21,6 +12,16 @@ use lapin::types::{AMQPValue, FieldTable};
 use lapin::{BasicProperties, Channel, Connection, ConnectionProperties, ExchangeKind};
 use tracing::{debug, trace};
 
+use crate::config::RabbitMQConfig;
+use crate::easyamqp::{
+    BindQueueConfig, ChannelExt, ConsumeConfig, ConsumerExt, ExchangeConfig, ExchangeType,
+    QueueConfig,
+};
+use crate::metrics;
+use crate::notifyworker::{NotificationReceiver, SimpleNotifyWorker};
+use crate::ofborg;
+use crate::worker::{Action, SimpleWorker};
+
 pub fn from_config(cfg: &RabbitMQConfig) -> Result<Connection, lapin::Error> {
     let mut props = FieldTable::default();
     props.insert(
@@ -166,6 +167,8 @@ impl<'a, W: SimpleNotifyWorker + 'a> ConsumerExt<'a, W> for NotifyChannel {
         Ok(Box::pin(async move {
             while let Some(Ok(deliver)) = consumer.next().await {
                 debug!(?deliver.delivery_tag, "consumed delivery");
+                metrics::JOBS_RECEIVED.inc();
+
                 let mut receiver = ChannelNotificationReceiver {
                     channel: &mut chan,
                     deliver: &deliver,

ofborg/src/lib.rs

@@ -31,6 +31,7 @@ pub mod ghevent;
 pub mod locks;
 pub mod maintainers;
 pub mod message;
+pub mod metrics;
 pub mod nix;
 pub mod nixenv;
 pub mod nixstats;

ofborg/src/message/buildresult.rs

@@ -11,6 +11,18 @@ pub enum BuildStatus {
     UnexpectedError { err: String },
 }
 
+impl BuildStatus {
+    pub fn as_label(&self) -> &'static str {
+        match self {
+            BuildStatus::Skipped => "skipped",
+            BuildStatus::Success => "success",
+            BuildStatus::Failure => "failure",
+            BuildStatus::TimedOut => "timeout",
+            BuildStatus::UnexpectedError { err: _ } => "unexpected",
+        }
+    }
+}
+
 impl From<BuildStatus> for String {
     fn from(status: BuildStatus) -> String {
         match status {

ofborg/src/metrics.rs

@@ -0,0 +1,77 @@
+use std::env;
+
+use async_std::task;
+use lazy_static::lazy_static;
+use prometheus::{
+    register_counter_vec, register_gauge_vec, register_int_counter, CounterVec, Encoder, GaugeVec,
+    IntCounter, TextEncoder,
+};
+use tide::http::StatusCode;
+use tide::{Body, Request, Response};
+use tracing::info;
+
+use crate::ofborg;
+
+lazy_static! {
+    pub static ref JOBS_RECEIVED: IntCounter = register_int_counter!(
+        "ofborg_jobs_received_total",
+        "Total number of jobs received."
+    )
+    .unwrap();
+    pub static ref BUILDS_RECEIVED: CounterVec = register_counter_vec!(
+        "ofborg_builds_received_total",
+        "Total number of builds received.",
+        &["system"]
+    )
+    .unwrap();
+    pub static ref BUILDS_FINISHED: CounterVec = register_counter_vec!(
+        "ofborg_builds_finished_total",
+        "Total number of builds finished.",
+        &["system", "status"]
+    )
+    .unwrap();
+    pub static ref BUILDS_ATTRIBUTES_ATTEMPTED: CounterVec = register_counter_vec!(
+        "ofborg_builds_attributes_attempted_total",
+        "Total number of attributes attempted to build.",
+        &["system"]
+    )
+    .unwrap();
+    pub static ref BUILDS_ATTRIBUTES_NOT_ATTEMPTED: CounterVec = register_counter_vec!(
+        "ofborg_builds_attributes_not_attempted_total",
+        "Total number of attributes not attempted to build.",
+        &["system"]
+    )
+    .unwrap();
+}
+
+lazy_static! {
+    static ref VERSION: GaugeVec = register_gauge_vec!(
+        "ofborg_version",
+        "Labeled OfBorg version information.",
+        &["version"]
+    )
+    .unwrap();
+}
+
+pub fn spawn_server() {
+    let port = env::var("METRICS_PORT").unwrap_or_else(|_err| String::from("9128"));
+
+    let mut server = tide::new();
+    server.at("/metrics").get(metrics);
+
+    // Initialize version metric.
+    VERSION.with_label_values(&[ofborg::VERSION]).set(1_f64);
+
+    info!("Listening on http://127.0.0.1:{}/metrics", port);
+    task::spawn(server.listen(format!("0.0.0.0:{}", port)));
+}
+
+async fn metrics(mut _req: Request<()>) -> Result<Response, tide::Error> {
+    let encoder = TextEncoder::new();
+    let metric_families = prometheus::gather();
+    let mut buffer = vec![];
+    encoder.encode(&metric_families, &mut buffer)?;
+
+    let resp = Response::new(StatusCode::Ok).body(Body::from(buffer));
+    Ok(resp)
+}

ofborg/src/tasks/build.rs

@@ -2,6 +2,7 @@ use crate::checkout;
 use crate::commentparser;
 use crate::message::buildresult::{BuildResult, BuildStatus, V1Tag};
 use crate::message::{buildjob, buildlogmsg};
+use crate::metrics;
 use crate::nix;
 use crate::notifyworker;
 use crate::worker;
@@ -280,6 +281,9 @@ impl notifyworker::SimpleNotifyWorker for BuildWorker {
     ) {
         let span = debug_span!("job", pr = ?job.pr.number);
         let _enter = span.enter();
+        metrics::BUILDS_RECEIVED
+            .with_label_values(&[&self.system])
+            .inc();
 
         let mut actions = self.actions(&job, notifier);
 
@@ -347,6 +351,14 @@ impl notifyworker::SimpleNotifyWorker for BuildWorker {
             cannot_build_attrs.join(", ")
         );
 
+        metrics::BUILDS_ATTRIBUTES_ATTEMPTED
+            .with_label_values(&[&self.system])
+            .inc_by(can_build.len() as f64);
+
+        metrics::BUILDS_ATTRIBUTES_NOT_ATTEMPTED
+            .with_label_values(&[&self.system])
+            .inc_by(cannot_build_attrs.len() as f64);
+
         actions.log_started(can_build.clone(), cannot_build_attrs.clone());
         actions.log_instantiation_errors(cannot_build);
 
@@ -375,7 +387,13 @@ impl notifyworker::SimpleNotifyWorker for BuildWorker {
             .last();
         info!("----->8-----");
 
+        let status_label = status.as_label();
         actions.build_finished(status, can_build, cannot_build_attrs);
+
+        metrics::BUILDS_FINISHED
+            .with_label_values(&[&self.system, status_label])
+            .inc();
+
         info!("Build done!");
     }
 }