Merge pull request #2457 from subspace/farmer-events-and-metrics

Farmer events and metrics

Author: nazar-pc

crates/subspace-farmer/src/bin/subspace-farmer/commands/farm.rs

@@ -1,7 +1,9 @@
 mod dsn;
+mod metrics;
 
 use crate::commands::farm::dsn::configure_dsn;
-use crate::utils::{shutdown_signal, FarmerMetrics};
+use crate::commands::farm::metrics::FarmerMetrics;
+use crate::utils::shutdown_signal;
 use anyhow::anyhow;
 use bytesize::ByteSize;
 use clap::{Parser, ValueHint};
@@ -22,6 +24,7 @@ use subspace_core_primitives::crypto::kzg::{embedded_kzg_settings, Kzg};
 use subspace_core_primitives::{PublicKey, Record, SectorIndex};
 use subspace_erasure_coding::ErasureCoding;
 use subspace_farmer::piece_cache::PieceCache;
+use subspace_farmer::single_disk_farm::farming::FarmingNotification;
 use subspace_farmer::single_disk_farm::{
     SectorPlottingDetails, SectorUpdate, SingleDiskFarm, SingleDiskFarmError, SingleDiskFarmOptions,
 };
@@ -694,27 +697,72 @@ where
                     }
                 };
 
-            // Register audit plot events
-            let farmer_metrics = farmer_metrics.clone();
-            let on_plot_audited_callback = move |audit_event: &_| {
-                farmer_metrics.observe_audit_event(audit_event);
-            };
-
             single_disk_farm
