split service&instance metrics, serve metrics for build-server & watcher

* split service & instance metrics into separate objects, serve them via
  separate endpoints

* add a small webserver to serve metrics for the build-server & registry
  watcher

Author: syphar

src/bin/cratesfyi.rs

@@ -14,8 +14,8 @@ use docs_rs::utils::{
     remove_crate_priority, set_crate_priority, ConfigName,
 };
 use docs_rs::{
-    start_web_server, BuildQueue, Config, Context, Index, Metrics, PackageKind, RustwideBuilder,
-    Storage,
+    start_background_metrics_webserver, start_web_server, BuildQueue, Config, Context, Index,
+    Metrics, PackageKind, RustwideBuilder, Storage,
 };
 use humantime::Duration;
 use once_cell::sync::OnceCell;
@@ -111,6 +111,8 @@ enum CommandLine {
     },
 
     StartRegistryWatcher {
+        #[arg(name = "SOCKET_ADDR", default_value = "0.0.0.0:3000")]
+        metric_server_socket_addr: String,
         /// Enable or disable the repository stats updater
         #[arg(
             long = "repository-stats-updater",
@@ -122,7 +124,10 @@ enum CommandLine {
         cdn_invalidator: Toggle,
     },
 
-    StartBuildServer,
+    StartBuildServer {
+        #[arg(name = "SOCKET_ADDR", default_value = "0.0.0.0:3000")]
+        metric_server_socket_addr: String,
+    },
 
     /// Starts the daemon
     Daemon {
@@ -154,6 +159,7 @@ impl CommandLine {
                 subcommand,
             } => subcommand.handle_args(ctx, skip_if_exists)?,
             Self::StartRegistryWatcher {
+                metric_server_socket_addr,
                 repository_stats_updater,
                 cdn_invalidator,
             } => {
@@ -164,9 +170,15 @@ impl CommandLine {
                     docs_rs::utils::daemon::start_background_cdn_invalidator(&ctx)?;
                 }
 
+                start_background_metrics_webserver(Some(&metric_server_socket_addr), &ctx)?;
+
                 docs_rs::utils::watch_registry(ctx.build_queue()?, ctx.config()?, ctx.index()?)?;
             }
-            Self::StartBuildServer => {
+            Self::StartBuildServer {
+                metric_server_socket_addr,
+            } => {
+                start_background_metrics_webserver(Some(&metric_server_socket_addr), &ctx)?;
+
                 let build_queue = ctx.build_queue()?;
                 let rustwide_builder = RustwideBuilder::init(&ctx)?;
                 queue_builder(rustwide_builder, build_queue)?;

src/build_queue.rs

@@ -204,7 +204,7 @@ impl BuildQueue {
                 to_process.name, to_process.version
             )
         });
-        self.metrics.total_builds.inc();
+        self.metrics.instance.total_builds.inc();
         if let Err(err) =
             cdn::queue_crate_invalidation(&mut transaction, &self.config, &to_process.name)
         {
@@ -225,7 +225,7 @@ impl BuildQueue {
                     .get(0);
 
                 if attempt >= self.max_attempts {
-                    self.metrics.failed_builds.inc();
+                    self.metrics.instance.failed_builds.inc();
                 }
 
                 report_error(&e);
@@ -320,7 +320,7 @@ impl BuildQueue {
                             "{}-{} added into build queue",
                             release.name, release.version
                         );
-                        self.metrics.queued_builds.inc();
+                        self.metrics.instance.queued_builds.inc();
                         crates_added += 1;
                     }
                     Err(err) => report_error(&err),
@@ -601,8 +601,8 @@ mod tests {
 
             // Ensure metrics were recorded correctly
             let metrics = env.metrics();
-            assert_eq!(metrics.total_builds.get(), 9);
-            assert_eq!(metrics.failed_builds.get(), 1);
+            assert_eq!(metrics.instance.total_builds.get(), 9);
+            assert_eq!(metrics.instance.failed_builds.get(), 1);
 
             // no invalidations were run since we don't have a distribution id configured
             assert!(cdn::queued_or_active_crate_invalidations(&mut *env.db().conn())?.is_empty());

src/cdn.rs

@@ -363,6 +363,7 @@ pub(crate) fn handle_queued_invalidation_requests(
         if let Ok(duration) = (now - row.get::<_, DateTime<Utc>>(0)).to_std() {
             // This can only fail when the duration is negative, which can't happen anyways
             metrics
+                .instance
                 .cdn_invalidation_time
                 .with_label_values(&[distribution_id])
                 .observe(duration_to_seconds(duration));
@@ -400,6 +401,7 @@ pub(crate) fn handle_queued_invalidation_requests(
         if let Ok(duration) = (now - row.get::<_, DateTime<Utc>>("queued")).to_std() {
             // This can only fail when the duration is negative, which can't happen anyways
             metrics
+                .instance
                 .cdn_queue_time
                 .with_label_values(&[distribution_id])
                 .observe(duration_to_seconds(duration));

src/db/pool.rs

@@ -77,7 +77,7 @@ impl Pool {
         match self.with_pool(|p| p.get()) {
             Ok(conn) => Ok(conn),
             Err(err) => {
-                self.metrics.failed_db_connections.inc();
+                self.metrics.instance.failed_db_connections.inc();
                 Err(PoolError::ClientError(err))
             }
         }

src/docbuilder/rustwide_builder.rs

@@ -484,11 +484,11 @@ impl RustwideBuilder {
 
                     let has_examples = build.host_source_dir().join("examples").is_dir();
                     if res.result.successful {
-                        self.metrics.successful_builds.inc();
+                        self.metrics.instance.successful_builds.inc();
                     } else if res.cargo_metadata.root().is_library() {
-                        self.metrics.failed_builds.inc();
+                        self.metrics.instance.failed_builds.inc();
                     } else {
-                        self.metrics.non_library_builds.inc();
+                        self.metrics.instance.non_library_builds.inc();
                     }
 
                     let release_data = if !is_local {

src/lib.rs

@@ -10,7 +10,7 @@ pub use self::docbuilder::RustwideBuilder;
 pub use self::index::Index;
 pub use self::metrics::Metrics;
 pub use self::storage::Storage;
-pub use self::web::start_web_server;
+pub use self::web::{start_background_metrics_webserver, start_web_server};
 
 mod build_queue;
 pub mod cdn;

src/metrics/macros.rs

@@ -2,7 +2,6 @@ pub(super) trait MetricFromOpts: Sized {
     fn from_opts(opts: prometheus::Opts) -> Result<Self, prometheus::Error>;
 }
 
-#[macro_export]
 macro_rules! metrics {
     (
         $vis:vis struct $name:ident {
@@ -82,7 +81,6 @@ macro_rules! metrics {
     };
 }
 
-#[macro_export]
 macro_rules! load_metric_type {
     ($name:ident as single) => {
         use prometheus::$name;

src/metrics/mod.rs

@@ -35,22 +35,23 @@ pub const CDN_INVALIDATION_HISTOGRAM_BUCKETS: &[f64; 11] = &[
     24000.0, // 240
 ];
 
-metrics! {
-    pub struct Metrics {
-        /// Number of crates in the build queue
-        queued_crates_count: IntGauge,
-        /// Number of crates in the build queue that have a positive priority
-        prioritized_crates_count: IntGauge,
-        /// Number of crates that failed to build
-        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"],
-
-        /// queued CDN invalidations
-        queued_cdn_invalidations_by_distribution: IntGaugeVec["distribution"],
+#[derive(Debug)]
+pub struct Metrics {
+    pub instance: InstanceMetrics,
+    pub service: ServiceMetrics,
+}
+
+impl Metrics {
+    pub fn new() -> Result<Self, prometheus::Error> {
+        Ok(Self {
+            instance: InstanceMetrics::new()?,
+            service: ServiceMetrics::new()?,
+        })
+    }
+}
 
+metrics! {
+    pub struct InstanceMetrics {
         /// The number of idle database connections
         idle_db_connections: IntGauge,
         /// The number of used database connections
@@ -143,7 +144,7 @@ impl RecentlyAccessedReleases {
             .insert((version, TargetAtom::from(target)), now);
     }
 
-    pub(crate) fn gather(&self, metrics: &Metrics) {
+    pub(crate) fn gather(&self, metrics: &InstanceMetrics) {
         fn inner<K: std::hash::Hash + Eq>(map: &DashMap<K, Instant>, metric: &IntGaugeVec) {
             let mut hour_count = 0;
             let mut half_hour_count = 0;
@@ -182,18 +183,112 @@ impl RecentlyAccessedReleases {
     }
 }
 
-impl Metrics {
+impl InstanceMetrics {
+    pub(crate) fn gather(&self, pool: &Pool) -> Result<Vec<MetricFamily>, Error> {
+        self.idle_db_connections.set(pool.idle_connections() as i64);
+        self.used_db_connections.set(pool.used_connections() as i64);
+        self.max_db_connections.set(pool.max_size() as i64);
+
+        self.recently_accessed_releases.gather(self);
+        self.gather_system_performance();
+        Ok(self.registry.gather())
+    }
+
+    #[cfg(not(target_os = "linux"))]
+    fn gather_system_performance(&self) {}
+
+    #[cfg(target_os = "linux")]
+    fn gather_system_performance(&self) {
+        use procfs::process::Process;
+
+        let process = Process::myself().unwrap();
+        self.open_file_descriptors
+            .set(process.fd_count().unwrap() as i64);
+        self.running_threads
+            .set(process.stat().unwrap().num_threads);
+    }
+}
+
+fn metric_from_opts<T: MetricFromOpts + Clone + prometheus::core::Collector + 'static>(
+    registry: &prometheus::Registry,
+    metric: &str,
+    help: &str,
+    variable_label: Option<&str>,
+) -> Result<T, prometheus::Error> {
+    let mut opts = prometheus::Opts::new(metric, help).namespace("docsrs");
+
+    if let Some(label) = variable_label {
+        opts = opts.variable_label(label);
+    }
+
+    let metric = T::from_opts(opts)?;
+    registry.register(Box::new(metric.clone()))?;
+    Ok(metric)
+}
+
+#[derive(Debug)]
+pub struct ServiceMetrics {
+    pub queued_crates_count: IntGauge,
+    pub prioritized_crates_count: IntGauge,
+    pub failed_crates_count: IntGauge,
+    pub queue_is_locked: IntGauge,
+    pub queued_crates_count_by_priority: IntGaugeVec,
+    pub queued_cdn_invalidations_by_distribution: IntGaugeVec,
+
+    registry: prometheus::Registry,
+}
+
+impl ServiceMetrics {
+    pub fn new() -> Result<Self, prometheus::Error> {
+        let registry = prometheus::Registry::new();
+        Ok(Self {
+            registry: registry.clone(),
+            queued_crates_count: metric_from_opts(
+                &registry,
+                "queued_crates_count",
+                "Number of crates in the build queue",
+                None,
+            )?,
+            prioritized_crates_count: metric_from_opts(
+                &registry,
+                "prioritized_crates_count",
+                "Number of crates in the build queue that have a positive priority",
+                None,
+            )?,
+            failed_crates_count: metric_from_opts(
+                &registry,
+                "failed_crates_count",
+                "Number of crates that failed to build",
+                None,
+            )?,
+            queue_is_locked: metric_from_opts(
+                &registry,
+                "queue_is_locked",
+                "Whether the build queue is locked",
+                None,
+            )?,
+            queued_crates_count_by_priority: metric_from_opts(
+                &registry,
+                "queued_crates_count_by_priority",
+                "queued crates by priority",
+                Some("priority"),
+            )?,
+            queued_cdn_invalidations_by_distribution: metric_from_opts(
+                &registry,
+                "queued_cdn_invalidations_by_distribution",
+                "queued CDN invalidations",
+                Some("distribution"),
+            )?,
+        })
+    }
+
     pub(crate) fn gather(
         &self,
         pool: &Pool,
         queue: &BuildQueue,
         config: &Config,
     ) -> Result<Vec<MetricFamily>, Error> {
-        self.idle_db_connections.set(pool.idle_connections() as i64);
-        self.used_db_connections.set(pool.used_connections() as i64);
-        self.max_db_connections.set(pool.max_size() as i64);
         self.queue_is_locked.set(queue.is_locked()? as i64);
-
         self.queued_crates_count.set(queue.pending_count()? as i64);
         self.prioritized_crates_count
             .set(queue.prioritized_count()? as i64);
@@ -216,23 +311,6 @@ impl Metrics {
         }
 
         self.failed_crates_count.set(queue.failed_count()? as i64);
-
-        self.recently_accessed_releases.gather(self);
-        self.gather_system_performance();
         Ok(self.registry.gather())
     }
-
-    #[cfg(not(target_os = "linux"))]
-    fn gather_system_performance(&self) {}
-
-    #[cfg(target_os = "linux")]
-    fn gather_system_performance(&self) {
-        use procfs::process::Process;
-
-        let process = Process::myself().unwrap();
-        self.open_file_descriptors
-            .set(process.fd_count().unwrap() as i64);
-        self.running_threads
-            .set(process.stat().unwrap().num_threads);
-    }
 }

src/storage/database.rs

@@ -160,7 +160,7 @@ impl<'a> StorageTransaction for DatabaseStorageTransaction<'a> {
                     SET mime = EXCLUDED.mime, content = EXCLUDED.content, compression = EXCLUDED.compression",
                 &[&blob.path, &blob.mime, &blob.content, &compression],
             )?;
-            self.metrics.uploaded_files_total.inc();
+            self.metrics.instance.uploaded_files_total.inc();
         }
         Ok(())
     }

src/storage/mod.rs

@@ -810,7 +810,10 @@ mod backend_tests {
             assert_eq!(blob.mime, actual.mime);
         }
 
-        assert_eq!(NAMES.len(), metrics.uploaded_files_total.get() as usize);
+        assert_eq!(
+            NAMES.len(),
+            metrics.instance.uploaded_files_total.get() as usize
+        );
 
         Ok(())
     }
@@ -886,7 +889,7 @@ mod backend_tests {
         assert_eq!(file.mime, "text/rust");
         assert_eq!(file.path, "folder/test.zip/src/main.rs");
 
-        assert_eq!(2, metrics.uploaded_files_total.get());
+        assert_eq!(2, metrics.instance.uploaded_files_total.get());
 
         Ok(())
     }
@@ -933,7 +936,7 @@ mod backend_tests {
         expected_algs.insert(CompressionAlgorithm::default());
         assert_eq!(algs, expected_algs);
 
-        assert_eq!(2, metrics.uploaded_files_total.get());
+        assert_eq!(2, metrics.instance.uploaded_files_total.get());
 
         Ok(())
     }