Record aggregator metrics in event store

docs/runbook/README.md

@@ -23,3 +23,4 @@ This page gathers the available guides to operate a Mithril network.
 | **Publish packages to npm manually**         | [manual-publish-npm](./manual-publish-npm/README.md)                                     | Manually publish packages to npm registry.                                |
 | **Client multi-platform test**               | [test-client-multiplatform](./test-client-multiplatform/README.md)                       | Run multi-platform client CLI binaries, docker and WASM package tests.    |
 | **Maintain the networks configuration file** | [maintain-networks-configuration-file](./maintain-networks-configuration-file/README.md) | Maintain the `networks.json` file                                         |
+| **Aggregator metrics**                       | [aggregator-metrics](./aggregator-metrics/README.md)                                     | Display aggregator daily metrics.                                         |

docs/runbook/aggregator-metrics/README.md

@@ -0,0 +1,24 @@
+# Aggregator metrics
+
+A view `metrics_per_day` is available to calculate the value of a metric over a day.
+
+The following request displays the sum of counter for each metric on the specify day.
+Format of `DAY` variable should be `YYYY-MM-DD` (ie: `2024-10-28`).
+
+```sh
+$> sqlite3 -table -batch \
+       $DATA_STORES_DIRECTORY/monitoring.sqlite3 \
+       `select date, counter_name, value from metrics_per_day where date='$DAY';`
+```
+
+The result looks like:
+
+```
++------------+-------------------------------------------------------------+--------+
+|    date    |                        counter_name                         | value  |
++------------+-------------------------------------------------------------+--------+
+| 2024-10-29 | mithril_aggregator_certificate_total_produced_since_startup | 135532 |
+| 2024-10-29 | mithril_aggregator_runtime_cycle_success_since_startup      | 563246 |
+| 2024-10-29 | mithril_aggregator_runtime_cycle_total_since_startup        | 237513 |
++------------+-------------------------------------------------------------+--------+
+```

docs/website/root/manual/developer-docs/nodes/mithril-aggregator.md

@@ -456,6 +456,7 @@ Here is a list of the available parameters:
 | `enable_metrics_server`                                          | `--enable-metrics-server`                                          |          -           | `ENABLE_METRICS_SERVER`                                                                                   | Enable metrics HTTP server (Prometheus endpoint on /metrics)                                                                                          | `false`                                       | -                                                                             |                        -                        |
 | `metrics_server_ip`                                              | `--metrics-server-ip`                                              |          -           | `METRICS_SERVER_IP`                                                                                       | Metrics HTTP server IP                                                                                                                                | `0.0.0.0`                                     | -                                                                             |                        -                        |
 | `metrics_server_port`                                            | `--metrics-server-port`                                            |          -           | `METRICS_SERVER_PORT`                                                                                     | Metrics HTTP server listening port                                                                                                                    | `9090`                                        | -                                                                             |                        -                        |
+| `persist_usage_report_interval_in_seconds`                       |                                                                    |          -           | `PERSIST_USAGE_REPORT_INTERVAL_IN_SECONDS`                                                                | Duration in seconds between two recording of usage metrics                                                                                            | `10`                                          | `5`                                                                           |                        -                        |
 
 `genesis bootstrap` command:
 

mithril-aggregator/src/commands/serve_command.rs

@@ -221,6 +221,18 @@ impl ServeCommand {
                 }
             }
         }
+
+        let mut usage_reporter = dependencies_builder
+            .create_usage_reporter()
+            .await
+            .with_context(|| "Dependencies Builder can not create usage reporter")?;
+        join_set.spawn(async move {
+            let interval_duration =
+                Duration::from_secs(config.persist_usage_report_interval_in_seconds);
+            usage_reporter.run_forever(interval_duration).await;
+            Ok(())
+        });
+
         let metrics_service = dependencies_builder
             .get_metrics_service()
             .await

mithril-aggregator/src/configuration.rs

@@ -183,6 +183,9 @@ pub struct Configuration {
 
     /// Metrics HTTP Server listening port.
     pub metrics_server_port: u16,
+
+    /// Time interval at which usage metrics are persisted in event database (in seconds).
+    pub persist_usage_report_interval_in_seconds: u64,
 }
 
 /// Uploader needed to copy the snapshot once computed.
@@ -266,6 +269,7 @@ impl Configuration {
             enable_metrics_server: true,
             metrics_server_ip: "0.0.0.0".to_string(),
             metrics_server_port: 9090,
+            persist_usage_report_interval_in_seconds: 10,
         }
     }
 
@@ -395,6 +399,9 @@ pub struct DefaultConfiguration {
 
     /// Metrics HTTP server listening port.
     pub metrics_server_port: u16,
+
+    /// Time interval at which metrics are persisted in event database (in seconds).
+    pub persist_usage_report_interval_in_seconds: u64,
 }
 
 impl Default for DefaultConfiguration {
@@ -426,6 +433,7 @@ impl Default for DefaultConfiguration {
             enable_metrics_server: "false".to_string(),
             metrics_server_ip: "0.0.0.0".to_string(),
             metrics_server_port: 9090,
+            persist_usage_report_interval_in_seconds: 10,
         }
     }
 }
