feat(fortuna): implement metrics

apps/fortuna/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "fortuna"
-version = "5.2.4"
+version = "5.3.0"
 edition = "2021"
 
 [dependencies]

apps/fortuna/src/api.rs

@@ -52,20 +52,45 @@ mod revelation;
 
 pub type ChainId = String;
 
+#[derive(Clone, Debug, Hash, PartialEq, Eq, EncodeLabelSet)]
+pub struct RequestLabel {
+ pub value: String,
+}
+
+pub struct ApiMetrics {
+ pub http_requests: Family<RequestLabel, Counter>,
+}
+
 #[derive(Clone)]
 pub struct ApiState {
  pub chains: Arc<HashMap<ChainId, BlockchainState>>,
 
+ pub metrics_registry: Arc<RwLock<Registry>>,
+
  /// Prometheus metrics
- pub metrics: Arc<Metrics>,
+ pub metrics: Arc<ApiMetrics>,
 }
 
 impl ApiState {
- pub fn new(chains: &[(ChainId, BlockchainState)]) -> ApiState {
- let map: HashMap<ChainId, BlockchainState> = chains.into_iter().cloned().collect();
+ pub async fn new(
+ chains: HashMap<ChainId, BlockchainState>,
+ metrics_registry: Arc<RwLock<Registry>>,
+ ) -> ApiState {
+ let metrics = ApiMetrics {
+ http_requests: Family::default(),
+ };
+
+ let http_requests = metrics.http_requests.clone();
+ metrics_registry.write().await.register(
+ "http_requests",
+ "Number of HTTP requests received",
+ http_requests,
+ );
+
  ApiState {
- chains: Arc::new(map),
- metrics: Arc::new(Metrics::new()),
+ chains: Arc::new(chains),
+ metrics: Arc::new(metrics),
+ metrics_registry,
  }
  }
 }
@@ -89,38 +114,6 @@ pub struct BlockchainState {
  pub confirmed_block_status: BlockStatus,
 }
 
-pub struct Metrics {
- pub registry: RwLock<Registry>,
- // TODO: track useful metrics. this counter is just a placeholder to get things set up.
- pub request_counter: Family<Label, Counter>,
-}
-
-#[derive(Clone, Debug, Hash, PartialEq, Eq, EncodeLabelSet)]
-pub struct Label {
- value: String,
-}
-
-impl Metrics {
- pub fn new() -> Self {
- let mut metrics_registry = Registry::default();
- let http_requests = Family::<Label, Counter>::default();
-
- // Register the metric family with the registry.
- metrics_registry.register(
- // With the metric name.
- "http_requests",
- // And the metric help text.
- "Number of HTTP requests received",
- http_requests.clone(),
- );
-
- Metrics {
- registry: RwLock::new(metrics_registry),
- request_counter: http_requests,
- }
- }
-}
-
 pub enum RestError {
  /// The caller passed a sequence number that isn't within the supported range
  InvalidSequenceNumber,
@@ -225,7 +218,12 @@ mod test {
  },
  ethers::prelude::Address,
  lazy_static::lazy_static,
- std::sync::Arc,
+ prometheus_client::registry::Registry,
+ std::{
+ collections::HashMap,
+ sync::Arc,
+ },
+ tokio::sync::RwLock,
  };
 
  const PROVIDER: Address = Address::zero();
@@ -243,7 +241,7 @@ mod test {
  ));
  }
 
