Merge pull request #1239 from input-output-hk/greg/1127/download_stat…

…istics

add snapshot downloads statistics

.gitignore

@@ -4,6 +4,7 @@
 .DS_Store
 .direnv/
 .tmp/
+.vscode/
 target
 target-*
 test-results.xml

docs/runbook/README.md

@@ -10,6 +10,7 @@ This page gathers the available guides to operate a Mithril network.
 |------------|------------|------------
 | **Genesis manually** | [manual-genesis](./genesis-manually/README.md) | Proceed to manual (re)genesis of the aggregator certificate chain.
 | **Era markers** | [era-markers](./era-markers/README.md) | Create and update era markers on the Cardano chain.
+| **Downloads statistics** | [downloads statistics](./statistics/README.md) | Display the number of downloads per day.
 | **Signer registrations monitoring** | [registrations-monitoring](./registrations-monitoring/README.md) | Gather aggregated data about signer registrations (versions, stake, ...).
 | **Update protocol parameters** | [protocol-parameters](./protocol-parameters/README.md) | Update the protocol parameters of a Mithril network.
 | **Recompute certificates hash** | [recompute-certificates-hash](./recompute-certificates-hash/README.md) | Recompute the certificates has of an aggregator.

docs/runbook/statistics/README.md

@@ -0,0 +1,14 @@
+# Statistics
+
+## Snapshot downloads per day
+
+```sh
+$> sqlite3 -table -batch \
+       $DATA_STORES_DIRECTORY/monitoring.sqlite3 \
+       <  snapshot_downloads.sql
+```
+
+The variable `$DATA_STORES_DIRECTORY` should point to the directory where the
+databases files are stored (see files in `mithril-aggregator/config` using the
+key `data_stores_directory` to know where they are).
+

docs/runbook/statistics/snapshot_downloads.sql

@@ -0,0 +1,12 @@
+select
+  date(created_at)   as downloaded_at,
+  count(*)           as downloads
+from event
+where
+  source = 'HTTP::statistics'
+  and action = 'snapshot_downloaded'
+group by 1
+order by
+  downloaded_at desc,
+  downloads desc
+

mithril-aggregator/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mithril-aggregator"
-version = "0.3.93"
+version = "0.3.94"
 description = "A Mithril Aggregator server"
 authors = { workspace = true }
 edition = { workspace = true }

mithril-aggregator/src/http_server/routes/mod.rs

@@ -6,3 +6,4 @@ mod reply;
 pub mod router;
 mod signatures_routes;
 mod signer_routes;
+mod statistics_routes;

mithril-aggregator/src/http_server/routes/router.rs

@@ -1,5 +1,6 @@
 use crate::http_server::routes::{
     artifact_routes, certificate_routes, epoch_routes, signatures_routes, signer_routes,
+    statistics_routes,
 };
 use crate::http_server::SERVER_BASE_PATH;
 use crate::DependencyContainer;
@@ -49,6 +50,7 @@ pub fn routes(
                 .or(signer_routes::routes(dependency_manager.clone()))
                 .or(signatures_routes::routes(dependency_manager.clone()))
                 .or(epoch_routes::routes(dependency_manager.clone()))
+                .or(statistics_routes::routes(dependency_manager.clone()))
                 .with(cors),
         )
         .recover(handle_custom)

mithril-aggregator/src/http_server/routes/statistics_routes.rs