-                .on_sector_update(Arc::new(move |(_sector_index, sector_state)| {
-                    if let SectorUpdate::Plotting(SectorPlottingDetails::Finished {
-                        plotted_sector,
-                        old_plotted_sector,
-                        ..
-                    }) = sector_state
-                    {
-                        on_plotted_sector_callback(plotted_sector, old_plotted_sector);
+                .on_sector_update(Arc::new({
+                    let single_disk_farm_id = *single_disk_farm.id();
+                    let farmer_metrics = farmer_metrics.clone();
+
+                    move |(_sector_index, sector_state)| match sector_state {
+                        SectorUpdate::Plotting(SectorPlottingDetails::Starting { .. }) => {
+                            farmer_metrics.sector_plotting.inc();
+                        }
+                        SectorUpdate::Plotting(SectorPlottingDetails::Downloading) => {
+                            farmer_metrics.sector_downloading.inc();
+                        }
+                        SectorUpdate::Plotting(SectorPlottingDetails::Downloaded(time)) => {
+                            farmer_metrics
+                                .observe_sector_downloading_time(&single_disk_farm_id, time);
+                            farmer_metrics.sector_downloaded.inc();
+                        }
+                        SectorUpdate::Plotting(SectorPlottingDetails::Encoding) => {
+                            farmer_metrics.sector_encoding.inc();
+                        }
+                        SectorUpdate::Plotting(SectorPlottingDetails::Encoded(time)) => {
+                            farmer_metrics.observe_sector_encoding_time(&single_disk_farm_id, time);
+                            farmer_metrics.sector_encoded.inc();
+                        }
+                        SectorUpdate::Plotting(SectorPlottingDetails::Writing) => {
+                            farmer_metrics.sector_writing.inc();
+                        }
+                        SectorUpdate::Plotting(SectorPlottingDetails::Written(time)) => {
+                            farmer_metrics.observe_sector_writing_time(&single_disk_farm_id, time);
+                            farmer_metrics.sector_written.inc();
+                        }
+                        SectorUpdate::Plotting(SectorPlottingDetails::Finished {
+                            plotted_sector,
+                            old_plotted_sector,
+                            time,
+                        }) => {
+                            on_plotted_sector_callback(plotted_sector, old_plotted_sector);
+                            farmer_metrics.observe_sector_plotting_time(&single_disk_farm_id, time);
+                            farmer_metrics.sector_plotted.inc();
+                        }
+                        _ => {}
                     }
                 }))
                 .detach();
 
             single_disk_farm
-                .on_plot_audited(Arc::new(on_plot_audited_callback))
+                .on_farming_notification(Arc::new({
+                    let single_disk_farm_id = *single_disk_farm.id();
+                    let farmer_metrics = farmer_metrics.clone();
+
+                    move |farming_notification| match farming_notification {
+                        FarmingNotification::Auditing(auditing_details) => {
+                            farmer_metrics.observe_auditing_time(
+                                &single_disk_farm_id,
+                                &auditing_details.time,
+                            );
+                        }
+                        FarmingNotification::Proving(proving_details) => {
+                            farmer_metrics.observe_proving_time(
+                                &single_disk_farm_id,
+                                &proving_details.time,
+                                proving_details.result,
+                            );
+                        }
+                    }
+                }))
                 .detach();
 
             single_disk_farm.run()

crates/subspace-farmer/src/bin/subspace-farmer/commands/farm/metrics.rs

@@ -0,0 +1,266 @@
+use prometheus_client::metrics::counter::Counter;
+use prometheus_client::metrics::family::Family;
+use prometheus_client::metrics::histogram::{exponential_buckets, Histogram};
+use prometheus_client::registry::{Registry, Unit};
+use std::sync::atomic::AtomicU64;
+use std::time::Duration;
+use subspace_farmer::single_disk_farm::farming::ProvingResult;
+use subspace_farmer::single_disk_farm::SingleDiskFarmId;
+
+#[derive(Debug, Clone)]
+pub(super) struct FarmerMetrics {
+    auditing_time: Family<Vec<(String, String)>, Histogram>,
+    proving_time: Family<Vec<(String, String)>, Histogram>,
+    sector_downloading_time: Family<Vec<(String, String)>, Histogram>,
+    sector_encoding_time: Family<Vec<(String, String)>, Histogram>,
+    sector_writing_time: Family<Vec<(String, String)>, Histogram>,
+    sector_plotting_time: Family<Vec<(String, String)>, Histogram>,
+    pub(super) sector_downloading: Counter<u64, AtomicU64>,
+    pub(super) sector_downloaded: Counter<u64, AtomicU64>,
+    pub(super) sector_encoding: Counter<u64, AtomicU64>,
+    pub(super) sector_encoded: Counter<u64, AtomicU64>,
+    pub(super) sector_writing: Counter<u64, AtomicU64>,
+    pub(super) sector_written: Counter<u64, AtomicU64>,
+    pub(super) sector_plotting: Counter<u64, AtomicU64>,
+    pub(super) sector_plotted: Counter<u64, AtomicU64>,
+}
+
+impl FarmerMetrics {
+    pub(super) fn new(registry: &mut Registry) -> Self {
+        let sub_registry = registry.sub_registry_with_prefix("subspace_farmer");
+
+        let auditing_time = Family::<_, _>::new_with_constructor(|| {
+            Histogram::new(exponential_buckets(0.0001, 2.0, 15))
+        });
+
+        sub_registry.register_with_unit(
+            "auditing_time",
+            "Auditing time",
+            Unit::Seconds,
+            auditing_time.clone(),
+        );
+
+        let proving_time = Family::<_, _>::new_with_constructor(|| {
+            Histogram::new(exponential_buckets(0.0001, 2.0, 15))
+        });
+
+        sub_registry.register_with_unit(
+            "proving_time",
+            "Proving time",
+            Unit::Seconds,
+            proving_time.clone(),
+        );
+
+        let sector_downloading_time = Family::<_, _>::new_with_constructor(|| {
+            Histogram::new(exponential_buckets(0.0001, 2.0, 15))
+        });
+
+        sub_registry.register_with_unit(
+            "sector_downloading_time",
+            "Sector downloading time",
+            Unit::Seconds,
+            sector_downloading_time.clone(),
+        );
+
+        let sector_encoding_time = Family::<_, _>::new_with_constructor(|| {
+            Histogram::new(exponential_buckets(0.0001, 2.0, 15))
+        });
+
+        sub_registry.register_with_unit(
+            "sector_encoding_time",
+            "Sector encoding time",
+            Unit::Seconds,
+            sector_encoding_time.clone(),
+        );
+
+        let sector_writing_time = Family::<_, _>::new_with_constructor(|| {
+            Histogram::new(exponential_buckets(0.0001, 2.0, 15))
+        });
+
+        sub_registry.register_with_unit(
+            "sector_writing_time",
+            "Sector writing time",
+            Unit::Seconds,
+            sector_writing_time.clone(),
+        );
+
+        let sector_plotting_time = Family::<_, _>::new_with_constructor(|| {
+            Histogram::new(exponential_buckets(0.0001, 2.0, 15))
+        });
+
+        sub_registry.register_with_unit(
+            "sector_plotting_time",
+            "Sector plotting time",
+            Unit::Seconds,
+            sector_plotting_time.clone(),
+        );
+
+        let sector_downloading = Counter::<_, _>::default();
+
+        sub_registry.register_with_unit(
+            "sector_downloading_counter",
+            "Number of sectors being downloaded",
+            Unit::Other("sectors".to_string()),
+            sector_downloading.clone(),
+        );
+
+        let sector_downloaded = Counter::<_, _>::default();
+
+        sub_registry.register_with_unit(
+            "sector_downloaded_counter",
+            "Number of sectors being downloaded",
+            Unit::Other("sectors".to_string()),
+            sector_downloaded.clone(),
+        );
+
+        let sector_encoding = Counter::<_, _>::default();
+
+        sub_registry.register_with_unit(
+            "sector_encoding_counter",
+            "Number of sectors being downloaded",
+            Unit::Other("sectors".to_string()),
+            sector_encoding.clone(),
+        );
+
+        let sector_encoded = Counter::<_, _>::default();
+
+        sub_registry.register_with_unit(
+            "sector_encoded_counter",
+            "Number of sectors being downloaded",
+            Unit::Other("sectors".to_string()),
+            sector_encoded.clone(),
+        );
+
+        let sector_writing = Counter::<_, _>::default();
+
+        sub_registry.register_with_unit(
+            "sector_writing_counter",
+            "Number of sectors being downloaded",
+            Unit::Other("sectors".to_string()),
+            sector_writing.clone(),
+        );
+
+        let sector_written = Counter::<_, _>::default();
+
+        sub_registry.register_with_unit(
+            "sector_written_counter",
+            "Number of sectors being downloaded",
+            Unit::Other("sectors".to_string()),
+            sector_written.clone(),
+        );
+
+        let sector_plotting = Counter::<_, _>::default();
+
+        sub_registry.register_with_unit(
+            "sector_plotting_counter",
+            "Number of sectors being downloaded",
+            Unit::Other("sectors".to_string()),
+            sector_plotting.clone(),
+        );
+
+        let sector_plotted = Counter::<_, _>::default();
+
+        sub_registry.register_with_unit(
+            "sector_plotted_counter",
+            "Number of sectors being downloaded",
+            Unit::Other("sectors".to_string()),
+            sector_plotted.clone(),
+        );
+
+        Self {
+            auditing_time,
+            proving_time,
+            sector_downloading_time,
+            sector_encoding_time,
+            sector_writing_time,
+            sector_plotting_time,
+            sector_downloading,
+            sector_downloaded,
+            sector_encoding,
+            sector_encoded,
+            sector_writing,
+            sector_written,
+            sector_plotting,
+            sector_plotted,
+        }
+    }
+
+    pub(super) fn observe_auditing_time(
+        &self,
+        single_disk_farm_id: &SingleDiskFarmId,
+        time: &Duration,
+    ) {
+        self.auditing_time
+            .get_or_create(&vec![(
+                "farm_id".to_string(),
+                single_disk_farm_id.to_string(),
+            )])
+            .observe(time.as_secs_f64());
+    }
+
+    pub(super) fn observe_proving_time(
+        &self,
+        single_disk_farm_id: &SingleDiskFarmId,
+        time: &Duration,
+        result: ProvingResult,
+    ) {
+        self.proving_time
+            .get_or_create(&vec![
+                ("farm_id".to_string(), single_disk_farm_id.to_string()),
+                ("result".to_string(), result.to_string()),
+            ])
+            .observe(time.as_secs_f64());
+    }
+
+    pub(super) fn observe_sector_downloading_time(
+        &self,
+        single_disk_farm_id: &SingleDiskFarmId,
+        time: &Duration,
+    ) {
+        self.sector_downloading_time
+            .get_or_create(&vec![(
+                "farm_id".to_string(),
+                single_disk_farm_id.to_string(),
+            )])
+            .observe(time.as_secs_f64());
+    }
+
+    pub(super) fn observe_sector_encoding_time(
+        &self,
+        single_disk_farm_id: &SingleDiskFarmId,
+        time: &Duration,
+    ) {
+        self.sector_encoding_time
+            .get_or_create(&vec![(
+                "farm_id".to_string(),
+                single_disk_farm_id.to_string(),
+            )])
+            .observe(time.as_secs_f64());
+    }
+
+    pub(super) fn observe_sector_writing_time(
+        &self,
+        single_disk_farm_id: &SingleDiskFarmId,
+        time: &Duration,
+    ) {
+        self.sector_writing_time
+            .get_or_create(&vec![(
+                "farm_id".to_string(),
+                single_disk_farm_id.to_string(),
+            )])
+            .observe(time.as_secs_f64());
+    }
+
+    pub(super) fn observe_sector_plotting_time(
+        &self,
+        single_disk_farm_id: &SingleDiskFarmId,
+        time: &Duration,
+    ) {
+        self.sector_plotting_time
+            .get_or_create(&vec![(
+                "farm_id".to_string(),
+                single_disk_farm_id.to_string(),
+            )])
+            .observe(time.as_secs_f64());
+    }
+}