Merge pull request #2059 from input-output-hk/sfauvel/2023/record_aggregator_metrics_in_event_store

sfauvel · web-flow · commit 678e6e9c25d4 · 2024-10-30T18:41:21.000+01:00
Record aggregator metrics in event store
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/docs/runbook/README.md b/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.                                         |
diff --git a/docs/runbook/aggregator-metrics/README.md b/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 |
++------------+-------------------------------------------------------------+--------+
+```
diff --git a/docs/website/root/manual/developer-docs/nodes/mithril-aggregator.md b/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:
 
diff --git a/mithril-aggregator/Cargo.toml b/mithril-aggregator/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "mithril-aggregator"
-version = "0.5.94"
+version = "0.5.95"
 description = "A Mithril Aggregator server"
 authors = { workspace = true }
 edition = { workspace = true }
diff --git a/mithril-aggregator/src/commands/serve_command.rs b/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
diff --git a/mithril-aggregator/src/configuration.rs b/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)
     }
 }
diff --git a/mithril-aggregator/src/dependency_injection/builder.rs b/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 {
diff --git a/mithril-aggregator/src/event_store/database/migration.rs b/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')));
+            "#,
+        ),
     ]
 }
diff --git a/mithril-aggregator/src/event_store/database/repository.rs b/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)));
+        }
+    }
 }
diff --git a/mithril-aggregator/src/event_store/event.rs b/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,
diff --git a/mithril-aggregator/src/metrics/service.rs b/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,80 @@ build_metrics_service!(
     )
 
 );
+
+impl MetricsService {
+    /// Export metrics in map.
+    // `get metric` returns a list of Metrics for CounterVec purposes for example.
+    // We therefore add up the values ​​even though we will always only have one value with our Counter type metrics.
+    pub fn export_metrics_map(&self) -> HashMap<String, u32> {
+        self.registry
+            .gather()
+            .iter()
+            .map(|metric_family| {
+                (
+                    metric_family.get_name().to_string(),
+                    metric_family
+                        .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()]);
+    }
+
+    #[test]
+    fn metric_service_should_only_contain_counters_as_export_metrics_map_does_not_yet_support_other_types(
+    ) {
+        let metrics_service = MetricsService::new(TestLogger::stdout()).unwrap();
+
+        for metric_family in metrics_service.registry.gather() {
+            for metric in metric_family.get_metric() {
+                assert!(metric.has_counter());
+            }
+        }
+    }
+}
diff --git a/mithril-aggregator/src/services/mod.rs b/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::*;
diff --git a/mithril-aggregator/src/services/usage_reporter.rs b/mithril-aggregator/src/services/usage_reporter.rs
diff --git a/mithril-test-lab/mithril-end-to-end/Cargo.toml b/mithril-test-lab/mithril-end-to-end/Cargo.toml
diff --git a/mithril-test-lab/mithril-end-to-end/src/mithril/aggregator.rs b/mithril-test-lab/mithril-end-to-end/src/mithril/aggregator.rs

Original file line number	Diff line number	Diff line change
`@@ -221,6 +221,18 @@ impl ServeCommand {`
`221`	`221`	`}`
`222`	`222`	`}`
`223`	`223`	`}`
	`224`	`+`
	`225`	`+ let mut usage_reporter = dependencies_builder`
	`226`	`+ .create_usage_reporter()`
	`227`	`+ .await`
	`228`	`+ .with_context(\|\| "Dependencies Builder can not create usage reporter")?;`
	`229`	`+ join_set.spawn(async move {`
	`230`	`+ let interval_duration =`
	`231`	`+ Duration::from_secs(config.persist_usage_report_interval_in_seconds);`
	`232`	`+ usage_reporter.run_forever(interval_duration).await;`
	`233`	`+ Ok(())`
	`234`	`+ });`
	`235`	`+`
`224`	`236`	`let metrics_service = dependencies_builder`
`225`	`237`	`.get_metrics_service()`
`226`	`238`	`.await`
Original file line number	Diff line number	Diff line change
`@@ -183,6 +183,9 @@ pub struct Configuration {`
`183`	`183`
`184`	`184`	`/// Metrics HTTP Server listening port.`
`185`	`185`	`pub metrics_server_port: u16,`
	`186`	`+`
	`187`	`+ /// Time interval at which usage metrics are persisted in event database (in seconds).`
	`188`	`+ pub persist_usage_report_interval_in_seconds: u64,`
`186`	`189`	`}`
`187`	`190`
`188`	`191`	`/// Uploader needed to copy the snapshot once computed.`
`@@ -266,6 +269,7 @@ impl Configuration {`
`266`	`269`	`enable_metrics_server: true,`
`267`	`270`	`metrics_server_ip: "0.0.0.0".to_string(),`
`268`	`271`	`metrics_server_port: 9090,`
	`272`	`+ persist_usage_report_interval_in_seconds: 10,`
`269`	`273`	`}`
`270`	`274`	`}`
`271`	`275`
`@@ -395,6 +399,9 @@ pub struct DefaultConfiguration {`
`395`	`399`
`396`	`400`	`/// Metrics HTTP server listening port.`
`397`	`401`	`pub metrics_server_port: u16,`
	`402`	`+`
	`403`	`+ /// Time interval at which metrics are persisted in event database (in seconds).`
	`404`	`+ pub persist_usage_report_interval_in_seconds: u64,`
`398`	`405`	`}`
`399`	`406`
`400`	`407`	`impl Default for DefaultConfiguration {`
`@@ -426,6 +433,7 @@ impl Default for DefaultConfiguration {`
`426`	`433`	`enable_metrics_server: "false".to_string(),`
`427`	`434`	`metrics_server_ip: "0.0.0.0".to_string(),`
`428`	`435`	`metrics_server_port: 9090,`
	`436`	`+ persist_usage_report_interval_in_seconds: 10,`
`429`	`437`	`}`
`430`	`438`	`}`
`431`	`439`	`}`
`@@ -465,7 +473,6 @@ impl Source for DefaultConfiguration {`
`465`	`473`	`}`
`466`	`474`	`let mut result = Map::new();`
`467`	`475`	`let myself = self.clone();`
`468`		`-`
`469`	`476`	`insert_default_configuration!(result, myself.environment);`
`470`	`477`	`insert_default_configuration!(result, myself.server_ip);`
`471`	`478`	`insert_default_configuration!(result, myself.server_port);`
`@@ -496,6 +503,7 @@ impl Source for DefaultConfiguration {`
`496`	`503`	`insert_default_configuration!(result, myself.enable_metrics_server);`
`497`	`504`	`insert_default_configuration!(result, myself.metrics_server_ip);`
`498`	`505`	`insert_default_configuration!(result, myself.metrics_server_port);`
	`506`	`+ insert_default_configuration!(result, myself.persist_usage_report_interval_in_seconds);`
`499`	`507`	`result.insert(`
`500`	`508`	`"cardano_transactions_signing_config".to_string(),`
`501`	`509`	`into_value(HashMap::from([`
`@@ -513,7 +521,6 @@ impl Source for DefaultConfiguration {`
`513`	`521`	`),`
`514`	`522`	`])),`
`515`	`523`	`);`
`516`		`-`
`517`	`524`	`Ok(result)`
`518`	`525`	`}`
`519`	`526`	`}`