@@ -0,0 +1,108 @@
+use std::sync::Arc;
+use warp::Filter;
+
+use crate::http_server::routes::middlewares;
+use crate::DependencyContainer;
+
+pub fn routes(
+    dependency_manager: Arc<DependencyContainer>,
+) -> impl Filter<Extract = (impl warp::Reply,), Error = warp::Rejection> + Clone {
+    post_statistics(dependency_manager)
+}
+
+/// POST /statistics/snapshot
+fn post_statistics(
+    dependency_manager: Arc<DependencyContainer>,
+) -> impl Filter<Extract = (impl warp::Reply,), Error = warp::Rejection> + Clone {
+    warp::path!("statistics" / "snapshot")
+        .and(warp::post())
+        .and(warp::body::json())
+        .and(middlewares::with_event_transmitter(
+            dependency_manager.clone(),
+        ))
+        .and_then(handlers::post_snapshot_statistics)
+}
+
+mod handlers {
+    use std::{convert::Infallible, sync::Arc};
+
+    use mithril_common::messages::SnapshotMessage;
+    use reqwest::StatusCode;
+
+    use crate::event_store::{EventMessage, TransmitterService};
+    use crate::http_server::routes::reply;
+
+    pub async fn post_snapshot_statistics(
+        snapshot_message: SnapshotMessage,
+        event_transmitter: Arc<TransmitterService<EventMessage>>,
+    ) -> Result<impl warp::Reply, Infallible> {
+        let headers: Vec<(&str, &str)> = Vec::new();
+
+        match event_transmitter.send_event_message(
+            "HTTP::statistics",
+            "snapshot_downloaded",
+            &snapshot_message,
+            headers,
+        ) {
+            Err(e) => Ok(reply::internal_server_error(e)),
+            Ok(_) => Ok(reply::empty(StatusCode::CREATED)),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    use mithril_common::messages::SnapshotMessage;
+    use mithril_common::test_utils::apispec::APISpec;
+
+    use warp::{http::Method, test::request};
+
+    use crate::{
+        dependency_injection::DependenciesBuilder, http_server::SERVER_BASE_PATH, Configuration,
+    };
+
+    fn setup_router(
+        dependency_manager: Arc<DependencyContainer>,
+    ) -> impl Filter<Extract = (impl warp::Reply,), Error = warp::Rejection> + Clone {
+        let cors = warp::cors()
+            .allow_any_origin()
+            .allow_headers(vec!["content-type"])
+            .allow_methods(vec![Method::GET, Method::POST, Method::OPTIONS]);
+
+        warp::any()
+            .and(warp::path(SERVER_BASE_PATH))
+            .and(routes(dependency_manager).with(cors))
+    }
+
+    #[tokio::test]
+    async fn post_statistics_ok() {
+        let config = Configuration::new_sample();
+        let mut builder = DependenciesBuilder::new(config);
+        let mut rx = builder.get_event_transmitter_receiver().await.unwrap();
+        let dependency_manager = builder.build_dependency_container().await.unwrap();
+        let snapshot_message = SnapshotMessage::dummy();
+
+        let method = Method::POST.as_str();
+        let path = "/statistics/snapshot";
+
+        let response = request()
+            .method(method)
+            .json(&snapshot_message)
+            .path(&format!("/{SERVER_BASE_PATH}{path}"))
+            .reply(&setup_router(Arc::new(dependency_manager)))
+            .await;
+
+        APISpec::verify_conformity(
+            APISpec::get_all_spec_files(),
+            method,
+            path,
+            "application/json",
+            &snapshot_message,
+            &response,
+        );
+
+        let _ = rx.try_recv().unwrap();
+    }
+}

mithril-client/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mithril-client"
-version = "0.4.0"
+version = "0.4.1"
 description = "A Mithril Client"
 authors = { workspace = true }
 edition = { workspace = true }

mithril-client/src/aggregator_client/http_client.rs

@@ -54,6 +54,13 @@ pub trait AggregatorClient: Sync + Send {
     /// Get the content back from the Aggregator, the URL is a relative path for a resource
     async fn get_content(&self, url: &str) -> Result<String, AggregatorHTTPClientError>;
 
+    /// Post information to the Aggregator, the URL is a relative path for a resource
+    async fn post_content(
+        &self,
+        url: &str,
+        json: &str,
+    ) -> Result<String, AggregatorHTTPClientError>;
+
     /// Download and unpack large archives on the disk
     async fn download_unpack(
         &self,
@@ -149,6 +156,46 @@ impl AggregatorHTTPClient {
         }
     }
 
+    /// Issue a POST HTTP request.
+    #[async_recursion]
+    async fn post(&self, url: &str, json: &str) -> Result<Response, AggregatorHTTPClientError> {
+        let request_builder = Client::new().post(url.to_owned()).json(json);
+        let current_api_version = self
+            .compute_current_api_version()
+            .await
+            .unwrap()
+            .to_string();
+        debug!("Prepare request with version: {current_api_version}");
+        let request_builder =
+            request_builder.header(MITHRIL_API_VERSION_HEADER, current_api_version);
+
+        let response = request_builder.send().await.map_err(|e| {
+            AggregatorHTTPClientError::SubsystemError {
+                message: format!("Error while POSTing data '{json}' to URL='{url}'."),
+                error: e.into(),
+            }
+        })?;
+
+        match response.status() {
+            StatusCode::OK => Ok(response),
+            StatusCode::PRECONDITION_FAILED => {
+                if self.discard_current_api_version().await.is_some()
+                    && !self.api_versions.read().await.is_empty()
+                {
+                    return self.post(url, json).await;
+                }
+
+                Err(self.handle_api_error(&response).await)
+            }
+            StatusCode::NOT_FOUND => Err(AggregatorHTTPClientError::RemoteServerLogical(format!(
+                "Url='{url} not found"
+            ))),
+            status_code => Err(AggregatorHTTPClientError::RemoteServerTechnical(format!(
+                "Unhandled error {status_code}"
+            ))),
+        }
+    }
+
     /// API version error handling
     async fn handle_api_error(&self, response: &Response) -> AggregatorHTTPClientError {
         if let Some(version) = response.headers().get(MITHRIL_API_VERSION_HEADER) {
@@ -183,6 +230,23 @@ impl AggregatorClient for AggregatorHTTPClient {
             })
     }
 
+    async fn post_content(
+        &self,
+        url: &str,
+        json: &str,
+    ) -> Result<String, AggregatorHTTPClientError> {
+        let url = format!("{}/{}", self.aggregator_endpoint.trim_end_matches('/'), url);
+        let response = self.post(&url, json).await?;
+
+        response
+            .text()
+            .await
+            .map_err(|e| AggregatorHTTPClientError::SubsystemError {
+                message: "Could not find a text body in the response.".to_string(),
+                error: e.into(),
+            })
+    }
+
     async fn download_unpack(
         &self,
         url: &str,

mithril-client/src/aggregator_client/snapshot_client.rs

@@ -75,7 +75,12 @@ impl SnapshotClient {
                     )
                     .await
                 {
-                    Ok(()) => Ok(()),
+                    Ok(()) => {
+                        // the snapshot download does not fail if the statistic call fails.
+                        let _ = self.add_statistics(snapshot).await;
+
+                        Ok(())
+                    }
                     Err(e) => {
                         warn!("Failed downloading snapshot from '{url}' Error: {e}.");
                         Err(e.into())
@@ -92,4 +97,13 @@ impl SnapshotClient {
         }
         .into())
     }
+
+    /// Increments Aggregator's download statistics
+    pub async fn add_statistics(&self, snapshot: &Snapshot) -> StdResult<()> {
+        let url = "statistics/snapshot";
+        let json = serde_json::to_string(snapshot)?;
+        let _response = self.http_client.post_content(url, &json).await?;
+
+        Ok(())
+    }
 }

mithril-client/src/services/snapshot.rs

@@ -260,7 +260,12 @@ impl SnapshotService for MithrilClientSnapshotService {
                 DownloadProgressReporter::new(pb, progress_output_type),
             )
             .await
-            .with_context(|| format!("Could not download file in '{}'", download_dir.display()))?;
+            .with_context(|| {
+                format!(
+                    "Could not download file in directory '{}'",
+                    download_dir.display()
+                )
+            })?;
 
         // Append 'clean' file to speedup node bootstrap
         if let Err(error) = File::create(db_dir.join("clean")) {
@@ -541,8 +546,11 @@ mod tests {
         let test_path = std::env::temp_dir().join("test_download_snapshot_ok");
         let _ = std::fs::remove_dir_all(&test_path);
 
-        let (http_client, certificate_verifier, digester) =
+        let (mut http_client, certificate_verifier, digester) =
             get_mocks_for_snapshot_service_configured_to_make_download_succeed();
+        http_client
+            .expect_post_content()
+            .returning(|_, _| Ok(String::new()));
 
         let mut builder = get_dep_builder(Arc::new(http_client));
         builder.certificate_verifier = Some(Arc::new(certificate_verifier));
@@ -581,8 +589,11 @@ mod tests {
             .join("test_download_snapshot_ok_add_clean_file_allowing_node_bootstrap_speedup");
         let _ = std::fs::remove_dir_all(&test_path);
 
-        let (http_client, certificate_verifier, digester) =
+        let (mut http_client, certificate_verifier, digester) =
             get_mocks_for_snapshot_service_configured_to_make_download_succeed();
+        http_client
+            .expect_post_content()
+            .returning(|_, _| Ok(String::new()));
 
         let mut builder = get_dep_builder(Arc::new(http_client));
         builder.certificate_verifier = Some(Arc::new(certificate_verifier));
@@ -625,8 +636,11 @@ mod tests {
         let test_path = std::env::temp_dir().join("test_download_snapshot_invalid_digest");
         let _ = std::fs::remove_dir_all(&test_path);
 
-        let (http_client, certificate_verifier, _) =
+        let (mut http_client, certificate_verifier, _) =
             get_mocks_for_snapshot_service_configured_to_make_download_succeed();
+        http_client
+            .expect_post_content()
+            .returning(|_, _| Ok(String::new()));
         let immutable_digester = DumbImmutableDigester::new("snapshot-digest-KO", true);
 
         let mut dep_builder = get_dep_builder(Arc::new(http_client));