Merge pull request #759 from input-output-hk/greg/743/extract_stake_distribution

#743 monitor stake distribution Vs Signer version

Author: ghubertpalo

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

@@ -3,6 +3,7 @@ use crate::{
     dependency::MultiSignerWrapper, CertificatePendingStore, CertificateStore, Configuration,
     DependencyManager, ProtocolParametersStore, SignerRegisterer, SnapshotStore,
 };
+use mithril_common::BeaconProvider;
 use std::convert::Infallible;
 use std::sync::Arc;
 use warp::Filter;
@@ -62,3 +63,10 @@ pub fn with_event_transmitter(
 ) -> impl Filter<Extract = (Arc<TransmitterService<EventMessage>>,), Error = Infallible> + Clone {
     warp::any().map(move || dependency_manager.event_transmitter.clone())
 }
+
+/// With round_opener middleware
+pub fn with_beacon_provider(
+    dependency_manager: Arc<DependencyManager>,
+) -> impl Filter<Extract = (Arc<dyn BeaconProvider>,), Error = Infallible> + Clone {
+    warp::any().map(move || dependency_manager.beacon_provider.clone())
+}

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

@@ -24,7 +24,10 @@ fn register_signer(
         .and(middlewares::with_signer_registerer(
             dependency_manager.clone(),
         ))
-        .and(middlewares::with_event_transmitter(dependency_manager))
+        .and(middlewares::with_event_transmitter(
+            dependency_manager.clone(),
+        ))
+        .and(middlewares::with_beacon_provider(dependency_manager))
         .and_then(handlers::register_signer)
 }
 
