new metric "queued crates by priority"

to be able to query by priority in grafana

Author: syphar

src/build_queue.rs

@@ -10,6 +10,7 @@ use anyhow::Context;
 use crates_index_diff::Change;
 use tracing::{debug, info, warn};
 
+use std::collections::HashMap;
 use std::sync::Arc;
 use std::thread;
 use std::time::Duration;
@@ -93,19 +94,32 @@ impl BuildQueue {
     }
 
     pub(crate) fn pending_count(&self) -> Result<usize> {
-        let res = self.db.get()?.query(
-            "SELECT COUNT(*) FROM queue WHERE attempt < $1;",
-            &[&self.max_attempts],
-        )?;
-        Ok(res[0].get::<_, i64>(0) as usize)
+        Ok(self.pending_count_by_priority()?.values().sum::<usize>())
     }
 
     pub(crate) fn prioritized_count(&self) -> Result<usize> {
+        Ok(self
+            .pending_count_by_priority()?
+            .iter()
+            .filter(|(&priority, _)| priority <= 0)
+            .map(|(_, count)| count)
+            .sum::<usize>())
+    }
+
+    pub(crate) fn pending_count_by_priority(&self) -> Result<HashMap<i32, usize>> {
         let res = self.db.get()?.query(
-            "SELECT COUNT(*) FROM queue WHERE attempt < $1 AND priority <= 0;",
+            "SELECT 
+                priority, 
+                COUNT(*) 
+            FROM queue 
+            WHERE attempt < $1
+            GROUP BY priority",
             &[&self.max_attempts],
         )?;
-        Ok(res[0].get::<_, i64>(0) as usize)
+        Ok(res
+            .iter()
+            .map(|row| (row.get::<_, i32>(0), row.get::<_, i64>(1) as usize))
+            .collect())
     }
 
     pub(crate) fn failed_count(&self) -> Result<usize> {
@@ -656,6 +670,31 @@ mod tests {
         });
     }
 
+    #[test]
+    fn test_count_by_priority() {
+        crate::test::wrapper(|env| {
+            let queue = env.build_queue();
+
+            assert!(queue.pending_count_by_priority()?.is_empty());
+
+            queue.add_crate("one", "1.0.0", 1, None)?;
+            queue.add_crate("two", "2.0.0", 2, None)?;
+            queue.add_crate("two_more", "2.0.0", 2, None)?;
+
+            assert_eq!(
+                queue.pending_count_by_priority()?,
+                HashMap::from_iter(vec![(1, 1), (2, 2)])
+            );
+
+            while queue.pending_count()? > 0 {
+                queue.process_next_crate(|_| Ok(()))?;
+            }
+            assert!(queue.pending_count_by_priority()?.is_empty());
+
+            Ok(())
+        });
+    }
+
     #[test]
     fn test_failed_count() {
         const MAX_ATTEMPTS: u16 = 3;

src/metrics/mod.rs

@@ -26,6 +26,8 @@ metrics! {
         failed_crates_count: IntGauge,
         /// Whether the build queue is locked
         queue_is_locked: IntGauge,
+        /// queued crates by priority
+        queued_crates_count_by_priority: IntGaugeVec["priority"],
 
         /// The number of idle database connections
         idle_db_connections: IntGauge,
@@ -163,6 +165,15 @@ impl Metrics {
         self.queued_crates_count.set(queue.pending_count()? as i64);
         self.prioritized_crates_count
             .set(queue.prioritized_count()? as i64);
+
+        let queue_pending_count = queue.pending_count_by_priority()?;
+
+        for (priority, count) in queue_pending_count.iter() {
+            self.queued_crates_count_by_priority
+                .with_label_values(&[&priority.to_string()])
+                .set(*count as i64);
+        }
+
         self.failed_crates_count.set(queue.failed_count()? as i64);
 
         self.recently_accessed_releases.gather(self);