- fn test_server() -> (TestServer, Arc<MockEntropyReader>, Arc<MockEntropyReader>) {
+ async fn test_server() -> (TestServer, Arc<MockEntropyReader>, Arc<MockEntropyReader>) {
  let eth_read = Arc::new(MockEntropyReader::with_requests(10, &[]));
 
  let eth_state = BlockchainState {
@@ -255,6 +253,8 @@ mod test {
  confirmed_block_status: BlockStatus::Latest,
  };
 
+ let metrics_registry = Arc::new(RwLock::new(Registry::default()));
+
  let avax_read = Arc::new(MockEntropyReader::with_requests(10, &[]));
 
  let avax_state = BlockchainState {
@@ -266,10 +266,11 @@ mod test {
  confirmed_block_status: BlockStatus::Latest,
  };
 
- let api_state = ApiState::new(&[
- ("ethereum".into(), eth_state),
- ("avalanche".into(), avax_state),
- ]);
+ let mut chains = HashMap::new();
+ chains.insert("ethereum".into(), eth_state);
+ chains.insert("avalanche".into(), avax_state);
+
+ let api_state = ApiState::new(chains, metrics_registry).await;
 
  let app = api::routes(api_state);
  (TestServer::new(app).unwrap(), eth_read, avax_read)
@@ -287,7 +288,7 @@ mod test {
 
  #[tokio::test]
  async fn test_revelation() {
- let (server, eth_contract, avax_contract) = test_server();
+ let (server, eth_contract, avax_contract) = test_server().await;
 
  // Can't access a revelation if it hasn't been requested
  get_and_assert_status(
@@ -416,7 +417,7 @@ mod test {
 
  #[tokio::test]
  async fn test_revelation_confirmation_delay() {
- let (server, eth_contract, avax_contract) = test_server();
+ let (server, eth_contract, avax_contract) = test_server().await;
 
  eth_contract.insert(PROVIDER, 0, 10, false);
  eth_contract.insert(PROVIDER, 1, 11, false);

apps/fortuna/src/api/metrics.rs

@@ -9,7 +9,7 @@ use {
 };
 
 pub async fn metrics(State(state): State<crate::api::ApiState>) -> impl IntoResponse {
- let registry = state.metrics.registry.read().await;
+ let registry = state.metrics_registry.read().await;
  let mut buffer = String::new();
 
  // Should not fail if the metrics are valid and there is memory available

apps/fortuna/src/api/revelation.rs

@@ -1,7 +1,7 @@
 use {
  crate::api::{
  ChainId,
- Label,
+ RequestLabel,
  RestError,
  },
  anyhow::Result,
@@ -45,8 +45,8 @@ pub async fn revelation(
 ) -> Result<Json<GetRandomValueResponse>, RestError> {
  state
  .metrics
- .request_counter
- .get_or_create(&Label {
+ .http_requests
+ .get_or_create(&RequestLabel {
  value: "/v1/chains/{chain_id}/revelations/{sequence}".to_string(),
  })
  .inc();

apps/fortuna/src/command/run.rs

@@ -25,14 +25,18 @@ use {
  Result,
  },
  axum::Router,
+ prometheus_client::registry::Registry,
  std::{
  collections::HashMap,
  net::SocketAddr,
  sync::Arc,
  },
  tokio::{
  spawn,
- sync::watch,
+ sync::{
+ watch,
+ RwLock,
+ },
  },
  tower_http::cors::CorsLayer,
  utoipa::OpenApi,
@@ -42,6 +46,7 @@ use {
 pub async fn run_api(
  socket_addr: SocketAddr,
  chains: HashMap<String, api::BlockchainState>,
+ metrics_registry: Arc<RwLock<Registry>>,
  mut rx_exit: watch::Receiver<bool>,
 ) -> Result<()> {
  #[derive(OpenApi)]
@@ -63,11 +68,7 @@ pub async fn run_api(
  )]
  struct ApiDoc;
 
- let metrics_registry = api::Metrics::new();
- let api_state = api::ApiState {
- chains: Arc::new(chains),
- metrics: Arc::new(metrics_registry),
- };
+ let api_state = api::ApiState::new(chains, metrics_registry).await;
 
  // Initialize Axum Router. Note the type here is a `Router<State>` due to the use of the
  // `with_state` method which replaces `Body` with `State` in the type signature.
@@ -101,6 +102,7 @@ pub async fn run_keeper(
  chains: HashMap<String, api::BlockchainState>,
  config: Config,
  private_key: String,
+ metrics_registry: Arc<RwLock<Registry>>,
 ) -> Result<()> {
  let mut handles = Vec::new();
  for (chain_id, chain_config) in chains {
@@ -114,6 +116,7 @@ pub async fn run_keeper(
  private_key,
  chain_eth_config,
  chain_config.clone(),
+ metrics_registry.clone(),
  )));
  }
 
@@ -218,11 +221,18 @@ pub async fn run(opts: &RunOptions) -> Result<()> {
  Ok::<(), Error>(())
  });
 
+ let metrics_registry = Arc::new(RwLock::new(Registry::default()));
+
  if let Some(keeper_private_key) = opts.load_keeper_private_key()? {
- spawn(run_keeper(chains.clone(), config, keeper_private_key));
+ spawn(run_keeper(
+ chains.clone(),
+ config.clone(),
+ keeper_private_key,
+ metrics_registry.clone(),
+ ));
  }
 
- run_api(opts.addr.clone(), chains, rx_exit).await?;
+ run_api(opts.addr.clone(), chains, metrics_registry.clone(), rx_exit).await?;
 
  Ok(())
 }