@@ -33,6 +36,7 @@ mod handlers {
     use crate::FromRegisterSignerAdapter;
     use crate::{http_server::routes::reply, SignerRegisterer, SignerRegistrationError};
     use mithril_common::messages::RegisterSignerMessage;
+    use mithril_common::BeaconProvider;
     use slog_scope::{debug, warn};
     use std::convert::Infallible;
     use std::sync::Arc;
@@ -44,34 +48,43 @@ mod handlers {
         register_signer_message: RegisterSignerMessage,
         signer_registerer: Arc<dyn SignerRegisterer>,
         event_transmitter: Arc<TransmitterService<EventMessage>>,
+        beacon_provider: Arc<dyn BeaconProvider>,
     ) -> Result<impl warp::Reply, Infallible> {
         debug!(
             "⇄ HTTP SERVER: register_signer/{:?}",
             register_signer_message
         );
         let signer = FromRegisterSignerAdapter::adapt(register_signer_message);
-        let headers: Vec<(&str, &str)> = match signer_node_version.as_ref() {
+        let mut headers: Vec<(&str, &str)> = match signer_node_version.as_ref() {
             Some(version) => vec![("signer-node-version", version)],
             None => Vec::new(),
         };
+        let epoch_str = if let Ok(beacon) = beacon_provider.get_current_beacon().await {
+            format!("{}", beacon.epoch)
+        } else {
+            String::new()
+        };
 
+        if epoch_str.is_empty() {
+            headers.push(("epoch", epoch_str.as_str()));
+        }
         match signer_registerer.register_signer(&signer).await {
-            Ok(()) => {
+            Ok(signer_with_stake) => {
                 let _ = event_transmitter.send_event_message(
                     "HTTP::signer_register",
                     "register_signer",
-                    &signer,
+                    &signer_with_stake,
                     headers,
                 );
 
                 Ok(reply::empty(StatusCode::CREATED))
             }
-            Err(SignerRegistrationError::ExistingSigner()) => {
+            Err(SignerRegistrationError::ExistingSigner(signer_with_stake)) => {
                 debug!("register_signer::already_registered");
                 let _ = event_transmitter.send_event_message(
                     "HTTP::signer_register",
                     "register_signer",
-                    &signer,
+                    &signer_with_stake,
                     headers,
                 );
                 Ok(reply::empty(StatusCode::CREATED))
@@ -105,6 +118,7 @@ mod tests {
     use mithril_common::crypto_helper::ProtocolRegistrationError;
     use mithril_common::messages::RegisterSignerMessage;
     use mithril_common::test_utils::apispec::APISpec;
+    use mithril_common::test_utils::fake_data;
     use warp::http::Method;
     use warp::test::request;
 
@@ -128,10 +142,11 @@ mod tests {
 
     #[tokio::test]
     async fn test_register_signer_post_ok() {
+        let signer_with_stake = fake_data::signers_with_stakes(1).pop().unwrap();
         let mut mock_signer_registerer = MockSignerRegisterer::new();
         mock_signer_registerer
             .expect_register_signer()
-            .return_once(|_| Ok(()));
+            .return_once(|_| Ok(signer_with_stake));
         let (mut dependency_manager, _) = initialize_dependencies().await;
         dependency_manager.signer_registerer = Arc::new(mock_signer_registerer);
 
@@ -158,10 +173,11 @@ mod tests {
 
     #[tokio::test]
     async fn test_register_signer_post_ok_existing() {
+        let signer_with_stake = fake_data::signers_with_stakes(1).pop().unwrap();
         let mut mock_signer_registerer = MockSignerRegisterer::new();
         mock_signer_registerer
             .expect_register_signer()
-            .return_once(|_| Err(SignerRegistrationError::ExistingSigner()));
+            .return_once(|_| Err(SignerRegistrationError::ExistingSigner(signer_with_stake)));
         let (mut dependency_manager, _) = initialize_dependencies().await;
         dependency_manager.signer_registerer = Arc::new(mock_signer_registerer);
 

mithril-aggregator/src/signer_registerer.rs

@@ -9,7 +9,7 @@ use mithril_common::{
     crypto_helper::{
         key_decode_hex, KESPeriod, ProtocolKeyRegistration, ProtocolRegistrationError,
     },
-    entities::{Epoch, Signer, StakeDistribution},
+    entities::{Epoch, Signer, SignerWithStake, StakeDistribution},
     store::StoreError,
 };
 
@@ -33,7 +33,7 @@ pub enum SignerRegistrationError {
 
     /// Signer is already registered.
     #[error("signer already registered")]
-    ExistingSigner(),
+    ExistingSigner(SignerWithStake),
 
     /// Store error.
     #[error("store error: {0}")]
@@ -66,7 +66,10 @@ impl SignerRegistrationRound {
 #[async_trait]
 pub trait SignerRegisterer: Sync + Send {
     /// Register a signer
-    async fn register_signer(&self, signer: &Signer) -> Result<(), SignerRegistrationError>;
+    async fn register_signer(
+        &self,
+        signer: &Signer,
+    ) -> Result<SignerWithStake, SignerRegistrationError>;
 }
 
 /// Trait to open a signer registration round
@@ -141,7 +144,10 @@ impl SignerRegistrationRoundOpener for MithrilSignerRegisterer {
 
 #[async_trait]
 impl SignerRegisterer for MithrilSignerRegisterer {
-    async fn register_signer(&self, signer: &Signer) -> Result<(), SignerRegistrationError> {
+    async fn register_signer(
+        &self,
+        signer: &Signer,
+    ) -> Result<SignerWithStake, SignerRegistrationError> {
         let registration_round = self.current_round.read().await;
         let registration_round = registration_round
             .as_ref()
@@ -191,16 +197,22 @@ impl SignerRegisterer for MithrilSignerRegisterer {
             kes_period,
             verification_key,
         )?;
-        let mut signer_save = signer.to_owned();
+        let mut signer_save = SignerWithStake::from_signer(
+            signer.to_owned(),
+            *registration_round
+                .stake_distribution
+                .get(&party_id_save)
+                .unwrap(),
+        );
         signer_save.party_id = party_id_save;
 
         match self
             .verification_key_store
-            .save_verification_key(registration_round.epoch, signer_save)
+            .save_verification_key(registration_round.epoch, signer_save.clone().into())
             .await?
         {
-            Some(_) => Err(SignerRegistrationError::ExistingSigner()),
-            None => Ok(()),
+            Some(_) => Err(SignerRegistrationError::ExistingSigner(signer_save)),
+            None => Ok(signer_save),
         }
     }
 }

mithril-aggregator/utils/monitoring/README.md

@@ -0,0 +1,17 @@
+# Monitoring
+
+## Stakes vs Signer Version distribution
+
+In order to set the epoch of Era changes, it is required to know the
+distribution of the stakes that run a compatible Signer node. There is a SQL
+query for that.
+
+```sh
+$> sqlite3 -table -batch \
+       $DATA_STORES_DIRECTORY/monitoring.sqlite3 \
+       <  mithril-aggregator/utils/monitoring/stake_signer_version.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).

mithril-aggregator/utils/monitoring/stake_signer_version.sql

@@ -0,0 +1,44 @@
+with
+  signer_version as (
+    select
+       content->>'$.content.party_id' as party_id,
+       content->>'$.headers.signer-node-version' as node_version,
+       content->>'$.headers.epoch' as epoch,
+       content->>'$.content.stake' as stakes
+    from event
+    order by created_at desc
+  ),
+  stakes_version as (
+    select
+      party_id,
+      case
+        when instr(signer_version.node_version, '+') > 0
+          then substr(signer_version.node_version, 0, instr(signer_version.node_version, '+'))
+          else signer_version.node_version
+      end as version,
+      stakes,
+      cast(epoch as int) as epoch
+    from signer_version
+    group by party_id, epoch
+  ),
+  summed_stakes_version as (
+    select
+      epoch,
+      version,
+      party_id,
+      sum(stakes) over (partition by epoch) as total_epoch_stakes,
+      sum(stakes) over (partition by epoch, version) as stakes_version
+    from stakes_version
+    order by epoch desc
+  )
+select
+  epoch,
+  version,
+  total_epoch_stakes,
+  stakes_version,
+  printf("%02d %%", round((stakes_version * 100) / (total_epoch_stakes * 1.0))) as stakes_ratio,
+  count(party_id) as pool_count
+from summed_stakes_version
+group by epoch, version
+order by epoch desc, version desc
+;

mithril-common/src/entities/signer.rs

@@ -129,6 +129,18 @@ impl SignerWithStake {
         }
     }
 
+    /// Turn a [Signer] into a [SignerWithStake].
+    pub fn from_signer(signer: Signer, stake: Stake) -> Self {
+        Self {
+            party_id: signer.party_id,
+            verification_key: signer.verification_key,
+            verification_key_signature: signer.verification_key_signature,
+            operational_certificate: signer.operational_certificate,
+            kes_period: signer.kes_period,
+            stake,
+        }
+    }
+
     /// Computes the hash of SignerWithStake
     pub fn compute_hash(&self) -> String {
         let mut hasher = Sha256::new();