add omni-executor metrics (#3423)

Author: kziemianek

tee-worker/omni-executor/Cargo.lock

@@ -52,6 +52,8 @@ jsonrpsee = { version = "0.24", features = ["server"] }
 jsonwebtoken = "9.3.0"
 libsecp256k1 = "0.7.1"
 log = "0.4.22"
+metrics = "0.24.1"
+metrics-exporter-prometheus = "0.16.2"
 mockall = "0.13.1"
 parity-scale-codec = "3.6.12"
 rand = "0.8.5"

tee-worker/omni-executor/Cargo.toml

@@ -22,12 +22,12 @@ Gramine is required for running inside TEE, please refer to [installation option
 2. Build omni-executor docker image:
 
    ```bash
-   make docker-build
+   make build-docker
    ```
 
 3. Start local omni-executor:
    ```bash
    make start-local
    ```
 
-First service run will generate substrate account, it needs to set as omni executor in `omniAccount` pallet. 
+First service run will generate substrate account, it needs to set as omni executor in `omniAccount` pallet. 

tee-worker/omni-executor/README.md

@@ -25,6 +25,7 @@ services:
       - litentry-test-network
     ports:
       - "2100:2100"
+      - "9090:9090"
     healthcheck:
       test: [ "CMD-SHELL", "curl -X POST -H 'Content-Type: application/json' -d '{\"jsonrpc\":\"2.0\",\"method\":\"omni_getHealth\",\"id\":1}' http://localhost:2100 || exit 1"]
       interval: 30s

tee-worker/omni-executor/docker/docker-compose.yml

@@ -7,6 +7,7 @@ edition.workspace = true
 [dependencies]
 async-trait = { workspace = true }
 log = { workspace = true }
+metrics = { workspace = true }
 parity-scale-codec = { workspace = true, features = ["derive"] }
 rand = { workspace = true }
 regex = { workspace = true }

tee-worker/omni-executor/executor-core/Cargo.toml

@@ -20,6 +20,7 @@ use tokio::sync::mpsc;
 use executor_primitives::AccountId;
 use executor_primitives::Intent;
 use executor_primitives::IntentId;
+use metrics::describe_gauge;
 
 pub type IntentExecutionResult = (Option<Vec<u8>>, bool);
 
@@ -32,6 +33,13 @@ pub trait IntentExecutor: Send {
 		intent_id: IntentId,
 		intent: Intent,
 	) -> Result<IntentExecutionResult, ()>;
+
+	async fn name(&self) -> &'static str;
+
+	async fn on_execution_error(&self) {
+		let name = self.name().await;
+		describe_gauge!(executor_gauge_name(name), executor_gauge_desc(name));
+	}
 }
 
 pub struct MockedIntentExecutor {
@@ -55,4 +63,16 @@ impl IntentExecutor for MockedIntentExecutor {
 	) -> Result<IntentExecutionResult, ()> {
 		self.sender.send(()).map(|_| (None, false)).map_err(|_| ())
 	}
+
+	async fn name(&self) -> &'static str {
+		"mocked"
+	}
+}
+
+fn executor_gauge_name(name: &str) -> String {
+	format!("{}_intent_execution_failures", name)
+}
+
+fn executor_gauge_desc(name: &str) -> String {
+	format!("Number of {} intent executor failures", name)
 }

tee-worker/omni-executor/executor-core/src/intent_executor.rs

@@ -23,6 +23,7 @@ use crate::event_handler::{Error, EventHandler};
 use crate::fetcher::{EventsFetcher, LastFinalizedBlockNumFetcher};
 use crate::sync_checkpoint_repository::{Checkpoint, CheckpointRepository};
 use executor_primitives::GetEventId;
+use metrics::{describe_gauge, gauge};
 
 /// Component, used to listen to chain and execute requested intents
 /// Requires specific implementations of:
@@ -65,6 +66,7 @@ impl<
 		stop_signal: Receiver<()>,
 		last_processed_log_repository: CheckpointRepositoryT,
 	) -> Result<Self, ()> {
+		describe_gauge!(synced_block_gauge_name(id), "Last synced block");
 		Ok(Self {
 			id: id.to_string(),
 			handle,
@@ -185,6 +187,7 @@ impl<
 						self.checkpoint_repository
 							.save(CheckpointT::from(block_number_to_sync))
 							.expect("Could not save checkpoint");
+						gauge!(synced_block_gauge_name(&self.id)).set(block_number_to_sync as f64);
 						log::debug!("Finished syncing block: {}", block_number_to_sync);
 						block_number_to_sync += 1;
 					},
@@ -205,3 +208,7 @@ impl<
 		}
 	}
 }
+
+fn synced_block_gauge_name(listener_id: &str) -> String {
+	format!("{}_synced_block", listener_id)
+}

tee-worker/omni-executor/executor-core/src/listener.rs

@@ -10,6 +10,7 @@ clap = { workspace = true, features = ["derive"] }
 env_logger = { workspace = true }
 hex = { workspace = true }
 log = { workspace = true }
+metrics-exporter-prometheus = { workspace = true }
 scale-encode = { workspace = true }
 serde_json = { workspace = true }
 tokio = { workspace = true, features = ["macros", "rt-multi-thread", "signal"] }

tee-worker/omni-executor/executor-worker/Cargo.toml

@@ -43,6 +43,8 @@ pub struct RunArgs {
 	pub accounting_contract_address: String,
 	#[arg(long, value_name = "should sync with parentchain")]
 	pub parentchain_sync: bool,
+	#[arg(short, long, default_value = "9090", value_name = "metrics port")]
+	pub metrics_port: String,
 }
 
 #[derive(Args)]

tee-worker/omni-executor/executor-worker/src/cli.rs

@@ -31,6 +31,7 @@ use executor_crypto::{ecdsa, PairTrait};
 use executor_primitives::AccountId;
 use executor_storage::{init_storage, StorageDB};
 use log::{error, info};
+use metrics_exporter_prometheus::PrometheusBuilder;
 use native_task_handler::{
 	run_native_task_handler, Aes256KeyStore, TaskHandlerContext, MAX_CONCURRENT_TASKS,
 };
@@ -49,7 +50,9 @@ use solana::SolanaClient;
 use solana_intent_executor::SolanaIntentExecutor;
 use std::env;
 use std::io::Write;
+use std::net::SocketAddr;
 use std::path::Path;
+use std::str::FromStr;
 use std::sync::Arc;
 use std::thread;
 use std::thread::JoinHandle;
@@ -80,6 +83,14 @@ async fn main() -> Result<(), ()> {
 
 	match cli.cmd {
 		Commands::Run(args) => {
+			let builder = PrometheusBuilder::new();
+
+			let address = SocketAddr::from_str(&format!("0.0.0.0:{}", args.metrics_port)).unwrap();
+			builder
+				.with_http_listener(address)
+				.install()
+				.expect("failed to install Prometheus recorder");
+
 			let auth_token_key_store = AuthTokenKeyStore::new(
 				Path::new(&args.local_directory_path)
 					.join("keystore/auth_token_key.bin")

tee-worker/omni-executor/executor-worker/src/main.rs

@@ -1078,6 +1078,10 @@ impl<Provider: EthereumRpcProvider<Transaction = TransactionRequest> + Send + Sy
 			},
 		}
 	}
+
+	async fn name(&self) -> &'static str {
+		"cross-chain"
+	}
 }
 
 fn str_to_u256(amount: &str, decimals: u32) -> Option<U256> {