@@ -465,7 +473,6 @@ impl Source for DefaultConfiguration {
         }
         let mut result = Map::new();
         let myself = self.clone();
-
         insert_default_configuration!(result, myself.environment);
         insert_default_configuration!(result, myself.server_ip);
         insert_default_configuration!(result, myself.server_port);
@@ -496,6 +503,7 @@ impl Source for DefaultConfiguration {
         insert_default_configuration!(result, myself.enable_metrics_server);
         insert_default_configuration!(result, myself.metrics_server_ip);
         insert_default_configuration!(result, myself.metrics_server_port);
+        insert_default_configuration!(result, myself.persist_usage_report_interval_in_seconds);
         result.insert(
             "cardano_transactions_signing_config".to_string(),
             into_value(HashMap::from([
@@ -513,7 +521,6 @@ impl Source for DefaultConfiguration {
                 ),
             ])),
         );
-
         Ok(result)
     }
 }

mithril-aggregator/src/dependency_injection/builder.rs

@@ -69,7 +69,7 @@ use crate::{
         CardanoTransactionsImporter, CertifierService, MessageService, MithrilCertifierService,
         MithrilEpochService, MithrilMessageService, MithrilProverService,
         MithrilSignedEntityService, MithrilStakeDistributionService, ProverService,
-        SignedEntityService, StakeDistributionService, UpkeepService,
+        SignedEntityService, StakeDistributionService, UpkeepService, UsageReporter,
     },
     tools::{CExplorerSignerRetriever, GcpFileUploader, GenesisToolsDependency, SignersImporter},
     AggregatorConfig, AggregatorRunner, AggregatorRuntime, CertificatePendingStore,
@@ -1392,6 +1392,17 @@ impl DependenciesBuilder {
         Ok(self.metrics_service.as_ref().cloned().unwrap())
     }
 
+    /// Create a [UsageReporter] instance.
+    pub async fn create_usage_reporter(&mut self) -> Result<UsageReporter> {
+        let usage_reporter = UsageReporter::new(
+            self.get_event_transmitter().await?,
+            self.get_metrics_service().await?,
+            self.root_logger(),
+        );
+
+        Ok(usage_reporter)
+    }
+
     /// Return an unconfigured [DependencyContainer]
     pub async fn build_dependency_container(&mut self) -> Result<DependencyContainer> {
         let dependency_manager = DependencyContainer {

mithril-aggregator/src/event_store/database/migration.rs

@@ -20,5 +20,18 @@ create table if not exists event (
 );
             "#,
         ),
+        SqlMigration::new(
+            2,
+            r#"
+create view if not exists metrics_per_day as select metric_date as date, action as counter_name, sum(counter) value from 
+    (
+        select action, json_extract(content, '$.content.counter') counter, date(json_extract(content, '$.content.date')) metric_date 
+        from event 
+        where source='Metrics'
+    ) 
+group by action, date;
+create index metric_date_index on event(date(json_extract(content, '$.content.date')));
+            "#,
+        ),
     ]
 }

mithril-aggregator/src/event_store/database/repository.rs

@@ -58,4 +58,93 @@ mod tests {
         let _event = persister.persist(message)?;
         Ok(())
     }
