feat: Add prometheus exporter and metrics service (#98)

* feat: Add prometheus endpoint and metrics service

* add docker prometheus and grafana

* update Cargo.toml

* rm monitoring

* transition_commitments -> block_commitments

Cargo.toml

@@ -290,6 +290,7 @@ hypercore-signer = { path = "hypercore/signer", default-features = false }
 hypercore-sequencer = { path = "hypercore/sequencer", default-features = false }
 hypercore-ethereum = { path = "hypercore/ethereum", default-features = false }
 hypercore-runtime-common = { path = "hypercore/runtime/common", default-features = false }
+hypercore-prometheus-endpoint = { path = "hypercore/utils/prometheus", default-features = false }
 hypercore-utils = { path = "hypercore/utils", default-features = false }
 hypercore-validator = { path = "hypercore/validator", default-features = false }
 hypercore-rpc = { path = "hypercore/rpc", default-features = false }

hypercore/cli/Cargo.toml

@@ -23,6 +23,9 @@ hypercore-sequencer.workspace = true
 hypercore-ethereum.workspace = true
 hypercore-validator.workspace = true
 hypercore-common.workspace = true
+hypercore-prometheus-endpoint.workspace = true
+hypercore-rpc.workspace = true
+hypercore-utils.workspace = true
 gprimitives.workspace = true
 
 clap = { workspace = true, features = ["derive"] }
@@ -44,6 +47,6 @@ parity-scale-codec = { workspace = true, features = ["std", "derive"] }
 hex.workspace = true
 rand.workspace = true
 tempfile.workspace = true
-hypercore-rpc.workspace = true
+futures-timer.workspace = true
 
 static_init = "1.0.3"

hypercore/cli/src/args.rs

@@ -18,7 +18,10 @@
 
 //! CLI arguments in one place.
 
-use crate::{config, params::NetworkParams};
+use crate::{
+    config,
+    params::{NetworkParams, PrometheusParams},
+};
 use anyhow::{anyhow, bail, Result};
 use clap::{Parser, Subcommand};
 use gprimitives::{ActorId, CodeId};
@@ -91,6 +94,10 @@ pub struct Args {
     #[clap(flatten)]
     pub network_params: NetworkParams,
 
+    #[allow(missing_docs)]
+    #[clap(flatten)]
+    pub prometheus_params: PrometheusParams,
+
     #[command(subcommand)]
     pub extra_command: Option<ExtraCommands>,
 }

hypercore/cli/src/config.rs

@@ -23,10 +23,13 @@ use crate::args::Args;
 use anyhow::{Context as _, Result};
 use directories::ProjectDirs;
 use hypercore_network::NetworkConfiguration;
+use hypercore_prometheus_endpoint::Registry;
 use hypercore_signer::PublicKey;
-use std::path::PathBuf;
+use std::{iter, net::SocketAddr, path::PathBuf};
 use tempfile::TempDir;
 
+const DEFAULT_PROMETHEUS_PORT: u16 = 9635;
+
 #[static_init::dynamic(drop, lazy)]
 static mut BASE_PATH_TEMP: Option<TempDir> = None;
 
@@ -44,6 +47,27 @@ pub enum ValidatorConfig {
     Disabled,
 }
 
+/// Configuration of the Prometheus endpoint.
+#[derive(Debug, Clone)]
+pub struct PrometheusConfig {
+    /// Port to use.
+    pub port: SocketAddr,
+    /// A metrics registry to use. Useful for setting the metric prefix.
+    pub registry: Registry,
+}
+
+impl PrometheusConfig {
+    /// Create a new config using the default registry.
+    pub fn new_with_default_registry(port: SocketAddr, chain_id: String) -> Self {
+        let param = iter::once((String::from("chain"), chain_id)).collect();
+        Self {
+            port,
+            registry: Registry::new_custom(None, Some(param))
+                .expect("this can only fail if the prefix is empty"),
+        }
+    }
+}
+
 #[derive(Debug)]
 pub struct Config {
     /// RPC of the Ethereum endpoint
@@ -79,6 +103,9 @@ pub struct Config {
     // Network configuration
     pub net_config: NetworkConfiguration,
 
+    // Prometheus configuration
+    pub prometheus_config: Option<PrometheusConfig>,
+
     /// RPC port
     pub rpc_port: u16,
 }
@@ -133,6 +160,9 @@ impl TryFrom<Args> for Config {
                 .unwrap_or(chain_spec.ethereum_router_address),
             max_commitment_depth: args.max_commitment_depth.unwrap_or(1000),
             net_config,
+            prometheus_config: args
+                .prometheus_params
+                .prometheus_config(DEFAULT_PROMETHEUS_PORT, "hypercore-dev".to_string()),
             database_path: base_path.join("db"),
             network_path: base_path.join("net"),
             key_path: base_path.join("key"),

hypercore/cli/src/main.rs

@@ -19,6 +19,7 @@
 mod args;
 mod chain_spec;
 mod config;
+mod metrics;
 mod params;
 mod service;
 

hypercore/cli/src/metrics.rs

@@ -0,0 +1,199 @@
+// This file is part of Gear.
+//
+// Copyright (C) 2024 Gear Technologies Inc.
+// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+use std::time::SystemTime;
+
+use crate::config::Config;
+use futures_timer::Delay;
+use hypercore_observer::ObserverStatus;
+use hypercore_prometheus_endpoint::{register, Gauge, Opts, PrometheusError, Registry, U64};
+use hypercore_sequencer::SequencerStatus;
+use hypercore_utils::metrics::register_globals;
+use std::time::{Duration, Instant};
+use tokio::sync::watch;
+
+struct PrometheusMetrics {
+    // generic info
+    eth_block_height: Gauge<U64>,
+    pending_upload_code: Gauge<U64>,
+    last_router_state: Gauge<U64>,
+    aggregated_commitments: Gauge<U64>,
+    submitted_code_commitments: Gauge<U64>,
+    submitted_block_commitments: Gauge<U64>,
+}
+
+impl PrometheusMetrics {
+    fn setup(registry: &Registry, name: &str) -> Result<Self, PrometheusError> {
+        register(
+            Gauge::<U64>::with_opts(
+                Opts::new(
+                    "hypercore_build_info",
+                    "A metric with a constant '1' value labeled by name, version",
+                )
+                .const_label("name", name),
+            )?,
+            registry,
+        )?
+        .set(1);
+
+        register_globals(registry)?;
+
+        let start_time_since_epoch = SystemTime::now()
+            .duration_since(SystemTime::UNIX_EPOCH)
+            .unwrap_or_default();
+        register(
+            Gauge::<U64>::new(
+                "hypercore_process_start_time_seconds",
+                "Number of seconds between the UNIX epoch and the moment the process started",
+            )?,
+            registry,
+        )?
+        .set(start_time_since_epoch.as_secs());
+
+        Ok(Self {
+            // generic internals
+            eth_block_height: register(
+                Gauge::<U64>::new(
+                    "hypercore_eth_block_height",
+                    "Block height info of the ethereum observer",
+                )?,
+                registry,
+            )?,
+
+            pending_upload_code: register(
+                Gauge::<U64>::new(
+                    "hypercore_pending_upload_code",
+                    "Pending upload code events of the ethereum observer",
+                )?,
+                registry,
+            )?,
+
+            last_router_state: register(
+                Gauge::<U64>::new(
+                    "hypercore_last_router_state",
+                    "Block height of the latest state of the router contract",
+                )?,
+                registry,
+            )?,
+
+            aggregated_commitments: register(
+                Gauge::<U64>::new(
+                    "hypercore_aggregated_commitments",
+                    "Number of commitments aggregated in sequencer",
+                )?,
+                registry,
+            )?,
+
+            submitted_code_commitments: register(
+                Gauge::<U64>::new(
+                    "hypercore_submitted_code_commitments",
+                    "Number of submitted code commitments in sequencer",
+                )?,
+                registry,
+            )?,
+
+            submitted_block_commitments: register(
+                Gauge::<U64>::new(
+                    "hypercore_submitted_block_commitments",
+                    "Number of submitted block commitments in sequencer",
+                )?,
+                registry,
+            )?,
+        })
+    }
+}
+
+/// A `MetricsService` periodically sends general client and
+/// network state to the telemetry as well as (optionally)
+/// a Prometheus endpoint.
+pub struct MetricsService {
+    metrics: Option<PrometheusMetrics>,
+    last_update: Instant,
+}
+
+impl MetricsService {
+    /// Creates a `MetricsService` that sends metrics
+    /// to prometheus alongside the telemetry.
+    pub fn with_prometheus(registry: &Registry, config: &Config) -> Result<Self, PrometheusError> {
+        PrometheusMetrics::setup(registry, &config.net_config.node_name).map(|p| MetricsService {
+            metrics: Some(p),
+            last_update: Instant::now(),
+        })
+    }
+
+    /// Returns a never-ending `Future` that performs the
+    /// metric and telemetry updates with information from
+    /// the given sources.
+    pub async fn run(
+        mut self,
+        mut observer_status: watch::Receiver<ObserverStatus>,
+        mut sequencer_status: Option<watch::Receiver<SequencerStatus>>,
+    ) {
+        let mut timer = Delay::new(Duration::from_secs(0));
+        let timer_interval = Duration::from_secs(5);
+
+        loop {
+            // Wait for the next tick of the timer.
+            (&mut timer).await;
+
+            // Update / Send the metrics.
+            self.update(
+                *observer_status.borrow_and_update(),
+                sequencer_status.as_mut().map(|s| *s.borrow_and_update()),
+            );
+
+            // Schedule next tick.
+            timer.reset(timer_interval);
+        }
+    }
+
+    fn update(
+        &mut self,
+        observer_status: ObserverStatus,
+        sequencer_status: Option<SequencerStatus>,
+    ) {
+        let now = Instant::now();
+        self.last_update = now;
+
+        let eth_number: u64 = observer_status.eth_block_number;
+        let pending_upload_code: u64 = observer_status.pending_upload_code;
+        let last_router_state: u64 = observer_status.last_router_state;
+
+        if let Some(metrics) = self.metrics.as_ref() {
+            metrics.eth_block_height.set(eth_number);
+            metrics.pending_upload_code.set(pending_upload_code);
+            metrics.last_router_state.set(last_router_state);
+            log::trace!("Observer status: {:?}", observer_status);
+            if let Some(sequencer_status) = sequencer_status {
+                metrics
+                    .aggregated_commitments
+                    .set(sequencer_status.aggregated_commitments);
+                metrics
+                    .submitted_code_commitments
+                    .set(sequencer_status.submitted_code_commitments);
+                metrics
+                    .submitted_block_commitments
+                    .set(sequencer_status.submitted_block_commitments);
+                log::trace!("Sequencer status: {:?}", sequencer_status);
+            }
+        }
+
+        // TODO: Use network status
+        // Update/send network status information, if any.
+    }
+}

hypercore/cli/src/params/mod.rs

@@ -16,6 +16,8 @@
 // You should have received a copy of the GNU General Public License
 // along with this program. If not, see <https://www.gnu.org/licenses/>.
 
-pub mod network_params;
+mod network_params;
+mod prometheus_params;
 
 pub use network_params::*;
+pub use prometheus_params::*;