+
+    mod metrics_per_day_view {
+        use std::time::Duration;
+
+        use crate::event_store::database::test_helper::event_store_db_connection;
+        use chrono::{DateTime, Utc};
+        use serde::{Deserialize, Serialize};
+
+        use mithril_common::StdResult;
+
+        use sqlite::ConnectionThreadSafe;
+
+        use super::*;
+        #[derive(Serialize, Deserialize)]
+        struct MetricMessage {
+            counter: i64,
+            duration: Duration,
+            date: DateTime<Utc>,
+        }
+
+        fn get_all_metrics(
+            connection: Arc<ConnectionThreadSafe>,
+        ) -> StdResult<Vec<(String, String, i64)>> {
+            let query = "select date, counter_name, value from metrics_per_day";
+            let mut statement = connection.prepare(query)?;
+            let mut result = Vec::new();
+            while let Ok(sqlite::State::Row) = statement.next() {
+                result.push((
+                    statement.read::<String, _>("date")?,
+                    statement.read::<String, _>("counter_name")?,
+                    statement.read::<i64, _>("value")?,
+                ));
+            }
+
+            Ok(result)
+        }
+
+        /// Insert a metric event in the database.
+        /// date format is "%Y-%m-%d %H:%M:%S %z", example: "2015-09-05 23:56:04 +0000"
+        fn insert_metric_event(
+            persister: &EventPersister,
+            date: &str,
+            metric_name: &str,
+            value: i64,
+        ) {
+            let metric_date =
+                DateTime::parse_from_str(&format!("{date} +0000"), "%Y-%m-%d %H:%M:%S %z").unwrap();
+
+            let message = EventMessage::new(
+                "Metrics",
+                metric_name,
+                serde_json::json!(MetricMessage {
+                    counter: value,
+                    duration: Duration::from_secs(3),
+                    date: metric_date.into(),
+                }),
+            );
+
+            let _event = persister.persist(message).unwrap();
+        }
+
+        #[test]
+        fn retrieved_inserted_event() {
+            let connection = Arc::new(event_store_db_connection().unwrap());
+            let persister = EventPersister::new(connection.clone());
+            insert_metric_event(&persister, "2024-10-29 23:56:04", "metric_1", 15);
+
+            let result = get_all_metrics(connection).unwrap();
+
+            assert!(result.contains(&("2024-10-29".to_string(), "metric_1".to_string(), 15)));
+        }
+
+        #[test]
+        fn sum_metric_per_day() {
+            let connection = Arc::new(event_store_db_connection().unwrap());
+            let persister = EventPersister::new(connection.clone());
+            insert_metric_event(&persister, "2024-10-29 21:00:00", "metric_1", 15);
+            insert_metric_event(&persister, "2024-10-29 22:00:00", "metric_1", 60);
+            insert_metric_event(&persister, "2024-10-29 23:00:00", "metric_2", 100);
+            insert_metric_event(&persister, "2024-10-30 17:00:00", "metric_1", 12);
+            insert_metric_event(&persister, "2024-10-30 18:00:00", "metric_1", 4);
+
+            let result = get_all_metrics(connection).unwrap();
+
+            assert!(result.contains(&("2024-10-29".to_string(), "metric_1".to_string(), 75)));
+            assert!(result.contains(&("2024-10-29".to_string(), "metric_2".to_string(), 100)));
+            assert!(result.contains(&("2024-10-30".to_string(), "metric_1".to_string(), 16)));
+        }
+    }
 }

mithril-aggregator/src/event_store/event.rs

@@ -3,7 +3,7 @@ use chrono::{DateTime, Utc};
 use std::collections::HashMap;
 
 /// Event that is sent from a thread to be persisted.
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq)]
 pub struct EventMessage {
     /// The source of the message shall be composed of the name of the thread
     /// that sends the message, the name of the method can be added to it,

mithril-aggregator/src/metrics/service.rs

@@ -1,3 +1,5 @@
+use std::collections::HashMap;
+
 use mithril_metric::{build_metrics_service, MetricsServiceExporter};
 
 use mithril_metric::metric::{MetricCollector, MetricCounter};
@@ -74,3 +76,66 @@ build_metrics_service!(
     )
 
 );
+
+impl MetricsService {
+    /// Export metrics in map.
+    pub fn export_metrics_map(&self) -> HashMap<String, u32> {
+        self.registry
+            .gather()
+            .iter()
+            .map(|metric| {
+                (
+                    metric.get_name().to_string(),
+                    metric
+                        .get_metric()
+                        .iter()
+                        .map(|m| m.get_counter().get_value() as u32)
+                        .sum(),
+                )
+            })
+            .collect()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::test_tools::TestLogger;
+
+    use super::*;
+
+    #[test]
+    fn should_export_counter_metrics_in_a_map() {
+        let metrics_service = MetricsService::new(TestLogger::stdout()).unwrap();
+        let metric_a = metrics_service.get_runtime_cycle_total_since_startup();
+        let metric_b = metrics_service.get_certificate_total_produced_since_startup();
+        metric_a.increment_by(5);
+        metric_b.increment_by(12);
+
+        let export = metrics_service.export_metrics_map();
+        assert_eq!(5, export[&metric_a.name()]);
+        assert_eq!(12, export[&metric_b.name()]);
+    }
+
+    #[test]
+    fn should_export_several_times_and_counter_return_values_since_start() {
+        let metrics_service = MetricsService::new(TestLogger::stdout()).unwrap();
+        let metric_a = metrics_service.get_runtime_cycle_total_since_startup();
+        metric_a.increment_by(5);
+
+        let export = metrics_service.export_metrics_map();
+        assert_eq!(5, export[&metric_a.name()]);
+
+        metric_a.increment();
+        let export = metrics_service.export_metrics_map();
+        assert_eq!(6, export[&metric_a.name()]);
+    }
+
+    #[test]
+    fn should_export_counter_even_the_value_is_0() {
+        let metrics_service = MetricsService::new(TestLogger::stdout()).unwrap();
+        let metric_a = metrics_service.get_runtime_cycle_total_since_startup();
+
+        let export = metrics_service.export_metrics_map();
+        assert_eq!(0, export[&metric_a.name()]);
+    }
+}

mithril-aggregator/src/services/mod.rs

@@ -18,6 +18,7 @@ mod signable_builder;
 mod signed_entity;
 mod stake_distribution;
 mod upkeep;
+mod usage_reporter;
 
 pub use cardano_transactions_importer::*;
 pub use certifier::*;
@@ -28,3 +29,4 @@ pub use signable_builder::*;
 pub use signed_entity::*;
 pub use stake_distribution::*;
 pub use upkeep::*;
+pub use usage_reporter::*;