diff --git a/Cargo.lock b/Cargo.lock index 252e5faa8..32ed13a68 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4310,9 +4310,9 @@ dependencies = [ "magicblock-keeper", "magicblock-ledger", "magicblock-ledger-deprecated", - "magicblock-metrics", "num_cpus", "parking_lot", + "prometheus", "reqwest 0.12.28", "serde", "solana-account 4.3.1", @@ -4441,7 +4441,6 @@ dependencies = [ "magicblock-delegation-program-api", "magicblock-engine", "magicblock-engine-nucleus", - "magicblock-metrics", "magicblock-program", "magicblock-rpc-client", "magicblock-table-mania", @@ -4730,21 +4729,6 @@ dependencies = [ "wincode", ] -[[package]] -name = "magicblock-metrics" -version = "1.0.1" -dependencies = [ - "http-body-util", - "hyper", - "hyper-util", - "lazy_static", - "magicblock-engine-nucleus", - "prometheus", - "solana-signature 3.4.1", - "tokio", - "tracing", -] - [[package]] name = "magicblock-processor" version = "0.15.0" @@ -4849,7 +4833,7 @@ version = "1.0.1" dependencies = [ "async-trait", "futures-util", - "magicblock-metrics", + "magicblock-engine-nucleus", "parking_lot", "serde_json", "solana-account 4.3.1", @@ -4924,7 +4908,7 @@ dependencies = [ name = "magicblock-table-mania" version = "1.0.1" dependencies = [ - "magicblock-metrics", + "magicblock-engine-nucleus", "magicblock-rpc-client", "parking_lot", "rand 0.9.5", @@ -4989,7 +4973,6 @@ dependencies = [ "magicblock-engine", "magicblock-engine-nucleus", "magicblock-ledger-deprecated", - "magicblock-metrics", "magicblock-program", "magicblock-replicator", "magicblock-runtime", @@ -5039,11 +5022,11 @@ name = "magicblock-verifier" version = "1.0.1" dependencies = [ "anyhow", + "magicblock-aperture", "magicblock-config", "magicblock-core", "magicblock-engine", "magicblock-engine-nucleus", - "magicblock-metrics", "magicblock-replicator", "magicblock-runtime", "mimalloc", diff --git a/Cargo.toml b/Cargo.toml index db059a596..38c1e1b60 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -12,7 +12,6 @@ members = [ "bins/validator-tui", "bins/verifier", "aperture", - "metrics", "chainsync", "programs/committor-interface", "programs/committor-program", @@ -62,7 +61,6 @@ unwrap_used = "warn" wildcard_dependencies = "deny" [workspace.dependencies] -metrics = { package = "magicblock-metrics", path = "metrics" } aperture = { package = "magicblock-aperture", path = "aperture" } chainsync = { package = "magicblock-chainsync", path = "chainsync" } committor-interface = { package = "magicblock-committor-interface", path = "programs/committor-interface" } diff --git a/aperture/Cargo.toml b/aperture/Cargo.toml index 438074aa1..34fb7545e 100644 --- a/aperture/Cargo.toml +++ b/aperture/Cargo.toml @@ -18,7 +18,7 @@ hyper-util = { workspace = true, features = ["http1", "http2", "server"] } futures = { workspace = true } libloading = { workspace = true } num_cpus = { workspace = true } -tokio = { workspace = true } +tokio = { workspace = true, features = ["rt-multi-thread"] } tokio-util = { workspace = true } # magicblock @@ -28,8 +28,8 @@ chainsync = { workspace = true } config = { workspace = true } mbv-core = { workspace = true } ledger-deprecated = { workspace = true } -metrics = { workspace = true } -nucleus = { workspace = true, features = ["runtime"] } +prometheus = { workspace = true } +nucleus = { workspace = true, features = ["metrics", "runtime"] } # solana agave-geyser-plugin-interface = { workspace = true } diff --git a/aperture/src/account.rs b/aperture/src/account.rs index 994de017e..847f3029c 100644 --- a/aperture/src/account.rs +++ b/aperture/src/account.rs @@ -1,7 +1,6 @@ use chainsync::{AccountProperty, ChainSync, ChainSyncAccount, Result as ChainSyncResult}; use engine::Engine; use json::{Deserialize, Serialize}; -use metrics::metrics::ENSURE_ACCOUNTS_TIME; use nucleus::runtime::TransactionView; use serde::{Deserializer, de::Error as _}; use solana_account::{AccountMode, AccountSharedData}; @@ -72,7 +71,6 @@ impl Accounts { fetches: &mut u64, ) -> RpcResult { let transaction = decode_transaction(transaction, encoding)?; - let _timer = ENSURE_ACCOUNTS_TIME.with_label_values(&["transaction"]).start_timer(); *fetches += self .chainsync .ensure_transaction_accounts(&transaction) @@ -92,7 +90,6 @@ impl Accounts { /// Account for successful synchronization before any later projection or admission error. async fn fetch(&self, pubkeys: &[Pubkey], fetches: &mut u64) -> ChainSyncResult<()> { - let _timer = ENSURE_ACCOUNTS_TIME.with_label_values(&["account"]).start_timer(); let requests = pubkeys.iter().map(|&pubkey| ChainSyncAccount { pubkey, property: AccountProperty::Readonly, diff --git a/aperture/src/lib.rs b/aperture/src/lib.rs index 9406d61c9..34a408426 100644 --- a/aperture/src/lib.rs +++ b/aperture/src/lib.rs @@ -126,3 +126,8 @@ mod requests; mod server; mod state; mod transaction; + +mod metrics; + +/// Standalone Prometheus scrape server for leaders and verifiers. +pub use server::metrics::MetricsService; diff --git a/aperture/src/metrics.rs b/aperture/src/metrics.rs new file mode 100644 index 000000000..7889cdbea --- /dev/null +++ b/aperture/src/metrics.rs @@ -0,0 +1,49 @@ +//! Event-driven aperture metrics registered in the default Prometheus registry. + +use nucleus::metrics::{ + HistogramOpts, HistogramTimer, HistogramVec, IntCounter, IntCounterVec, Opts, counter, + counter_vec, histogram_vec, +}; +use std::sync::OnceLock; + +static METRICS: OnceLock = OnceLock::new(); + +struct Metrics { + requests: IntCounterVec, + duration: HistogramVec, + skip_preflight: IntCounter, +} + +fn metrics() -> &'static Metrics { + METRICS.get_or_init(|| Metrics { + requests: counter_vec( + Opts::new("aperture_rpc_requests_total", "RPC requests by method"), + &["method"], + ), + duration: histogram_vec( + HistogramOpts::new( + "aperture_rpc_request_duration_seconds", + "HTTP RPC request duration in seconds", + ) + .buckets(vec![0.001, 0.01, 0.1, 1.0, 5.0, 10.0, 30.0, 60.0]), + &["method"], + ), + skip_preflight: counter( + Opts::new( + "aperture_transactions_skip_preflight_total", + "Transactions submitted with skipPreflight", + ), + 0, + ), + }) +} + +pub(crate) fn request(method: &str) { + metrics().requests.with_label_values(&[method]).inc(); +} +pub(crate) fn time_request(method: &str) -> HistogramTimer { + metrics().duration.with_label_values(&[method]).start_timer() +} +pub(crate) fn skip_preflight() { + metrics().skip_preflight.inc(); +} diff --git a/aperture/src/requests/http/dispatch.rs b/aperture/src/requests/http/dispatch.rs index b00eba052..57e15f720 100644 --- a/aperture/src/requests/http/dispatch.rs +++ b/aperture/src/requests/http/dispatch.rs @@ -1,8 +1,8 @@ use super::{ HandlerResult, RpcHandlers, get_blocks::BlockRange, get_token_accounts::TokenAccountAuthority, }; +use crate::metrics; use crate::{error::RpcError, requests::JsonHttpRequest}; -use ::metrics::metrics::{RPC_REQUEST_HANDLING_TIME, RPC_REQUESTS_COUNT}; impl RpcHandlers { pub(crate) async fn process( @@ -13,8 +13,8 @@ impl RpcHandlers { // Route the request to the correct handler based on the method name. use crate::requests::JsonRpcHttpMethod::*; let method = request.method.as_str(); - RPC_REQUESTS_COUNT.with_label_values(&[method]).inc(); - let _timer = RPC_REQUEST_HANDLING_TIME.with_label_values(&[method]).start_timer(); + metrics::request(method); + let _timer = metrics::time_request(method); match request.method { GetAccountInfo => self.get_account_info(request, fetches).await, diff --git a/aperture/src/requests/http/send_transaction.rs b/aperture/src/requests/http/send_transaction.rs index abaee44c8..85199ee9e 100644 --- a/aperture/src/requests/http/send_transaction.rs +++ b/aperture/src/requests/http/send_transaction.rs @@ -1,8 +1,8 @@ use super::{HandlerResult, RpcHandlers}; +use crate::metrics; use crate::requests::{ JsonHttpRequest as JsonRequest, params::SerdeSignature, payload::ResponsePayload, }; -use ::metrics::metrics::{TRANSACTION_PROCESSING_TIME, TRANSACTION_SKIP_PREFLIGHT}; use solana_rpc_client_api::config::RpcSendTransactionConfig; use solana_transaction_status::UiTransactionEncoding; impl RpcHandlers { @@ -11,7 +11,6 @@ impl RpcHandlers { request: &JsonRequest, fetches: &mut u64, ) -> HandlerResult { - let _timer = TRANSACTION_PROCESSING_TIME.start_timer(); let transaction_str = request.required::<&str>(0)?; let config = request.optional::(1)?.unwrap_or_default(); let encoding = config.encoding.unwrap_or(UiTransactionEncoding::Base58); @@ -21,7 +20,7 @@ impl RpcHandlers { let signature = transaction.signatures()[0]; if config.skip_preflight { - TRANSACTION_SKIP_PREFLIGHT.inc(); + metrics::skip_preflight(); self.engine.transaction(transaction)?.schedule().await?; } else { self.engine.transaction(transaction)?.execute().await??; diff --git a/aperture/src/server/metrics.rs b/aperture/src/server/metrics.rs new file mode 100644 index 000000000..727137156 --- /dev/null +++ b/aperture/src/server/metrics.rs @@ -0,0 +1,99 @@ +use std::net::SocketAddr; + +use http_body_util::{BodyExt, Empty, Full, combinators::BoxBody}; +use hyper::{ + Method, Request, Response, StatusCode, body::Bytes, server::conn::http1, service::service_fn, +}; +use hyper_util::rt::TokioIo; +use nucleus::shutdown::{ShutdownHandle, ShutdownReason}; +use prometheus::TextEncoder; +use tokio::{net::TcpListener, select, task::JoinSet}; +use tracing::{info, warn}; + +pub struct MetricsService { + listener: TcpListener, +} + +impl MetricsService { + pub async fn bind(addr: SocketAddr) -> std::io::Result { + let listener = TcpListener::bind(addr).await?; + Ok(Self { listener }) + } + + /// Serves metrics until its shutdown tier is cancelled. + pub async fn run(self, mut shutdown: ShutdownHandle) { + let reason = match serve(self.listener, &shutdown).await { + Err(error) => ShutdownReason::Error(Box::new(error)), + Ok(()) if shutdown.requested() => ShutdownReason::Signalled, + Ok(()) => ShutdownReason::Unexpected, + }; + shutdown.terminate(reason); + } +} + +async fn serve(listener: TcpListener, shutdown: &ShutdownHandle) -> std::io::Result<()> { + info!("Metrics server started"); + let mut connections = JoinSet::new(); + + let result = loop { + select!( + _ = shutdown.signalled() => { + break Ok(()); + } + _ = connections.join_next(), if !connections.is_empty() => {} + result = listener.accept() => { + let (stream, _) = match result { + Ok(connection) => connection, + Err(error) => break Err(error), + }; + let io = TokioIo::new(stream); + connections.spawn(async move { + let con = http1::Builder::new(); + let _ = con.serve_connection(io, service_fn(metrics_service_router)).await; + }); + } + ); + }; + + connections.shutdown().await; + info!("Metrics server shutdown"); + result +} + +async fn metrics_service_router( + req: Request, +) -> Result>, hyper::Error> { + // Record optional headers + let result = match (req.method(), req.uri().path()) { + (&Method::GET, "/metrics") => { + let metric_families = prometheus::gather(); + let metrics = + TextEncoder::new().encode_to_string(&metric_families).unwrap_or_else(|error| { + warn!(error = %error, "Failed to encode metrics"); + String::new() + }); + Ok(Response::new(full(metrics))) + } + _ => { + let mut not_found = Response::new(empty()); + *not_found.status_mut() = StatusCode::NOT_FOUND; + Ok(not_found) + } + }; + // We must consume the body fully to keep the connection alive. We + // iterate over all chunks and simply drop them. This prevents garbage + // data of previous requests from being stuck in connection buffer. + let mut body = req.into_body(); + while (body.frame().await).is_some() {} + + result +} + +fn full>(chunk: T) -> BoxBody { + Full::new(chunk.into()).map_err(|never| match never {}).boxed() +} + +fn empty() -> BoxBody { + let map_err = Empty::::new().map_err(|never| match never {}); + map_err.boxed() +} diff --git a/aperture/src/server/mod.rs b/aperture/src/server/mod.rs index 58f714279..99259ee19 100644 --- a/aperture/src/server/mod.rs +++ b/aperture/src/server/mod.rs @@ -1,2 +1,4 @@ pub(crate) mod http; pub(crate) mod websocket; + +pub(crate) mod metrics; diff --git a/aperture/src/server/websocket/dispatch.rs b/aperture/src/server/websocket/dispatch.rs index 9aacab8ac..2314a9978 100644 --- a/aperture/src/server/websocket/dispatch.rs +++ b/aperture/src/server/websocket/dispatch.rs @@ -1,6 +1,6 @@ use std::collections::HashMap; -use ::metrics::metrics::RPC_REQUESTS_COUNT; +use crate::metrics; use engine::Engine; use hyper::body::Bytes; use json::Serialize; @@ -48,7 +48,7 @@ impl WsDispatcher { /// Routes an incoming JSON-RPC request to the appropriate subscription handler. pub(crate) async fn dispatch(&mut self, request: &JsonWsRequest) -> RpcResult { use JsonRpcWsMethod::*; - RPC_REQUESTS_COUNT.with_label_values(&[request.method.as_str()]).inc(); + metrics::request(request.method.as_str()); match request.method { AccountSubscribe => self.account_subscribe(request), ProgramSubscribe => self.program_subscribe(request), diff --git a/bins/validator/Cargo.toml b/bins/validator/Cargo.toml index 9986482a0..a76712c5c 100644 --- a/bins/validator/Cargo.toml +++ b/bins/validator/Cargo.toml @@ -19,13 +19,12 @@ config = { workspace = true } mbv-core = { workspace = true } delegation-program-api = { workspace = true } ledger-deprecated = { workspace = true } -metrics = { workspace = true } program = { workspace = true } runtime = { workspace = true } services = { workspace = true } task-scheduler = { workspace = true } mimalloc = { workspace = true } -nucleus = { workspace = true } +nucleus = { workspace = true, features = ["metrics"] } num_cpus = { workspace = true } replicator = { workspace = true } solana-commitment-config = { workspace = true } diff --git a/bins/validator/src/leader.rs b/bins/validator/src/leader.rs index 4b1d22c30..972fb2777 100644 --- a/bins/validator/src/leader.rs +++ b/bins/validator/src/leader.rs @@ -3,7 +3,7 @@ use std::{ time::Duration, }; -use aperture::Aperture; +use aperture::{Aperture, MetricsService}; use chainsync::ChainSync; use committor::{ ComputeBudgetConfig, DEFAULT_ACTIONS_TIMEOUT, committor_processor::CommittorProcessor, @@ -12,7 +12,6 @@ use committor::{ }; use config::LeaderParams; use engine::Engine; -use metrics::MetricsService; use nucleus::{ metrics::EventTimer, shutdown::{Service, ShutdownManager, ShutdownReason}, diff --git a/bins/validator/src/magic_sys_adapter.rs b/bins/validator/src/magic_sys_adapter.rs index 12846d17a..07ad8cb1b 100644 --- a/bins/validator/src/magic_sys_adapter.rs +++ b/bins/validator/src/magic_sys_adapter.rs @@ -1,5 +1,6 @@ use std::{collections::HashMap, sync::Arc, time::Duration}; +use crate::metrics; use committor::{ committor_processor::CommittorProcessor, intent_engine::db::AccountsDbIntentBacklog, @@ -12,7 +13,6 @@ use mbv_core::{ intent::{MagicIntentBundle, types::CommittedAccount}, traits::MagicSys, }; -use metrics::metrics; use program::magic_sys::INTENT_TOO_LARGE_ERR; use solana_instruction::error::InstructionError; use solana_pubkey::Pubkey; @@ -78,7 +78,7 @@ impl MagicSys for MagicSysAdapter { let accounts: Vec<_> = commits.iter().map(|account| (account.pubkey, account.remote_slot)).collect(); - let _timer = metrics::start_fetch_commit_nonces_wait_timer(); + let _timer = metrics::time_nonce_wait(); let receiver = self.fetch_current_commit_nonces_sync(&accounts, min_context_slot); receiver .recv_timeout(Self::FETCH_TIMEOUT) diff --git a/bins/validator/src/main.rs b/bins/validator/src/main.rs index 65f37a098..435bb014b 100644 --- a/bins/validator/src/main.rs +++ b/bins/validator/src/main.rs @@ -119,3 +119,5 @@ fn print_info(message: impl std::fmt::Display) { info!("{message}"); } } + +mod metrics; diff --git a/bins/validator/src/metrics.rs b/bins/validator/src/metrics.rs new file mode 100644 index 000000000..89f729ab1 --- /dev/null +++ b/bins/validator/src/metrics.rs @@ -0,0 +1,26 @@ +//! Event-driven validator metrics registered in the default Prometheus registry. + +use nucleus::metrics::{Histogram, HistogramOpts, HistogramTimer, histogram}; +use std::sync::OnceLock; + +static METRICS: OnceLock = OnceLock::new(); + +struct Metrics { + nonce_wait: Histogram, +} + +fn metrics() -> &'static Metrics { + METRICS.get_or_init(|| Metrics { + nonce_wait: histogram( + HistogramOpts::new( + "validator_commit_nonce_wait_duration_seconds", + "Synchronous commit-nonce wait duration in seconds", + ) + .buckets(vec![0.001, 0.01, 0.1, 1.0, 5.0, 10.0, 30.0, 60.0]), + ), + }) +} + +pub(crate) fn time_nonce_wait() -> HistogramTimer { + metrics().nonce_wait.start_timer() +} diff --git a/bins/verifier/Cargo.toml b/bins/verifier/Cargo.toml index e6b8c41b5..08cc35d72 100644 --- a/bins/verifier/Cargo.toml +++ b/bins/verifier/Cargo.toml @@ -13,7 +13,7 @@ anyhow = { workspace = true } engine = { workspace = true } config = { workspace = true } mbv-core = { workspace = true } -metrics = { workspace = true } +aperture = { workspace = true } runtime = { workspace = true } mimalloc = { workspace = true } nucleus = { workspace = true } diff --git a/bins/verifier/src/main.rs b/bins/verifier/src/main.rs index 92f03c403..262ca8722 100644 --- a/bins/verifier/src/main.rs +++ b/bins/verifier/src/main.rs @@ -31,7 +31,7 @@ async fn run() -> Result { }; info!(config = ?config, "starting verifier"); - let metrics = metrics::MetricsService::bind(config.metrics.address.0) + let metrics = aperture::MetricsService::bind(config.metrics.address.0) .await .context("failed to bind metrics service")?; let mut metrics_shutdown = ShutdownManager::default(); diff --git a/committor/Cargo.toml b/committor/Cargo.toml index 932c0bfe1..97d9202a2 100644 --- a/committor/Cargo.toml +++ b/committor/Cargo.toml @@ -18,16 +18,14 @@ borsh = { workspace = true } engine = { workspace = true } futures-util = { workspace = true } lru = { workspace = true } - committor-interface = { workspace = true } mbv-core = { workspace = true } delegation-program-api = { workspace = true } -metrics = { workspace = true } program = { workspace = true } rpc-client = { workspace = true } parking_lot = { workspace = true } table-mania = { workspace = true } -nucleus = { workspace = true } +nucleus = { workspace = true, features = ["metrics"] } pin-project = { workspace = true } rusqlite = { workspace = true, features = ["fallible_uint"] } solana-account = { workspace = true } diff --git a/committor/src/intent_engine/db.rs b/committor/src/intent_engine/db.rs index 040fb9fee..6f1267849 100644 --- a/committor/src/intent_engine/db.rs +++ b/committor/src/intent_engine/db.rs @@ -1,8 +1,8 @@ use std::{cell::RefCell, collections::VecDeque}; +use crate::metrics; use engine::Engine; use mbv_core::intent::outbox::outbox_intent_pda; -use metrics::metrics; #[cfg(any(test, feature = "dev-context-only-utils"))] use parking_lot::Mutex; use program::outbox_intent_bundles::OutboxIntentBundle; @@ -35,20 +35,26 @@ impl AccountsDbIntentBacklog { } } +impl Drop for AccountsDbIntentBacklog { + fn drop(&mut self) { + metrics::backlog_removed(self.queue.get_mut().len()); + } +} + impl BacklogDB for AccountsDbIntentBacklog { fn store_intent_bundle(&self, intent_bundle: OutboxIntentBundle) -> DBResult<()> { let mut queue = self.queue.borrow_mut(); queue.push_back(intent_bundle.inner.intent_id); + metrics::backlog_added(1); - metrics::set_committor_intents_backlog_count(queue.len() as i64); Ok(()) } fn store_intent_bundles(&self, intent_bundles: Vec) -> DBResult<()> { let mut queue = self.queue.borrow_mut(); + metrics::backlog_added(intent_bundles.len()); queue.extend(intent_bundles.into_iter().map(|el| el.inner.intent_id)); - metrics::set_committor_intents_backlog_count(queue.len() as i64); Ok(()) } @@ -57,7 +63,7 @@ impl BacklogDB for AccountsDbIntentBacklog { let Some(id) = queue.pop_front() else { return Ok(None); }; - metrics::set_committor_intents_backlog_count(queue.len() as i64); + metrics::backlog_removed(1); drop(queue); let intent_pda = outbox_intent_pda(id); diff --git a/committor/src/intent_engine/intent_execution_engine.rs b/committor/src/intent_engine/intent_execution_engine.rs index da63c3ff6..262982b8b 100644 --- a/committor/src/intent_engine/intent_execution_engine.rs +++ b/committor/src/intent_engine/intent_execution_engine.rs @@ -5,9 +5,9 @@ use std::{ time::{Duration, Instant}, }; +use crate::metrics; use futures_util::stream::FuturesUnordered; use mbv_core::traits::CallbackScheduleError; -use metrics::metrics; use parking_lot::Mutex; use program::outbox_intent_bundles::OutboxIntentBundle; use solana_signature::Signature; @@ -186,17 +186,14 @@ where retries: self.retries_semaphore.clone(), }; - let handle = tokio::spawn(Self::execute( - executor_factory, - intent, - scheduler, - limits, - permit, - result_sender.clone(), - )); + let sender = result_sender.clone(); + let inflight = metrics::Inflight::new(); + let handle = tokio::spawn(async move { + let _inflight = inflight; + Self::execute(executor_factory, intent, scheduler, limits, permit, sender).await; + }); self.running_executors.push(handle); - metrics::set_committor_executors_busy_count(self.running_executors.len() as i64); } } @@ -304,7 +301,6 @@ where Vec::new() } }; - metrics::set_committor_poisoned_keys_count(scheduler.poisoned_keys_count() as i64); drop(scheduler); Self::broadcast_result( BroadcastedIntentExecutionResult::new(intent.intent_id, result), @@ -343,7 +339,7 @@ where poisoned_intents: Vec, result_sender: &broadcast::Sender, ) { - metrics::inc_committor_cascade_voided_intents_count_by(poisoned_intents.len() as u64); + metrics::voided(poisoned_intents.len()); for intent in poisoned_intents { warn!(poisoned_intent = ?intent.intent_id, "Intent poisoned"); Self::broadcast_result( @@ -429,17 +425,9 @@ where result: &IntentExecutorResult, ) { const EXECUTION_TIME_THRESHOLD: f64 = 5.0; - const INTENT_BUNDLE_LABEL: &str = "intent_bundle"; let intent_execution_secs = execution_time.as_secs_f64(); - metrics::observe_committor_intent_execution_time_histogram( - intent_execution_secs, - &INTENT_BUNDLE_LABEL, - result, - ); - if let Err(err) = &result { - metrics::inc_committor_failed_intents_count(&INTENT_BUNDLE_LABEL, err); - } + metrics::execution(intent_execution_secs, result); // Loki alerts if intent_execution_secs >= EXECUTION_TIME_THRESHOLD { diff --git a/committor/src/intent_engine/intent_scheduler.rs b/committor/src/intent_engine/intent_scheduler.rs index 62774fdcf..4f63b9654 100644 --- a/committor/src/intent_engine/intent_scheduler.rs +++ b/committor/src/intent_engine/intent_scheduler.rs @@ -1,3 +1,4 @@ +use crate::metrics; use std::collections::{BTreeSet, HashMap, HashSet, VecDeque, hash_map::Entry}; use program::outbox_intent_bundles::OutboxIntentBundle; @@ -84,7 +85,19 @@ pub(crate) struct IntentScheduler { poisoned_keys: HashSet, } +impl Drop for IntentScheduler { + fn drop(&mut self) { + metrics::poisoned_removed(self.poisoned_keys.len()); + } +} + impl IntentScheduler { + fn poison(&mut self, pubkeys: impl IntoIterator) { + let before = self.poisoned_keys.len(); + self.poisoned_keys.extend(pubkeys); + metrics::poisoned_added(self.poisoned_keys.len() - before); + } + pub(crate) fn new() -> Self { Self { blocked_keys: HashMap::new(), @@ -126,7 +139,7 @@ impl IntentScheduler { pubkeys = ?pubkeys, "Intent got poisoned" ); - self.poisoned_keys.extend(pubkeys); + self.poison(pubkeys); return None; } @@ -432,7 +445,7 @@ impl IntentScheduler { // Poison intents let mut worklist = BTreeSet::new(); for pubkey in pubkeys { - self.poisoned_keys.insert(pubkey); + self.poison([pubkey]); let Some(queue) = self.blocked_keys.remove(&pubkey) else { return Err(IntentSchedulerError::CorruptedIntentError); }; @@ -463,7 +476,7 @@ impl IntentScheduler { } } - self.poisoned_keys.extend(pubkeys); + self.poison(pubkeys); poisoned.push(meta.intent); } @@ -515,10 +528,6 @@ impl IntentScheduler { pub(crate) fn intents_blocked(&self) -> usize { self.blocked_intents.len() } - - pub(crate) fn poisoned_keys_count(&self) -> usize { - self.poisoned_keys.len() - } } #[derive(Error, Debug)] diff --git a/committor/src/intent_executor/accepted_intent_executor.rs b/committor/src/intent_executor/accepted_intent_executor.rs index 1d00f9c63..7f3838f98 100644 --- a/committor/src/intent_executor/accepted_intent_executor.rs +++ b/committor/src/intent_executor/accepted_intent_executor.rs @@ -206,13 +206,6 @@ where } } - // Gather metrics in separate task - let intent_client = self.ctx.intent_client.clone(); - let result = result.inspect(|output| { - let output_copy = *output; - tokio::spawn(async move { intent_client.intent_metrics(output_copy).await }); - }); - let close_buffers = result.is_ok(); let junk = execution_report.junk; let result = IntentExecutionResult { diff --git a/committor/src/intent_executor/error.rs b/committor/src/intent_executor/error.rs index 8d0ae85ff..81a562917 100644 --- a/committor/src/intent_executor/error.rs +++ b/committor/src/intent_executor/error.rs @@ -1,5 +1,4 @@ use mbv_core::traits::ActionError; -use metrics::metrics; use rpc_client::MagicBlockRpcClientError; use solana_message::CompileError; use solana_signature::Signature; @@ -181,15 +180,15 @@ impl IntentExecutorError { } } -impl metrics::LabelValue for IntentExecutorError { - fn value(&self) -> &str { +impl IntentExecutorError { + pub(crate) fn metric_label(&self) -> &'static str { match self { - IntentExecutorError::FailedToCommitError { err, signature: _ } => err.value(), + IntentExecutorError::FailedToCommitError { err, signature: _ } => err.metric_label(), IntentExecutorError::FailedToFinalizeError { err, commit_signature: _, finalize_signature: _, - } => err.value(), + } => err.metric_label(), _ => "failed", } } diff --git a/committor/src/intent_executor/intent_execution_client.rs b/committor/src/intent_executor/intent_execution_client.rs index f5a72619e..750ccb4e5 100644 --- a/committor/src/intent_executor/intent_execution_client.rs +++ b/committor/src/intent_executor/intent_execution_client.rs @@ -1,7 +1,7 @@ use std::{ops::ControlFlow, time::Duration}; +use crate::metrics; use delegation_program_api::pda::commit_record_pda_from_delegated_account; -use metrics::metrics; use rpc_client::{ MagicBlockRpcClientError, MagicBlockRpcClientResult, MagicBlockSendTransactionConfig, MagicBlockSendTransactionOutcome, MagicblockRpcClient, @@ -15,7 +15,6 @@ use solana_hash::Hash; use solana_keypair::Keypair; use solana_message::VersionedMessage; use solana_pubkey::Pubkey; -use solana_rpc_client_api::config::RpcTransactionConfig; use solana_signature::Signature; use solana_signer::Signer; use solana_transaction::versioned::VersionedTransaction; @@ -24,7 +23,6 @@ use tracing::warn; use crate::{ intent_executor::{ - ExecutionOutput, error::{IntentExecutorError, IntentExecutorResult, InternalError}, strategy_executor::error::{ IntentTransactionErrorMapper, TransactionStrategyExecutionError, @@ -194,7 +192,7 @@ impl IntentExecutionClient { signatures: &[Signature], commitment_config: CommitmentConfig, ) -> MagicBlockRpcClientResult>>> { - let _timer = metrics::start_rpc_client_signature_history_timer(); + let _timer = metrics::time_history(); let response = self .rpc_client .get_inner() @@ -275,51 +273,6 @@ impl IntentExecutionClient { Ok(result) } - - pub async fn intent_metrics(&self, execution_outcome: ExecutionOutput) { - use solana_transaction_status_client_types::EncodedTransactionWithStatusMeta; - fn extract_cu(tx: EncodedTransactionWithStatusMeta) -> Option { - let cu = tx.meta?.compute_units_consumed; - cu.into() - } - - let config = RpcTransactionConfig { - commitment: Some(self.rpc_client.commitment()), - max_supported_transaction_version: Some(1), - ..Default::default() - }; - let cu_metrics = || async { - match execution_outcome { - ExecutionOutput::SingleStage(signature) => { - let tx = self.rpc_client.get_transaction(&signature, Some(config)).await?; - Ok::<_, MagicBlockRpcClientError>(extract_cu(tx.transaction)) - } - ExecutionOutput::TwoStage { - commit_signature, - finalize_signature, - } => { - let commit_tx = - self.rpc_client.get_transaction(&commit_signature, Some(config)).await?; - let finalize_tx = - self.rpc_client.get_transaction(&finalize_signature, Some(config)).await?; - let commit_cu = extract_cu(commit_tx.transaction); - let finalize_cu = extract_cu(finalize_tx.transaction); - let (Some(commit_cu), Some(finalize_cu)) = (commit_cu, finalize_cu) else { - return Ok(None); - }; - Ok(Some(commit_cu + finalize_cu)) - } - } - }; - - match cu_metrics().await { - Ok(Some(cu)) => { - metrics::set_commmittor_intent_cu_usage(i64::try_from(cu).unwrap_or(i64::MAX)) - } - Err(err) => warn!(error = ?err, "Failed to fetch CUs for intent"), - _ => {} - } - } } struct IntentErrorMapper { diff --git a/committor/src/intent_executor/mod.rs b/committor/src/intent_executor/mod.rs index beceb4cca..0c44fab8f 100644 --- a/committor/src/intent_executor/mod.rs +++ b/committor/src/intent_executor/mod.rs @@ -12,7 +12,6 @@ use std::{sync::Arc, time::Duration}; use async_trait::async_trait; use mbv_core::traits::{ActionsCallbackScheduler, CallbackScheduleError}; -use metrics::metrics; use program::{ magic_scheduled_base_intent::ScheduledIntentBundle, outbox::ExecutionStage, outbox_intent_bundles::OutboxIntentBundleStatus, @@ -101,8 +100,8 @@ pub enum ExecutionOutput { }, } -impl metrics::LabelValue for ExecutionOutput { - fn value(&self) -> &str { +impl ExecutionOutput { + pub(crate) fn metric_label(&self) -> &'static str { match self { Self::SingleStage(_) => "single_stage_succeeded", Self::TwoStage { diff --git a/committor/src/intent_executor/single_stage_intent_executor.rs b/committor/src/intent_executor/single_stage_intent_executor.rs index 897b7fb2a..953ea2c81 100644 --- a/committor/src/intent_executor/single_stage_intent_executor.rs +++ b/committor/src/intent_executor/single_stage_intent_executor.rs @@ -145,13 +145,6 @@ where } } - // Gather metrics in separate task - let intent_client = self.ctx.intent_client.clone(); - let result = result.inspect(|output| { - let output_copy = *output; - tokio::spawn(async move { intent_client.intent_metrics(output_copy).await }); - }); - let close_buffers = result.is_ok(); let junk = execution_report.junk; let result = IntentExecutionResult { diff --git a/committor/src/intent_executor/strategy_executor/error.rs b/committor/src/intent_executor/strategy_executor/error.rs index bf7104600..f01b08355 100644 --- a/committor/src/intent_executor/strategy_executor/error.rs +++ b/committor/src/intent_executor/strategy_executor/error.rs @@ -1,5 +1,4 @@ use mbv_core::traits::ActionError; -use metrics::metrics; use rpc_client::{MagicBlockRpcClientError, utils::TransactionErrorMapper}; use solana_instruction::error::InstructionError; use solana_signature::Signature; @@ -177,8 +176,8 @@ impl TransactionStrategyExecutionError { } } -impl metrics::LabelValue for TransactionStrategyExecutionError { - fn value(&self) -> &str { +impl TransactionStrategyExecutionError { + pub(crate) fn metric_label(&self) -> &'static str { match self { Self::ActionsError(_, _) => "actions_failed", Self::CpiLimitError(_, _) => "cpi_limit_failed", diff --git a/committor/src/lib.rs b/committor/src/lib.rs index 77052060c..5c190e0d1 100644 --- a/committor/src/lib.rs +++ b/committor/src/lib.rs @@ -18,3 +18,4 @@ pub(crate) mod utils; pub use committor_interface::{ChangedAccount, Changeset, ChangesetMeta}; pub use compute_budget::ComputeBudgetConfig; pub use config::DEFAULT_ACTIONS_TIMEOUT; +mod metrics; diff --git a/committor/src/metrics.rs b/committor/src/metrics.rs new file mode 100644 index 000000000..8990d5b38 --- /dev/null +++ b/committor/src/metrics.rs @@ -0,0 +1,190 @@ +//! Event-driven committor metrics registered in the default Prometheus registry. + +use crate::{ + intent_executor::{ExecutionOutput, error::IntentExecutorError}, + tasks::BaseTaskImpl, +}; +use nucleus::metrics::{ + Histogram, HistogramOpts, HistogramTimer, HistogramVec, IntCounter, IntCounterVec, IntGauge, + Opts, counter, counter_vec, gauge, gauge_value, histogram, histogram_vec, +}; +use std::sync::OnceLock; + +const DURATION_BUCKETS: [f64; 8] = [0.001, 0.01, 0.1, 1.0, 5.0, 10.0, 30.0, 60.0]; + +static METRICS: OnceLock = OnceLock::new(); + +struct Metrics { + scheduled: IntCounter, + backlog: IntGauge, + inflight: IntGauge, + poisoned: IntGauge, + voided: IntCounter, + failures: IntCounterVec, + execution: HistogramVec, + task_preparation: HistogramVec, + alt_preparation: Histogram, + alt_count: Histogram, + account_fetches: IntCounter, + retiring: IntGauge, + history: Histogram, +} + +fn metrics() -> &'static Metrics { + METRICS.get_or_init(|| Metrics { + scheduled: counter( + Opts::new( + "committor_scheduled_intents_total", + "Live intents submitted for execution", + ), + 0, + ), + backlog: gauge( + Opts::new( + "committor_backlog_intents", + "Intents queued in the account-backed backlog", + ), + 0, + ), + inflight: gauge( + Opts::new( + "committor_inflight_intents", + "Running intent tasks including retries", + ), + 0, + ), + poisoned: gauge( + Opts::new( + "committor_poisoned_keys", + "Keys poisoned in the intent scheduler", + ), + 0, + ), + voided: counter( + Opts::new( + "committor_voided_intents_total", + "Intents voided by a poisoning cascade", + ), + 0, + ), + failures: counter_vec( + Opts::new("committor_failed_intents_total", "Terminal intent failures"), + &["error"], + ), + execution: histogram_vec( + HistogramOpts::new( + "committor_intent_execution_duration_seconds", + "Intent execution duration including retries in seconds", + ) + .buckets(DURATION_BUCKETS.to_vec()), + &["outcome"], + ), + task_preparation: histogram_vec( + HistogramOpts::new( + "committor_task_preparation_duration_seconds", + "Task preparation duration in seconds", + ) + .buckets(DURATION_BUCKETS.to_vec()), + &["task"], + ), + alt_preparation: histogram( + HistogramOpts::new( + "committor_lookup_table_preparation_duration_seconds", + "Lookup-table preparation duration in seconds", + ) + .buckets(DURATION_BUCKETS.to_vec()), + ), + alt_count: histogram( + HistogramOpts::new( + "committor_transaction_lookup_tables", + "Lookup tables per prepared transaction when present", + ) + .buckets(vec![0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 10.0]), + ), + account_fetches: counter( + Opts::new( + "committor_account_fetch_attempts_total", + "Task account RPC fetch attempts", + ), + 0, + ), + retiring: gauge( + Opts::new( + "committor_retiring_nonce_locks", + "Evicted nonce locks still held by active operations", + ), + 0, + ), + history: histogram( + HistogramOpts::new( + "committor_signature_history_query_duration_seconds", + "Recovery signature-history query duration in seconds", + ) + .buckets(DURATION_BUCKETS.to_vec()), + ), + }) +} + +pub(crate) fn scheduled(count: usize) { + metrics().scheduled.inc_by(count as u64); +} +pub(crate) fn backlog_added(count: usize) { + metrics().backlog.add(gauge_value(count)); +} +pub(crate) fn backlog_removed(count: usize) { + metrics().backlog.sub(gauge_value(count)); +} +pub(crate) fn poisoned_added(count: usize) { + metrics().poisoned.add(gauge_value(count)); +} +pub(crate) fn poisoned_removed(count: usize) { + metrics().poisoned.sub(gauge_value(count)); +} +pub(crate) fn voided(count: usize) { + metrics().voided.inc_by(count as u64); +} +pub(crate) fn execution(seconds: f64, result: &Result) { + let outcome = match result { + Ok(output) => output.metric_label(), + Err(_) => "failed", + }; + metrics().execution.with_label_values(&[outcome]).observe(seconds); + if let Err(error) = result { + metrics().failures.with_label_values(&[error.metric_label()]).inc(); + } +} +pub(crate) fn time_task(task: &BaseTaskImpl) -> HistogramTimer { + metrics() + .task_preparation + .with_label_values(&[task.metric_label()]) + .start_timer() +} +pub(crate) fn time_alt() -> HistogramTimer { + metrics().alt_preparation.start_timer() +} +pub(crate) fn lookup_tables(count: usize) { + metrics().alt_count.observe(count as f64); +} +pub(crate) fn account_fetch() { + metrics().account_fetches.inc(); +} +pub(crate) fn retiring(before: usize, after: usize) { + metrics().retiring.add(gauge_value(after) - gauge_value(before)); +} +pub(crate) fn time_history() -> HistogramTimer { + metrics().history.start_timer() +} + +/// Tracks a running intent through completion, panic, or cancellation. +pub(crate) struct Inflight; +impl Inflight { + pub(crate) fn new() -> Self { + metrics().inflight.inc(); + Self + } +} +impl Drop for Inflight { + fn drop(&mut self) { + metrics().inflight.dec(); + } +} diff --git a/committor/src/service.rs b/committor/src/service.rs index 700cd9d8b..2187198ac 100644 --- a/committor/src/service.rs +++ b/committor/src/service.rs @@ -1,8 +1,8 @@ use std::{sync::Arc, time::Duration}; +use crate::metrics; use engine::Engine; use mbv_core::intent::outbox::outbox_intent_pda_with_bump; -use metrics::metrics; use nucleus::shutdown::{ShutdownHandle, ShutdownReason}; use program::outbox_intent_bundles::OutboxIntentBundle; use solana_account::AccountMode; @@ -153,7 +153,7 @@ where return Ok(()); } - metrics::inc_committor_intents_count_by(intent_bundles.len() as u64); + metrics::scheduled(intent_bundles.len()); self.processor.schedule_intent_bundles(intent_bundles).await } diff --git a/committor/src/tasks/mod.rs b/committor/src/tasks/mod.rs index 50c6b07c4..c5d689170 100644 --- a/committor/src/tasks/mod.rs +++ b/committor/src/tasks/mod.rs @@ -1,4 +1,3 @@ -use ::metrics::metrics::LabelValue; use delegation_program_api::{AccountSizeClass, args::CallHandlerArgs}; use mbv_core::intent::{BaseAction, BaseActionCallback}; use solana_instruction::{AccountMeta, Instruction}; @@ -67,8 +66,8 @@ impl BaseTask for BaseTaskImpl { } } -impl LabelValue for BaseTaskImpl { - fn value(&self) -> &str { +impl BaseTaskImpl { + pub(crate) fn metric_label(&self) -> &'static str { match self { Self::CommitFinalize(task) => { if task.is_buffer() { diff --git a/committor/src/tasks/task_info_fetcher.rs b/committor/src/tasks/task_info_fetcher.rs index 8032aa7a4..157bff0f1 100644 --- a/committor/src/tasks/task_info_fetcher.rs +++ b/committor/src/tasks/task_info_fetcher.rs @@ -1,5 +1,6 @@ use std::{collections::HashMap, mem, num::NonZeroUsize, sync::Arc, time::Duration}; +use crate::metrics; use async_trait::async_trait; use delegation_program_api::{ delegation_metadata_seeds_from_delegated_account, @@ -7,7 +8,6 @@ use delegation_program_api::{ state::{DelegationMetadata, DelegationRecord}, }; use lru::LruCache; -use metrics::metrics; use parking_lot::Mutex; use rpc_client::{MagicBlockRpcClientError, MagicblockRpcClient}; use solana_account::Account; @@ -205,7 +205,7 @@ impl RpcTaskInfoFetcher { return Ok(Vec::new()); } - metrics::inc_task_info_fetcher_a_count(); + metrics::account_fetch(); let commitment = rpc_client.commitment(); let mut accounts = rpc_client .get_multiple_accounts_with_config( @@ -330,6 +330,12 @@ struct CacheInner { retiring: HashMap, } +impl Drop for CacheInner { + fn drop(&mut self) { + metrics::retiring(self.retiring.len(), 0); + } +} + impl CacheInner { fn new(capacity: NonZeroUsize) -> Self { Self { @@ -365,6 +371,7 @@ impl<'a> CacheInnerGuard<'a> { impl<'a> Drop for CacheInnerGuard<'a> { fn drop(&mut self) { let mut inner = self.inner.lock(); + let retiring = inner.retiring.len(); let nonce_locks = mem::take(&mut self.nonce_locks); for (pubkey, lock) in nonce_locks { // Drop our clone first so strong_count reflects only other @@ -377,7 +384,7 @@ impl<'a> Drop for CacheInnerGuard<'a> { } } - metrics::set_task_info_fetcher_retiring_count(inner.retiring.len() as i64); + metrics::retiring(retiring, inner.retiring.len()); } } @@ -425,6 +432,7 @@ impl CacheTaskInfoFetcher { /// Resets cache for some or all accounts pub fn reset(&self, reset_type: ResetType<'_>) { let mut cache = self.cache.lock(); + let retiring = cache.retiring.len(); match reset_type { ResetType::All => { cache.active.clear(); @@ -437,6 +445,7 @@ impl CacheTaskInfoFetcher { } } } + metrics::retiring(retiring, cache.retiring.len()); } /// Ensures a [`NonceLock`] exists in the cache for each pubkey and returns @@ -451,6 +460,7 @@ impl CacheTaskInfoFetcher { let mut nonce_locks = vec![]; { let mut inner = self.cache.lock(); + let retiring = inner.retiring.len(); for pubkey in pubkeys { let (lock, evicted) = if let Some(val) = inner.active.get(&pubkey) { (val.clone(), None) @@ -492,7 +502,7 @@ impl CacheTaskInfoFetcher { nonce_locks.push((pubkey, lock)); } - metrics::set_task_info_fetcher_retiring_count(inner.retiring.len() as i64); + metrics::retiring(retiring, inner.retiring.len()); } CacheInnerGuard { inner: &self.cache, nonce_locks } diff --git a/committor/src/transaction_preparator/delivery_preparator.rs b/committor/src/transaction_preparator/delivery_preparator.rs index 8d35ec7e9..8cea0d3a2 100644 --- a/committor/src/transaction_preparator/delivery_preparator.rs +++ b/committor/src/transaction_preparator/delivery_preparator.rs @@ -1,8 +1,8 @@ use std::{collections::HashSet, ops::ControlFlow, time::Duration}; +use crate::metrics; use committor_interface::{Chunks, instruction_chunks::chunk_realloc_ixs}; use futures_util::future::{join, join_all, try_join_all}; -use metrics::metrics; use rpc_client::{ MagicBlockRpcClientError, MagicBlockSendTransactionConfig, MagicBlockSendTransactionOutcome, MagicblockRpcClient, @@ -62,13 +62,13 @@ impl DeliveryPreparator { ) -> DeliveryPreparatorResult> { let uniqueness_nonce = strategy.uniqueness_nonce; let preparation_futures = strategy.optimized_tasks.iter_mut().map(|task| async move { - let _timer = metrics::observe_committor_intent_task_preparation_time(&*task); + let _timer = metrics::time_task(task); self.prepare_task_handling_errors(authority, task, uniqueness_nonce).await }); let task_preparations = join_all(preparation_futures); let alts_preparations = async { - let _timer = metrics::observe_committor_intent_alt_preparation_time(); + let _timer = metrics::time_alt(); self.prepare_lookup_tables(authority, &strategy.lookup_tables_keys).await }; diff --git a/committor/src/transaction_preparator/mod.rs b/committor/src/transaction_preparator/mod.rs index 6eb022859..cc4bf6a4c 100644 --- a/committor/src/transaction_preparator/mod.rs +++ b/committor/src/transaction_preparator/mod.rs @@ -1,5 +1,5 @@ +use crate::metrics; use async_trait::async_trait; -use metrics::metrics; use rpc_client::MagicblockRpcClient; use solana_keypair::Keypair; use solana_message::VersionedMessage; @@ -83,7 +83,7 @@ impl TransactionPreparator for TransactionPreparatorImpl { tx_strategy.uniqueness_nonce, )?; self.delivery_preparator.prepare_for_delivery(authority, tx_strategy).await?; - metrics::observe_committor_intent_alt_count(0); + metrics::lookup_tables(0); return Ok( TransactionUtils::assemble_tasks_v1_tx_with_uniqueness_nonce( authority, @@ -112,7 +112,7 @@ impl TransactionPreparator for TransactionPreparatorImpl { // Pre tx preparations. Create buffer accs + lookup tables let lookup_tables = self.delivery_preparator.prepare_for_delivery(authority, tx_strategy).await?; - metrics::observe_committor_intent_alt_count(lookup_tables.len()); + metrics::lookup_tables(lookup_tables.len()); let message = TransactionUtils::assemble_tasks_tx_with_uniqueness_nonce( authority, diff --git a/config/src/config/metrics.rs b/config/src/config/metrics.rs index 06456acf3..6a9e2d816 100644 --- a/config/src/config/metrics.rs +++ b/config/src/config/metrics.rs @@ -1,5 +1,3 @@ -use std::time::Duration; - use serde::{Deserialize, Serialize}; use crate::{consts, types::BindAddress}; @@ -9,9 +7,6 @@ use crate::{consts, types::BindAddress}; pub struct MetricsConfig { /// Address at which the system metrics are exposed pub address: BindAddress, - /// Time frequency with which the metrics are collected - #[serde(with = "humantime")] - pub collect_frequency: Duration, } // The default bind address is a fixed, validated constant. @@ -20,7 +15,6 @@ impl Default for MetricsConfig { fn default() -> Self { Self { address: consts::DEFAULT_METRICS_ADDR.parse().unwrap(), - collect_frequency: Duration::from_secs(consts::DEFAULT_METRICS_COLLECT_FREQUENCY_SEC), } } } diff --git a/config/src/consts.rs b/config/src/consts.rs index cf3b701a3..27fd838fa 100644 --- a/config/src/consts.rs +++ b/config/src/consts.rs @@ -52,6 +52,3 @@ pub const DEFAULT_LEDGER_BLOCK_CACHE_SIZE: u64 = 512 * 1024 * 1024; /// Metrics Defaults /// Default address for the metrics endpoint (Prometheus format) pub const DEFAULT_METRICS_ADDR: &str = "0.0.0.0:9000"; - -/// Default frequency of metrics collection in seconds -pub const DEFAULT_METRICS_COLLECT_FREQUENCY_SEC: u64 = 30; diff --git a/metrics/Cargo.toml b/metrics/Cargo.toml deleted file mode 100644 index 284ffc3d8..000000000 --- a/metrics/Cargo.toml +++ /dev/null @@ -1,26 +0,0 @@ -[package] -authors.workspace = true -edition.workspace = true -homepage.workspace = true -license.workspace = true -name = "magicblock-metrics" -repository.workspace = true -version.workspace = true - -[features] -default = [] -dev-context = [] - -[dependencies] -http-body-util = { workspace = true } -hyper = { workspace = true, features = ["http1", "server"] } -hyper-util = { workspace = true, features = ["tokio"] } -lazy_static = { workspace = true } -nucleus = { workspace = true } -prometheus = { workspace = true } -solana-signature = { workspace = true } -tokio = { workspace = true, features = ["macros", "net", "rt-multi-thread"] } -tracing = { workspace = true } - -[lints] -workspace = true diff --git a/metrics/README.md b/metrics/README.md deleted file mode 100644 index 2ccaf4d16..000000000 --- a/metrics/README.md +++ /dev/null @@ -1,28 +0,0 @@ -# magicblock-metrics - -Provides Prometheus metrics for the validator and serves them alongside Engine -metrics at `/metrics`. Leaders and verifiers each have their own endpoint. - -## Scraping metrics - -Set `[metrics].address` in the process's -[configuration](../config/README.md), then add that address to -Prometheus. For example, if the listener is `127.0.0.1:9090`: - -```yaml -scrape_configs: - - job_name: magicblock-validator - static_configs: - - targets: ["127.0.0.1:9090"] -``` - -Use distinct addresses when running multiple processes on one host. A reachable -metrics endpoint does not necessarily mean a verifier has caught up. - -## Adding metrics - -Keep names, units, and labels useful to operators. Avoid unbounded labels such -as account keys or transaction signatures. Dashboard deployment is separate -from this crate. - -[Back to workspace](../README.md) diff --git a/metrics/src/lib.rs b/metrics/src/lib.rs deleted file mode 100644 index a7f626d84..000000000 --- a/metrics/src/lib.rs +++ /dev/null @@ -1,7 +0,0 @@ -#![doc = include_str!("../README.md")] -#![recursion_limit = "256"] - -pub mod metrics; -mod service; - -pub use service::MetricsService; diff --git a/metrics/src/metrics/mod.rs b/metrics/src/metrics/mod.rs deleted file mode 100644 index 2bf9bfd44..000000000 --- a/metrics/src/metrics/mod.rs +++ /dev/null @@ -1,1246 +0,0 @@ -// Fixed metric definitions are checked when initialized; registration cannot -// recover from an invalid name or label set. -#![allow(clippy::unwrap_used, clippy::panic)] - -use std::sync::Once; - -use prometheus::{ - Histogram, HistogramOpts, HistogramTimer, HistogramVec, IntCounter, IntCounterVec, IntGauge, - IntGaugeVec, Opts, Registry, -}; -pub use types::{ - AccountFetchContext, AccountFetchEntrypoint, AccountFetchReason, ChainlinkCloneIntent, - ChainlinkCloneOutcome, ChainlinkCloneRemoteResult, ChainlinkCompanionFetchKind, - ChainlinkCompanionFetchOutcome, ChainlinkEmptyPlaceholderStage, ChainlinkPendingFetchLayer, - ChainlinkPendingFetchOutcome, LabelValue, Outcome, SubscriptionCleanupOutcome, - SubscriptionCleanupSource, SubscriptionReasonLabel, SubscriptionRegistrationOrigin, - SubscriptionRegistrationOutcome, SubscriptionReleaseOutcome, -}; - -mod types; - -// ----------------- -// Buckets -// ----------------- -// Prometheus collects durations in seconds -const MICROS_10_90: [f64; 9] = [ - 0.000_01, 0.000_02, 0.000_03, 0.000_04, 0.000_05, 0.000_06, 0.000_07, 0.000_08, 0.000_09, -]; -const MICROS_100_900: [f64; 9] = - [0.000_1, 0.000_2, 0.000_3, 0.000_4, 0.000_5, 0.000_6, 0.000_7, 0.000_8, 0.000_9]; -const MILLIS_1_9: [f64; 9] = [0.001, 0.002, 0.003, 0.004, 0.005, 0.006, 0.007, 0.008, 0.009]; -const MILLIS_10_90: [f64; 9] = [0.01, 0.02, 0.03, 0.04, 0.05, 0.06, 0.07, 0.08, 0.09]; -const MILLIS_100_900: [f64; 9] = [0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9]; -const SECONDS_1_9: [f64; 9] = [1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0]; - -lazy_static::lazy_static! { - pub (crate) static ref REGISTRY: Registry = Registry::new_custom(Some("mbv".to_string()), None).unwrap(); - - // Needs to be a gauge so we can set it directly - // (instead of having to calcuate by which to inc it) - static ref CHAIN_SLOT_GAUGE: IntGauge = IntGauge::new( - "chain_slot_gauge", "Chain Slot Gauge", - ).unwrap(); -} - -lazy_static::lazy_static! { - - // ----------------- - // Accounts - // ----------------- - static ref MONITORED_ACCOUNTS_GAUGE: IntGauge = IntGauge::new( - "monitored_accounts_gauge", "number of undelegated accounts, being monitored via websocket", - ).unwrap(); - - static ref INFLIGHT_SUBSCRIPTION_UPDATES_GAUGE: IntGauge = IntGauge::new( - "inflight_subscription_updates_gauge", "Number of subscription-update tasks currently being processed", - ).unwrap_or_else(|err| { - panic!("failed to create inflight_subscription_updates_gauge: {err}") - }); - - static ref CHAINLINK_SUBSCRIPTION_REGISTRATION_ACCOUNTS_TOTAL: IntCounterVec = - IntCounterVec::new( - Opts::new( - "chainlink_subscription_registration_accounts_total", - "Account subscription registration attempts by origin, reason, and terminal outcome", - ), - &["entrypoint", "fetch_reason", "subscription_reason", "outcome"], - ) - .unwrap(); - - static ref CHAINLINK_SUBSCRIPTION_RELEASE_ACCOUNTS_TOTAL: IntCounterVec = - IntCounterVec::new( - Opts::new( - "chainlink_subscription_release_accounts_total", - "Account subscription release attempts by reason and terminal outcome", - ), - &["reason", "outcome"], - ) - .unwrap(); - - static ref CHAINLINK_SUBSCRIPTION_CLEANUP_ACCOUNTS_TOTAL: IntCounterVec = - IntCounterVec::new( - Opts::new( - "chainlink_subscription_cleanup_accounts_total", - "Account subscription cleanup actions by cleanup source and terminal outcome", - ), - &["cleanup_source", "outcome"], - ) - .unwrap(); - static ref PROGRAM_SUBSCRIPTION_DISCOVERED_DLP_UPDATE_DELEGATED_ELSEWHERE_COUNT: IntCounter = IntCounter::new( - "program_subscription_discovered_dlp_update_delegated_elsewhere_count", "DLP-owned subscription updates that, after fetching the delegation record, were found delegated to another validator and dropped", - ).unwrap(); - - static ref PROGRAM_SUBSCRIPTION_ACCOUNT_UPDATES_COUNT: IntCounterVec = - IntCounterVec::new( - Opts::new( - "program_subscription_account_updates_count", - "Number of account updates received via program subscription matching an existing subscription", - ), - &["client_id"], - ) - .unwrap(); - - static ref ACCOUNT_SUBSCRIPTION_ACCOUNT_UPDATES_COUNT: IntCounterVec = - IntCounterVec::new( - Opts::new( - "account_subscription_account_updates_count", - "Number of account updates received via account subscription", - ), - &["client_id"], - ) - .unwrap(); - - static ref CHAINLINK_PENDING_FETCH_ACCOUNTS_TOTAL: IntCounterVec = - IntCounterVec::new( - Opts::new( - "chainlink_pending_fetch_accounts_total", - "Account fetch/clone requests by origin, pending-dedup layer, and owner/waiter outcome", - ), - &["entrypoint", "fetch_reason", "layer", "outcome"], - ) - .unwrap(); - - static ref CHAINLINK_PENDING_FETCH_WAITERS_TOTAL: IntCounterVec = - IntCounterVec::new( - Opts::new( - "chainlink_pending_fetch_waiters_total", - "Account fetch/clone requests that joined existing pending work by origin and pending-dedup layer", - ), - &["entrypoint", "fetch_reason", "layer"], - ) - .unwrap(); - - static ref CHAINLINK_PENDING_FETCH_WAITERS_GAUGE: IntGaugeVec = - IntGaugeVec::new( - Opts::new( - "chainlink_pending_fetch_waiters_gauge", - "Currently active account fetch/clone waiters by pending-dedup layer", - ), - &["layer"], - ) - .unwrap(); - - static ref CHAINLINK_PENDING_FETCH_OWNER_DURATION_SECONDS: HistogramVec = - HistogramVec::new( - HistogramOpts::new( - "chainlink_pending_fetch_owner_duration_seconds", - "Time spent by pending fetch/clone owners by origin, pending-dedup layer, and terminal owner outcome", - ) - .buckets( - MICROS_100_900.iter().chain( - MILLIS_1_9.iter()).chain( - MILLIS_10_90.iter()).chain( - MILLIS_100_900.iter()).chain( - SECONDS_1_9.iter()).cloned().collect() - ), - &["entrypoint", "fetch_reason", "layer", "outcome"], -) - .unwrap(); - - static ref CHAINLINK_COMPANION_FETCH_ATTEMPTS: HistogramVec = - HistogramVec::new( - HistogramOpts::new( - "chainlink_companion_fetch_attempts", - "Number of fetch attempts needed to resolve slot-consistent companion-account fetches by entrypoint, fetch reason, companion kind, and outcome", - ) - .buckets(vec![1.0, 2.0, 3.0, 4.0, 5.0, 8.0, 10.0, 15.0, 20.0]), - &["entrypoint", "fetch_reason", "companion_kind", "outcome"], - ) - .unwrap(); - - static ref CHAINLINK_COMPANION_FETCH_DURATION_SECONDS: HistogramVec = - HistogramVec::new( - HistogramOpts::new( - "chainlink_companion_fetch_duration_seconds", - "Time spent resolving slot-consistent companion-account fetches by entrypoint, fetch reason, companion kind, and outcome", - ) - .buckets( - MILLIS_10_90.iter().chain( - MILLIS_100_900.iter()).chain( - SECONDS_1_9.iter()).cloned().collect() - ), - &["entrypoint", "fetch_reason", "companion_kind", "outcome"], - ) - .unwrap(); - - // ----------------- - // RPC/Aperture - // ----------------- - pub static ref ENSURE_ACCOUNTS_TIME: HistogramVec = HistogramVec::new( - HistogramOpts::new("ensure_accounts_time", "Time spent in ensuring account presence") - .buckets( - MICROS_10_90.iter().chain( - MICROS_100_900.iter()).chain( - MILLIS_1_9.iter()).chain( - MILLIS_10_90.iter()).chain( - MILLIS_100_900.iter()).chain( - SECONDS_1_9.iter()).cloned().collect() - ), - &["kind"] - ).unwrap(); - - pub static ref TRANSACTION_PROCESSING_TIME: Histogram = Histogram::with_opts( - HistogramOpts::new("transaction_processing_time", "Total time spent in transaction processing") - .buckets( - MICROS_10_90.iter().chain( - MICROS_100_900.iter()).chain( - MILLIS_1_9.iter()).chain( - MILLIS_10_90.iter()).chain( - MILLIS_100_900.iter()).chain( - SECONDS_1_9.iter()).cloned().collect() - ), - ).unwrap(); - - pub static ref RPC_REQUEST_HANDLING_TIME: HistogramVec = HistogramVec::new( - HistogramOpts::new("rpc_request_handling_time", "Time spent in rpc request handling") - .buckets( - MICROS_10_90.iter().chain( - MICROS_100_900.iter()).chain( - MILLIS_1_9.iter()).chain( - MILLIS_10_90.iter()).chain( - MILLIS_100_900.iter()).chain( - SECONDS_1_9.iter()).cloned().collect() - ), &["name"] - ).unwrap(); - - pub static ref TRANSACTION_SKIP_PREFLIGHT: IntCounter = IntCounter::new( - "transaction_skip_preflight_count", "Count of transactions that skipped the preflight check", - ).unwrap(); - - pub static ref RPC_REQUESTS_COUNT: IntCounterVec = IntCounterVec::new( - Opts::new("rpc_requests_count", "Count of different rpc requests"), - &["name"], - ).unwrap(); - - // Account fetch results from network (RPC) - static ref ACCOUNT_FETCHES_SUCCESS_COUNT: IntCounter = - IntCounter::new( - "account_fetches_success_count", - "Total number of successful network \ - account fetches", - ) - .unwrap(); - - static ref ACCOUNT_FETCHES_FAILED_COUNT: IntCounter = - IntCounter::new( - "account_fetches_failed_count", - "Total number of failed network account fetches \ - (RPC errors)", - ) - .unwrap(); - - static ref ACCOUNT_FETCHES_FOUND_COUNT: IntCounterVec = IntCounterVec::new( - Opts::new( - "account_fetches_found_count", - "Total number of network account fetches that found an account", - ), - &["entrypoint", "fetch_reason"], - ) - .unwrap(); - - static ref ACCOUNT_FETCHES_NOT_FOUND_COUNT: IntCounterVec = IntCounterVec::new( - Opts::new( - "account_fetches_not_found_count", - "Total number of network account fetches where account was not found", - ), - &["entrypoint", "fetch_reason"], - ) - .unwrap(); - - static ref CHAINLINK_CLONE_ACCOUNTS_TOTAL: IntCounterVec = - IntCounterVec::new( - Opts::new( - "chainlink_clone_accounts_total", - "Total number of Chainlink clone attempts and outcomes", - ), - &["entrypoint", "fetch_reason", "remote_result", "clone_intent", "outcome"], - ) - .unwrap(); - - static ref CHAINLINK_EMPTY_PLACEHOLDER_ACCOUNTS_TOTAL: IntCounterVec = - IntCounterVec::new( - Opts::new( - "chainlink_empty_placeholder_accounts_total", - "Total number of Chainlink empty-placeholder lifecycle events", - ), - &[ - "entrypoint", - "fetch_reason", - "stage", - "outcome", - ], - ) - .unwrap(); - - static ref PER_PROGRAM_ACCOUNT_UPDATES_COUNT: IntCounterVec = - IntCounterVec::new( - Opts::new( - "per_program_account_updates_count", - "Number of account updates received grouped by program id, no matter if they match an account subscription", - ), - &["client_id", "program"], - ) - .unwrap(); - - static ref UNDELEGATION_REQUESTED_COUNT: IntCounter = - IntCounter::new( - "undelegation_requested_count", - "Total number of undelegation requests received", - ) - .unwrap(); - - static ref UNDELEGATION_COMPLETED_COUNT: IntCounter = - IntCounter::new( - "undelegation_completed_count", - "Total number of completed undelegations detected", - ) - .unwrap(); - - static ref UNSTUCK_UNDELEGATION_COUNT: IntCounter = - IntCounter::new( - "unstuck_undelegation_count", - "Total number of undelegating accounts found to be already undelegated on chain", - ) - .unwrap(); - - // ----------------- - // CommittorService - // ----------------- - static ref COMMITTOR_INTENTS_COUNT: IntCounter = IntCounter::new( - "committor_intents_count", "Total number of scheduled committor intents" - ).unwrap(); - - static ref COMMITTOR_INTENTS_BACKLOG_COUNT: IntGauge = IntGauge::new( - "committor_intent_backlog_count", "Number of intents in backlog", - ).unwrap(); - - static ref COMMITTOR_FAILED_INTENTS_COUNT: IntCounterVec = IntCounterVec::new( - Opts::new("committor_failed_intents_count", "Number of failed to be executed intents"), - &["intent_kind", "error_kind"] - ).unwrap(); - - static ref COMMITTOR_EXECUTORS_BUSY_COUNT: IntGauge = IntGauge::new( - "committor_executors_busy_count", "Number of busy intent executors" - ).unwrap(); - - static ref COMMITTOR_POISONED_KEYS_COUNT: IntGauge = IntGauge::new( - "committor_poisoned_keys_count", - "Number of pubkeys currently poisoned in the intent scheduler (only cleared by a process restart)" - ).unwrap(); - - static ref COMMITTOR_CASCADE_VOIDED_INTENTS_COUNT: IntCounter = IntCounter::new( - "committor_cascade_voided_intents_count", - "Total number of intents voided by a poisoning cascade after a dependency failed" - ).unwrap(); - - static ref COMMITTOR_INTENT_EXECUTION_TIME_HISTOGRAM: HistogramVec = HistogramVec::new( - HistogramOpts::new( - "committor_intent_execution_time_histogram_v2", - "Time in seconds spent on intent execution" - ) - .buckets( - vec![0.01, 0.1, 1.0, 3.0, 5.0, 10.0, 15.0, 20.0, 25.0] - ), - &["intent_kind", "outcome_kind"], - ).unwrap(); - - static ref COMMITTOR_INTENT_CU_USAGE: IntGauge = IntGauge::new( - "committor_intent_cu_usage_gauge", "Compute units used for Intent" - ).unwrap(); - - static ref COMMITTOR_INTENT_TASK_PREPARATION_TIME: HistogramVec = HistogramVec::new( - HistogramOpts::new( - "committor_intent_task_preparation_time", - "Time in seconds spent on task preparation" - ) - .buckets( - vec![0.1, 1.0, 2.0, 3.0, 5.0] - ), - &["task_type"], - ).unwrap(); - - static ref COMMITTOR_INTENT_ALT_PREPARATION_TIME: Histogram = Histogram::with_opts( - HistogramOpts::new( - "committor_intent_alt_preparation_time", - "Time in seconds spent on ALTs preparation" - ) - .buckets( - vec![1.0, 3.0, 5.0, 10.0, 15.0, 17.0, 20.0] - ), - ).unwrap(); - - static ref COMMITTOR_INTENT_ALT_COUNT: Histogram = Histogram::with_opts( - HistogramOpts::new( - "committor_intent_alt_count", - "Number of address lookup tables used per intent transaction (only recorded when ALTs are present)" - ) - .buckets(vec![0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 10.0]) - ).unwrap(); - - static ref COMMITTOR_FETCH_COMMIT_NONCES_WAIT_TIME: Histogram = Histogram::with_opts( - HistogramOpts::new( - "committor_fetch_commit_nonces_wait_time_second", - "Time in seconds spent waiting for fetch_current_commit_nonces response" - ) - .buckets(vec![0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 20.0, 30.0]), - ).unwrap(); - - // GetMultiplAccount investigation - static ref REMOTE_ACCOUNT_PROVIDER_A_COUNT: IntCounter = IntCounter::new( - "remote_account_provider_a_count", "Get mupltiple account count" - ).unwrap(); - - static ref TASK_INFO_FETCHER_A_COUNT: IntCounter = IntCounter::new( - "task_info_fetcher_a_count", "Get mupltiple account count" - ).unwrap(); - - static ref TASK_INFO_FETCHER_RETIRING_GAUGE: IntGauge = IntGauge::new( - "task_info_fetcher_retiring_gauge", - "Number of pubkeys currently in the retiring map of CacheTaskInfoFetcher" - ).unwrap(); - - static ref TABLE_MANIA_A_COUNT: IntCounter = IntCounter::new( - "table_mania_a_count", "Get mupltiple account count" - ).unwrap(); - - static ref TABLE_MANIA_CLOSED_A_COUNT: IntCounter = IntCounter::new( - "table_mania_closed_a_count", "Get account counter" - ).unwrap(); - - static ref RPC_CLIENT_SIGNATURE_WS_SUBSCRIBE_COUNT: IntCounter = IntCounter::new( - "rpc_client_signature_ws_subscribe_count", - "Number of signatureSubscribe registrations created by MagicblockRpcClient" - ).unwrap(); - - static ref RPC_CLIENT_SIGNATURE_WS_NOTIFICATION_COUNT: IntCounter = IntCounter::new( - "rpc_client_signature_ws_notification_count", - "Number of signatureSubscribe notifications received by MagicblockRpcClient" - ).unwrap(); - - static ref RPC_CLIENT_SIGNATURE_WS_FALLBACK_COUNT: IntCounter = IntCounter::new( - "rpc_client_signature_ws_fallback_count", - "Number of signature confirmations that fell back to batched polling" - ).unwrap(); - - static ref RPC_CLIENT_SIGNATURE_STATUS_BATCH_COUNT: IntCounter = IntCounter::new( - "rpc_client_signature_status_batch_count", - "Number of batched getSignatureStatuses requests sent by MagicblockRpcClient" - ).unwrap(); - - static ref RPC_CLIENT_SIGNATURE_STATUS_BATCH_SIGNATURES_COUNT: IntCounter = IntCounter::new( - "rpc_client_signature_status_batch_signatures_count", - "Number of signatures included in batched getSignatureStatuses requests" - ).unwrap(); - - static ref RPC_CLIENT_SIGNATURE_HISTORY_DURATION_SECONDS: Histogram = Histogram::with_opts( - HistogramOpts::new( - "rpc_client_signature_history_duration_seconds", - "Time in seconds spent on getSignatureStatuses-with-history (restart recovery)" - ) - .buckets(vec![0.05, 0.1, 0.25, 0.5, 1.0, 2.0, 5.0, 10.0]) - ).unwrap(); - - // ----------------- - // Pubsub Clients - // ----------------- - static ref CONNECTED_PUBSUB_CLIENTS_GAUGE: IntGauge = IntGauge::new( - "connected_pubsub_clients_gauge", - "Total number of connected pubsub clients" - ).unwrap(); - - /// Gauge for pubsub clients that subscribe immediately when requested. - /// If this value goes to 0 then we should raise an alert as it means we are missing - /// account updates. - static ref CONNECTED_DIRECT_PUBSUB_CLIENTS_GAUGE: IntGauge = IntGauge::new( - "connected_direct_pubsub_clients_gauge", - "Total number of connected pubsub clients that subscribe immediately when requested." - ).unwrap(); - - static ref PUBSUB_CLIENT_UPTIME_GAUGE: IntGaugeVec = IntGaugeVec::new( - Opts::new( - "pubsub_client_uptime_gauge", - "Uptime of each pubsub client (1=connected, 0=disconnected)" - ), - &["client_id"], - ).unwrap(); - - static ref PUBSUB_CLIENT_RECONNECT_BACKOFF_DURATION_SECONDS_GAUGE: IntGaugeVec = IntGaugeVec::new( - Opts::new( - "pubsub_client_reconnect_backoff_duration_seconds_gauge", - "Current backoff wait duration in seconds during reconnect attempts (0 when connected)" - ), - &["client_id"], - ).unwrap(); - - static ref PUBSUB_CLIENT_FAILED_RECONNECT_ATTEMPTS_GAUGE: IntGaugeVec = IntGaugeVec::new( - Opts::new( - "pubsub_client_failed_reconnect_attempts_gauge", - "Current number of failed reconnect attempts for each pubsub client (0 when connected)" - ), - &["client_id"], - ).unwrap(); - - static ref PUBSUB_CLIENT_RESUBSCRIBE_DELAY_MILLISECONDS_GAUGE: IntGaugeVec = IntGaugeVec::new( - Opts::new( - "pubsub_client_resubscribe_delay_milliseconds_gauge", - "Current resubscription delay in milliseconds for each pubsub client" - ), - &["client_id"], - ).unwrap(); - - static ref PUBSUB_CLIENT_RESUBSCRIBED_GAUGE: IntGaugeVec = IntGaugeVec::new( - Opts::new( - "pubsub_client_resubscribed_gauge", - "Number of subscriptions successfully resubscribed for each pubsub client before complete or failed" - ), - &["client_id"], - ).unwrap(); - - static ref PUBSUB_CLIENT_CONNECTIONS_GAUGE: IntGaugeVec = IntGaugeVec::new( - Opts::new( - "pubsub_client_connections_gauge", - "Number of pooled websocket connections per pubsub client" - ), - &["client_id"], - ).unwrap(); - - static ref PUBSUB_UNSUBSCRIBE_TIMEOUT_COUNT: IntCounterVec = - IntCounterVec::new( - Opts::new( - "pubsub_unsubscribe_timeout_count", - "Number of unsubscribe calls that timed out", - ), - &["client_id", "scope"], - ) - .unwrap(); - - static ref PUBSUB_IDLE_CONNECTIONS_PRUNED_COUNT: IntCounterVec = - IntCounterVec::new( - Opts::new( - "pubsub_idle_connections_pruned_count", - "Number of idle pooled connections pruned", - ), - &["client_id"], - ) - .unwrap(); - - // ----------------- - // GRPC Streams - // ----------------- - static ref GRPC_OPTIMIZED_STREAMS_GAUGE: IntGaugeVec = - IntGaugeVec::new( - Opts::new( - "grpc_optimized_streams_gauge", - "Number of optimized GRPC streams", - ), - &["client_id"], - ) - .unwrap(); - - static ref GRPC_UNOPTIMIZED_STREAMS_GAUGE: IntGaugeVec = - IntGaugeVec::new( - Opts::new( - "grpc_unoptimized_streams_gauge", - "Number of unoptimized GRPC streams", - ), - &["client_id"], - ) - .unwrap(); - - static ref GRPC_TOTAL_STREAMS_GAUGE: IntGaugeVec = - IntGaugeVec::new( - Opts::new( - "grpc_total_streams_gauge", - "Total number of GRPC streams including current stream and program stream", - ), - &["client_id"], - ) - .unwrap(); -} - -pub(crate) fn register() { - static REGISTER: Once = Once::new(); - REGISTER.call_once(|| { - macro_rules! register { - ($collector:ident) => { - REGISTRY - .register(Box::new($collector.clone())) - .expect("collector can't be registered"); - }; - } - register!(CHAIN_SLOT_GAUGE); - register!(MONITORED_ACCOUNTS_GAUGE); - register!(INFLIGHT_SUBSCRIPTION_UPDATES_GAUGE); - register!(CHAINLINK_SUBSCRIPTION_REGISTRATION_ACCOUNTS_TOTAL); - register!(CHAINLINK_SUBSCRIPTION_RELEASE_ACCOUNTS_TOTAL); - register!(CHAINLINK_SUBSCRIPTION_CLEANUP_ACCOUNTS_TOTAL); - register!(PROGRAM_SUBSCRIPTION_DISCOVERED_DLP_UPDATE_DELEGATED_ELSEWHERE_COUNT); - register!(PROGRAM_SUBSCRIPTION_ACCOUNT_UPDATES_COUNT); - register!(ACCOUNT_SUBSCRIPTION_ACCOUNT_UPDATES_COUNT); - register!(CHAINLINK_PENDING_FETCH_ACCOUNTS_TOTAL); - register!(CHAINLINK_PENDING_FETCH_WAITERS_TOTAL); - register!(CHAINLINK_PENDING_FETCH_WAITERS_GAUGE); - register!(CHAINLINK_PENDING_FETCH_OWNER_DURATION_SECONDS); - register!(CHAINLINK_COMPANION_FETCH_ATTEMPTS); - register!(CHAINLINK_COMPANION_FETCH_DURATION_SECONDS); - register!(COMMITTOR_INTENTS_COUNT); - register!(COMMITTOR_INTENTS_BACKLOG_COUNT); - register!(COMMITTOR_FAILED_INTENTS_COUNT); - register!(COMMITTOR_EXECUTORS_BUSY_COUNT); - register!(COMMITTOR_POISONED_KEYS_COUNT); - register!(COMMITTOR_CASCADE_VOIDED_INTENTS_COUNT); - register!(COMMITTOR_INTENT_EXECUTION_TIME_HISTOGRAM); - register!(COMMITTOR_INTENT_CU_USAGE); - register!(COMMITTOR_INTENT_TASK_PREPARATION_TIME); - register!(COMMITTOR_INTENT_ALT_PREPARATION_TIME); - register!(COMMITTOR_INTENT_ALT_COUNT); - register!(COMMITTOR_FETCH_COMMIT_NONCES_WAIT_TIME); - register!(ENSURE_ACCOUNTS_TIME); - register!(RPC_REQUEST_HANDLING_TIME); - register!(TRANSACTION_PROCESSING_TIME); - register!(TRANSACTION_SKIP_PREFLIGHT); - register!(RPC_REQUESTS_COUNT); - register!(ACCOUNT_FETCHES_SUCCESS_COUNT); - register!(ACCOUNT_FETCHES_FAILED_COUNT); - register!(ACCOUNT_FETCHES_FOUND_COUNT); - register!(ACCOUNT_FETCHES_NOT_FOUND_COUNT); - register!(CHAINLINK_CLONE_ACCOUNTS_TOTAL); - register!(CHAINLINK_EMPTY_PLACEHOLDER_ACCOUNTS_TOTAL); - register!(PER_PROGRAM_ACCOUNT_UPDATES_COUNT); - register!(UNDELEGATION_REQUESTED_COUNT); - register!(UNDELEGATION_COMPLETED_COUNT); - register!(UNSTUCK_UNDELEGATION_COUNT); - register!(REMOTE_ACCOUNT_PROVIDER_A_COUNT); - register!(TASK_INFO_FETCHER_A_COUNT); - register!(TASK_INFO_FETCHER_RETIRING_GAUGE); - register!(TABLE_MANIA_A_COUNT); - register!(TABLE_MANIA_CLOSED_A_COUNT); - register!(RPC_CLIENT_SIGNATURE_WS_SUBSCRIBE_COUNT); - register!(RPC_CLIENT_SIGNATURE_WS_NOTIFICATION_COUNT); - register!(RPC_CLIENT_SIGNATURE_WS_FALLBACK_COUNT); - register!(RPC_CLIENT_SIGNATURE_STATUS_BATCH_COUNT); - register!(RPC_CLIENT_SIGNATURE_STATUS_BATCH_SIGNATURES_COUNT); - register!(RPC_CLIENT_SIGNATURE_HISTORY_DURATION_SECONDS); - register!(CONNECTED_PUBSUB_CLIENTS_GAUGE); - register!(CONNECTED_DIRECT_PUBSUB_CLIENTS_GAUGE); - register!(PUBSUB_CLIENT_UPTIME_GAUGE); - register!(PUBSUB_CLIENT_RECONNECT_BACKOFF_DURATION_SECONDS_GAUGE); - register!(PUBSUB_CLIENT_FAILED_RECONNECT_ATTEMPTS_GAUGE); - register!(PUBSUB_CLIENT_RESUBSCRIBE_DELAY_MILLISECONDS_GAUGE); - register!(PUBSUB_CLIENT_RESUBSCRIBED_GAUGE); - register!(PUBSUB_CLIENT_CONNECTIONS_GAUGE); - register!(PUBSUB_UNSUBSCRIBE_TIMEOUT_COUNT); - register!(PUBSUB_IDLE_CONNECTIONS_PRUNED_COUNT); - register!(GRPC_OPTIMIZED_STREAMS_GAUGE); - register!(GRPC_UNOPTIMIZED_STREAMS_GAUGE); - register!(GRPC_TOTAL_STREAMS_GAUGE); - }); -} - -pub fn set_chain_slot(value: u64) { - CHAIN_SLOT_GAUGE.set(value as i64); -} - -pub fn inc_inflight_subscription_updates() { - INFLIGHT_SUBSCRIPTION_UPDATES_GAUGE.inc() -} - -pub fn dec_inflight_subscription_updates() { - INFLIGHT_SUBSCRIPTION_UPDATES_GAUGE.dec() -} - -/// Sets the absolute number of monitored accounts. -/// -/// This metric reflects the current total count of accounts being monitored. -/// Callers must pass the total number of monitored accounts, not a delta. -pub fn set_monitored_accounts_count(count: usize) { - MONITORED_ACCOUNTS_GAUGE.set(count as i64); -} -pub fn inc_chainlink_subscription_registration_accounts( - origin: SubscriptionRegistrationOrigin, - subscription_reason: SubscriptionReasonLabel, - outcome: SubscriptionRegistrationOutcome, -) { - CHAINLINK_SUBSCRIPTION_REGISTRATION_ACCOUNTS_TOTAL - .with_label_values(&[ - origin.entrypoint_str(), - origin.fetch_reason_str(), - subscription_reason.value(), - outcome.value(), - ]) - .inc(); -} - -pub fn inc_chainlink_subscription_release_accounts( - reason: SubscriptionReasonLabel, - outcome: SubscriptionReleaseOutcome, -) { - CHAINLINK_SUBSCRIPTION_RELEASE_ACCOUNTS_TOTAL - .with_label_values(&[reason.value(), outcome.value()]) - .inc(); -} - -pub fn inc_chainlink_subscription_cleanup_accounts( - cleanup_source: SubscriptionCleanupSource, - outcome: SubscriptionCleanupOutcome, -) { - CHAINLINK_SUBSCRIPTION_CLEANUP_ACCOUNTS_TOTAL - .with_label_values(&[cleanup_source.value(), outcome.value()]) - .inc(); -} - -#[cfg(any(test, feature = "dev-context"))] -pub fn chainlink_subscription_registration_accounts_value( - origin: SubscriptionRegistrationOrigin, - subscription_reason: SubscriptionReasonLabel, - outcome: SubscriptionRegistrationOutcome, -) -> u64 { - CHAINLINK_SUBSCRIPTION_REGISTRATION_ACCOUNTS_TOTAL - .get_metric_with_label_values(&[ - origin.entrypoint_str(), - origin.fetch_reason_str(), - subscription_reason.value(), - outcome.value(), - ]) - .map(|m| m.get()) - .unwrap_or(0) -} - -#[cfg(any(test, feature = "dev-context"))] -pub fn chainlink_subscription_release_accounts_value( - reason: SubscriptionReasonLabel, - outcome: SubscriptionReleaseOutcome, -) -> u64 { - CHAINLINK_SUBSCRIPTION_RELEASE_ACCOUNTS_TOTAL - .get_metric_with_label_values(&[reason.value(), outcome.value()]) - .map(|m| m.get()) - .unwrap_or(0) -} - -#[cfg(any(test, feature = "dev-context"))] -pub fn chainlink_subscription_cleanup_accounts_value( - cleanup_source: SubscriptionCleanupSource, - outcome: SubscriptionCleanupOutcome, -) -> u64 { - CHAINLINK_SUBSCRIPTION_CLEANUP_ACCOUNTS_TOTAL - .get_metric_with_label_values(&[cleanup_source.value(), outcome.value()]) - .map(|m| m.get()) - .unwrap_or(0) -} - -pub fn inc_discovered_dlp_update_delegated_elsewhere() { - PROGRAM_SUBSCRIPTION_DISCOVERED_DLP_UPDATE_DELEGATED_ELSEWHERE_COUNT.inc(); -} - -pub fn inc_committor_intents_count_by(by: u64) { - COMMITTOR_INTENTS_COUNT.inc_by(by) -} - -pub fn set_committor_intents_backlog_count(value: i64) { - COMMITTOR_INTENTS_BACKLOG_COUNT.set(value) -} - -pub fn inc_committor_failed_intents_count( - intent_kind: &impl LabelValue, - error_kind: &impl LabelValue, -) { - COMMITTOR_FAILED_INTENTS_COUNT - .with_label_values(&[intent_kind.value(), error_kind.value()]) - .inc() -} - -pub fn set_committor_executors_busy_count(value: i64) { - COMMITTOR_EXECUTORS_BUSY_COUNT.set(value) -} - -pub fn set_committor_poisoned_keys_count(value: i64) { - COMMITTOR_POISONED_KEYS_COUNT.set(value) -} - -pub fn inc_committor_cascade_voided_intents_count_by(by: u64) { - COMMITTOR_CASCADE_VOIDED_INTENTS_COUNT.inc_by(by) -} - -pub fn observe_committor_intent_execution_time_histogram( - seconds: f64, - kind: &impl LabelValue, - outcome: &impl LabelValue, -) { - COMMITTOR_INTENT_EXECUTION_TIME_HISTOGRAM - .with_label_values(&[kind.value(), outcome.value()]) - .observe(seconds); -} - -pub fn set_commmittor_intent_cu_usage(value: i64) { - COMMITTOR_INTENT_CU_USAGE.set(value) -} - -pub fn observe_committor_intent_task_preparation_time( - task_type: &L, -) -> HistogramTimer { - COMMITTOR_INTENT_TASK_PREPARATION_TIME - .with_label_values(&[task_type.value()]) - .start_timer() -} - -pub fn observe_committor_intent_alt_preparation_time() -> HistogramTimer { - COMMITTOR_INTENT_ALT_PREPARATION_TIME.start_timer() -} - -pub fn observe_committor_intent_alt_count(count: usize) { - COMMITTOR_INTENT_ALT_COUNT.observe(count as f64); -} - -pub fn start_fetch_commit_nonces_wait_timer() -> HistogramTimer { - COMMITTOR_FETCH_COMMIT_NONCES_WAIT_TIME.start_timer() -} - -pub fn inc_account_fetches_success(count: u64) { - ACCOUNT_FETCHES_SUCCESS_COUNT.inc_by(count); -} - -pub fn inc_account_fetches_failed(count: u64) { - ACCOUNT_FETCHES_FAILED_COUNT.inc_by(count); -} - -pub fn inc_account_fetches_found_with_context(context: AccountFetchContext, count: u64) { - ACCOUNT_FETCHES_FOUND_COUNT - .with_label_values(&[context.entrypoint().value(), context.reason().value()]) - .inc_by(count); -} - -pub fn inc_account_fetches_not_found_with_context(context: AccountFetchContext, count: u64) { - ACCOUNT_FETCHES_NOT_FOUND_COUNT - .with_label_values(&[context.entrypoint().value(), context.reason().value()]) - .inc_by(count); -} - -pub fn inc_chainlink_clone_accounts_total_with_context( - context: AccountFetchContext, - remote_result: ChainlinkCloneRemoteResult, - clone_intent: ChainlinkCloneIntent, - outcome: ChainlinkCloneOutcome, -) { - CHAINLINK_CLONE_ACCOUNTS_TOTAL - .with_label_values(&[ - context.entrypoint().value(), - context.reason().value(), - remote_result.value(), - clone_intent.value(), - outcome.value(), - ]) - .inc(); -} - -pub fn inc_chainlink_empty_placeholder_accounts_total_with_context( - context: AccountFetchContext, - stage: ChainlinkEmptyPlaceholderStage, - outcome: Outcome, -) { - CHAINLINK_EMPTY_PLACEHOLDER_ACCOUNTS_TOTAL - .with_label_values(&[ - context.entrypoint().value(), - context.reason().value(), - stage.value(), - outcome.value(), - ]) - .inc(); -} - -pub fn inc_program_subscription_account_updates_count(client_id: &impl LabelValue) { - PROGRAM_SUBSCRIPTION_ACCOUNT_UPDATES_COUNT - .with_label_values(&[client_id.value()]) - .inc(); -} - -pub fn inc_account_subscription_account_updates_count(client_id: &impl LabelValue) { - ACCOUNT_SUBSCRIPTION_ACCOUNT_UPDATES_COUNT - .with_label_values(&[client_id.value()]) - .inc(); -} - -pub fn inc_chainlink_pending_fetch_accounts_with_context( - context: AccountFetchContext, - layer: ChainlinkPendingFetchLayer, - outcome: ChainlinkPendingFetchOutcome, - count: u64, -) { - CHAINLINK_PENDING_FETCH_ACCOUNTS_TOTAL - .with_label_values(&[ - context.entrypoint().value(), - context.reason().value(), - layer.value(), - outcome.value(), - ]) - .inc_by(count); -} - -pub fn inc_chainlink_pending_fetch_waiters_with_context( - context: AccountFetchContext, - layer: ChainlinkPendingFetchLayer, - count: u64, -) { - CHAINLINK_PENDING_FETCH_WAITERS_TOTAL - .with_label_values(&[context.entrypoint().value(), context.reason().value(), layer.value()]) - .inc_by(count); -} - -pub fn inc_chainlink_pending_fetch_waiters_gauge(layer: ChainlinkPendingFetchLayer) { - CHAINLINK_PENDING_FETCH_WAITERS_GAUGE.with_label_values(&[layer.value()]).inc(); -} - -pub fn dec_chainlink_pending_fetch_waiters_gauge(layer: ChainlinkPendingFetchLayer) { - CHAINLINK_PENDING_FETCH_WAITERS_GAUGE.with_label_values(&[layer.value()]).dec(); -} - -pub fn observe_chainlink_pending_fetch_owner_duration_seconds_with_context( - context: AccountFetchContext, - layer: ChainlinkPendingFetchLayer, - outcome: ChainlinkPendingFetchOutcome, - seconds: f64, -) { - CHAINLINK_PENDING_FETCH_OWNER_DURATION_SECONDS - .with_label_values(&[ - context.entrypoint().value(), - context.reason().value(), - layer.value(), - outcome.value(), - ]) - .observe(seconds); -} - -pub fn observe_chainlink_companion_fetch_attempts( - context: AccountFetchContext, - kind: ChainlinkCompanionFetchKind, - outcome: ChainlinkCompanionFetchOutcome, - attempts: f64, -) { - CHAINLINK_COMPANION_FETCH_ATTEMPTS - .with_label_values(&[ - context.entrypoint().value(), - context.reason().value(), - kind.value(), - outcome.value(), - ]) - .observe(attempts); -} - -pub fn observe_chainlink_companion_fetch_duration_seconds( - context: AccountFetchContext, - kind: ChainlinkCompanionFetchKind, - outcome: ChainlinkCompanionFetchOutcome, - seconds: f64, -) { - CHAINLINK_COMPANION_FETCH_DURATION_SECONDS - .with_label_values(&[ - context.entrypoint().value(), - context.reason().value(), - kind.value(), - outcome.value(), - ]) - .observe(seconds); -} - -#[cfg(any(test, feature = "dev-context"))] -pub fn chainlink_pending_fetch_accounts_value( - context: impl Into, - layer: ChainlinkPendingFetchLayer, - outcome: ChainlinkPendingFetchOutcome, -) -> u64 { - let context = context.into(); - CHAINLINK_PENDING_FETCH_ACCOUNTS_TOTAL - .get_metric_with_label_values(&[ - context.entrypoint().value(), - context.reason().value(), - layer.value(), - outcome.value(), - ]) - .map(|m| m.get()) - .unwrap_or(0) -} - -#[cfg(any(test, feature = "dev-context"))] -pub fn chainlink_pending_fetch_waiters_value( - context: impl Into, - layer: ChainlinkPendingFetchLayer, -) -> u64 { - let context = context.into(); - CHAINLINK_PENDING_FETCH_WAITERS_TOTAL - .get_metric_with_label_values(&[ - context.entrypoint().value(), - context.reason().value(), - layer.value(), - ]) - .map(|m| m.get()) - .unwrap_or(0) -} - -#[cfg(any(test, feature = "dev-context"))] -pub fn chainlink_pending_fetch_waiters_gauge_value(layer: ChainlinkPendingFetchLayer) -> i64 { - CHAINLINK_PENDING_FETCH_WAITERS_GAUGE - .get_metric_with_label_values(&[layer.value()]) - .map(|m| m.get()) - .unwrap_or(0) -} - -pub fn inc_per_program_account_updates_count(client_id: &str, program_id: &str) { - PER_PROGRAM_ACCOUNT_UPDATES_COUNT - .with_label_values(&[client_id, program_id]) - .inc() -} - -pub fn inc_undelegation_requested() { - UNDELEGATION_REQUESTED_COUNT.inc(); -} - -pub fn inc_undelegation_completed() { - UNDELEGATION_COMPLETED_COUNT.inc(); -} - -pub fn inc_unstuck_undelegation_count() { - UNSTUCK_UNDELEGATION_COUNT.inc(); -} - -pub fn inc_remote_account_provider_a_count() { - REMOTE_ACCOUNT_PROVIDER_A_COUNT.inc() -} - -pub fn inc_task_info_fetcher_a_count() { - TASK_INFO_FETCHER_A_COUNT.inc() -} - -pub fn set_task_info_fetcher_retiring_count(count: i64) { - TASK_INFO_FETCHER_RETIRING_GAUGE.set(count); -} - -pub fn inc_table_mania_a_count() { - TABLE_MANIA_A_COUNT.inc() -} - -pub fn inc_table_mania_close_a_count() { - TABLE_MANIA_CLOSED_A_COUNT.inc() -} - -pub fn inc_rpc_client_signature_ws_subscribe_count() { - RPC_CLIENT_SIGNATURE_WS_SUBSCRIBE_COUNT.inc() -} - -pub fn inc_rpc_client_signature_ws_notification_count() { - RPC_CLIENT_SIGNATURE_WS_NOTIFICATION_COUNT.inc() -} - -pub fn inc_rpc_client_signature_ws_fallback_count() { - RPC_CLIENT_SIGNATURE_WS_FALLBACK_COUNT.inc() -} - -pub fn inc_rpc_client_signature_status_batch_count() { - RPC_CLIENT_SIGNATURE_STATUS_BATCH_COUNT.inc() -} - -pub fn inc_rpc_client_signature_status_batch_signatures_count(count: u64) { - RPC_CLIENT_SIGNATURE_STATUS_BATCH_SIGNATURES_COUNT.inc_by(count) -} - -pub fn start_rpc_client_signature_history_timer() -> HistogramTimer { - RPC_CLIENT_SIGNATURE_HISTORY_DURATION_SECONDS.start_timer() -} - -pub fn set_connected_pubsub_clients_count(count: usize) { - CONNECTED_PUBSUB_CLIENTS_GAUGE.set(count as i64); -} - -pub fn set_connected_direct_pubsub_clients_count(count: usize) { - CONNECTED_DIRECT_PUBSUB_CLIENTS_GAUGE.set(count as i64); -} - -pub fn set_pubsub_client_uptime(client_id: &str, connected: bool) { - PUBSUB_CLIENT_UPTIME_GAUGE - .with_label_values(&[client_id]) - .set(if connected { 1 } else { 0 }); -} - -pub fn set_pubsub_client_reconnect_backoff_duration_seconds(client_id: &str, duration_secs: u64) { - PUBSUB_CLIENT_RECONNECT_BACKOFF_DURATION_SECONDS_GAUGE - .with_label_values(&[client_id]) - .set(duration_secs as i64); -} - -pub fn set_pubsub_client_failed_reconnect_attempts(client_id: &str, attempts: u64) { - PUBSUB_CLIENT_FAILED_RECONNECT_ATTEMPTS_GAUGE - .with_label_values(&[client_id]) - .set(attempts as i64); -} - -pub fn set_pubsub_client_resubscribe_delay(client_id: &str, delay_ms: u64) { - PUBSUB_CLIENT_RESUBSCRIBE_DELAY_MILLISECONDS_GAUGE - .with_label_values(&[client_id]) - .set(delay_ms as i64); -} - -pub fn set_pubsub_client_resubscribed_count(client_id: &str, count: usize) { - PUBSUB_CLIENT_RESUBSCRIBED_GAUGE - .with_label_values(&[client_id]) - .set(count as i64); -} - -pub fn set_pubsub_client_connections_count(client_id: &str, count: usize) { - PUBSUB_CLIENT_CONNECTIONS_GAUGE - .with_label_values(&[client_id]) - .set(count as i64); -} - -pub fn inc_pubsub_unsubscribe_timeout_count(client_id: &str, scope: &str) { - PUBSUB_UNSUBSCRIBE_TIMEOUT_COUNT.with_label_values(&[client_id, scope]).inc(); -} - -pub fn inc_pubsub_idle_connections_pruned_count(client_id: &str, count: u64) { - PUBSUB_IDLE_CONNECTIONS_PRUNED_COUNT - .with_label_values(&[client_id]) - .inc_by(count); -} - -pub fn set_grpc_optimized_streams_gauge(client_id: &str, count: usize) { - GRPC_OPTIMIZED_STREAMS_GAUGE.with_label_values(&[client_id]).set(count as i64); -} - -pub fn set_grpc_unoptimized_streams_gauge(client_id: &str, count: usize) { - GRPC_UNOPTIMIZED_STREAMS_GAUGE.with_label_values(&[client_id]).set(count as i64); -} - -pub fn set_grpc_total_streams_gauge(client_id: &str, count: usize) { - GRPC_TOTAL_STREAMS_GAUGE.with_label_values(&[client_id]).set(count as i64); -} - -#[cfg(test)] -mod fetch_context_metric_tests { - use super::*; - - fn counter_value(counter: &IntCounterVec, labels: &[&str]) -> u64 { - counter.get_metric_with_label_values(labels).map(|m| m.get()).unwrap_or(0) - } - - #[test] - fn fetch_context_metrics_keep_entrypoint_and_reason_separate() { - let action_context = AccountFetchContext::rpc_get_account() - .with_reason(AccountFetchReason::ActionDependencyMissing); - let sub_context = - AccountFetchContext::subscription_update(AccountFetchReason::SubscriptionUpdateClone); - let record_context = - AccountFetchContext::subscription_update(AccountFetchReason::DelegationRecord); - let program_context = - AccountFetchContext::subscription_update(AccountFetchReason::ProgramData); - let undelegating_context = AccountFetchContext::rpc_get_account() - .with_reason(AccountFetchReason::UndelegatingRefresh); - let simulation_context = - AccountFetchContext::simulate_transaction(solana_signature::Signature::from([1u8; 64])); - - let action_labels = &[ - "rpc_get_account", - "action_dependency_missing", - "remote_account_provider", - "owned", - ]; - let before_action = counter_value(&CHAINLINK_PENDING_FETCH_ACCOUNTS_TOTAL, action_labels); - inc_chainlink_pending_fetch_accounts_with_context( - action_context, - ChainlinkPendingFetchLayer::RemoteAccountProvider, - ChainlinkPendingFetchOutcome::Owned, - 1, - ); - assert_eq!( - counter_value(&CHAINLINK_PENDING_FETCH_ACCOUNTS_TOTAL, action_labels), - before_action + 1 - ); - - let sub_labels = &[ - "subscription_update", - "subscription_update_clone", - "found", - "normal_account", - "clone_succeeded", - ]; - let before_sub = counter_value(&CHAINLINK_CLONE_ACCOUNTS_TOTAL, sub_labels); - inc_chainlink_clone_accounts_total_with_context( - sub_context, - ChainlinkCloneRemoteResult::Found, - ChainlinkCloneIntent::NormalAccount, - ChainlinkCloneOutcome::CloneSucceeded, - ); - assert_eq!( - counter_value(&CHAINLINK_CLONE_ACCOUNTS_TOTAL, sub_labels), - before_sub + 1 - ); - - let record_labels = - &["subscription_update", "delegation_record", "direct_account", "added"]; - let before_record = counter_value( - &CHAINLINK_SUBSCRIPTION_REGISTRATION_ACCOUNTS_TOTAL, - record_labels, - ); - inc_chainlink_subscription_registration_accounts( - SubscriptionRegistrationOrigin::Fetch(record_context), - SubscriptionReasonLabel::DirectAccount, - SubscriptionRegistrationOutcome::Added, - ); - assert_eq!( - counter_value( - &CHAINLINK_SUBSCRIPTION_REGISTRATION_ACCOUNTS_TOTAL, - record_labels, - ), - before_record + 1 - ); - - let program_labels = &["subscription_update", "program_data"]; - let before_program = counter_value(&ACCOUNT_FETCHES_FOUND_COUNT, program_labels); - inc_account_fetches_found_with_context(program_context, 1); - assert_eq!( - counter_value(&ACCOUNT_FETCHES_FOUND_COUNT, program_labels), - before_program + 1 - ); - - let undelegating_labels = &["rpc_get_account", "undelegating_refresh"]; - let before_undelegating = - counter_value(&ACCOUNT_FETCHES_NOT_FOUND_COUNT, undelegating_labels); - inc_account_fetches_not_found_with_context(undelegating_context, 1); - assert_eq!( - counter_value(&ACCOUNT_FETCHES_NOT_FOUND_COUNT, undelegating_labels), - before_undelegating + 1 - ); - - let simulation_labels = - &["simulate_transaction", "requested_account", "remote_account_provider", "owned"]; - let before_simulation = - counter_value(&CHAINLINK_PENDING_FETCH_ACCOUNTS_TOTAL, simulation_labels); - inc_chainlink_pending_fetch_accounts_with_context( - simulation_context, - ChainlinkPendingFetchLayer::RemoteAccountProvider, - ChainlinkPendingFetchOutcome::Owned, - 1, - ); - assert_eq!( - counter_value(&CHAINLINK_PENDING_FETCH_ACCOUNTS_TOTAL, simulation_labels,), - before_simulation + 1 - ); - } -} diff --git a/metrics/src/metrics/types.rs b/metrics/src/metrics/types.rs deleted file mode 100644 index 6b437f74d..000000000 --- a/metrics/src/metrics/types.rs +++ /dev/null @@ -1,796 +0,0 @@ -use std::{ - fmt, - sync::{ - Arc, - atomic::{AtomicU64, Ordering}, - }, -}; - -use solana_signature::Signature; - -// ----------------- -// Outcome -// ----------------- -const OUTCOME_SUCCESS: &str = "success"; -const OUTCOME_ERROR: &str = "error"; - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum Outcome { - Success, - Error, -} - -impl fmt::Display for Outcome { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - use Outcome::*; - match self { - Success => write!(f, "{OUTCOME_SUCCESS}"), - Error => write!(f, "{OUTCOME_ERROR}"), - } - } -} - -impl Outcome { - fn as_str(&self) -> &str { - use Outcome::*; - match self { - Success => OUTCOME_SUCCESS, - Error => OUTCOME_ERROR, - } - } -} - -impl LabelValue for Outcome { - fn value(&self) -> &str { - self.as_str() - } -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum AccountFetchEntrypoint { - RpcGetAccount, - RpcGetMultipleAccounts, - SendTransaction(Signature), - SimulateTransaction(Signature), - SubscriptionUpdate, - ProjectAta, - Internal, -} - -impl AccountFetchEntrypoint { - fn as_str(self) -> &'static str { - match self { - Self::RpcGetAccount => "rpc_get_account", - Self::RpcGetMultipleAccounts => "rpc_get_multiple_accounts", - Self::SendTransaction(_) => "send_transaction", - Self::SimulateTransaction(_) => "simulate_transaction", - Self::SubscriptionUpdate => "subscription_update", - Self::ProjectAta => "project_ata", - Self::Internal => "internal", - } - } - - fn signature(&self) -> Option<&Signature> { - match self { - Self::SendTransaction(sig) | Self::SimulateTransaction(sig) => Some(sig), - _ => None, - } - } -} - -impl fmt::Display for AccountFetchEntrypoint { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - write!(f, "{}", self.as_str()) - } -} - -impl LabelValue for AccountFetchEntrypoint { - fn value(&self) -> &str { - self.as_str() - } -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum AccountFetchReason { - RequestedAccount, - DelegationRecord, - ProgramData, - ActionDependencyMissing, - ActionDependencyForcedRefresh, - UndelegatingRefresh, - SubscriptionUpdateClone, - SubscriptionUpdateGreedyDiscovery, - AtaProjection, - ProgramLoad, - Clock, -} - -impl AccountFetchReason { - fn as_str(self) -> &'static str { - match self { - Self::RequestedAccount => "requested_account", - Self::DelegationRecord => "delegation_record", - Self::ProgramData => "program_data", - Self::ActionDependencyMissing => "action_dependency_missing", - Self::ActionDependencyForcedRefresh => "action_dependency_forced_refresh", - Self::UndelegatingRefresh => "undelegating_refresh", - Self::SubscriptionUpdateClone => "subscription_update_clone", - Self::SubscriptionUpdateGreedyDiscovery => "subscription_update_greedy_discovery", - Self::AtaProjection => "ata_projection", - Self::ProgramLoad => "program_load", - Self::Clock => "clock", - } - } -} - -impl fmt::Display for AccountFetchReason { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - write!(f, "{}", self.as_str()) - } -} - -impl LabelValue for AccountFetchReason { - fn value(&self) -> &str { - self.as_str() - } -} - -#[derive(Debug, Clone)] -pub struct AccountFetchContext { - entrypoint: AccountFetchEntrypoint, - reason: AccountFetchReason, - remote_account_claims: Arc, -} - -impl AccountFetchContext { - fn new(entrypoint: AccountFetchEntrypoint, reason: AccountFetchReason) -> Self { - Self { - entrypoint, - reason, - remote_account_claims: Arc::new(AtomicU64::new(0)), - } - } - - pub fn rpc_get_account() -> Self { - AccountFetchEntrypoint::RpcGetAccount.into() - } - - pub fn rpc_get_multiple_accounts() -> Self { - AccountFetchEntrypoint::RpcGetMultipleAccounts.into() - } - - pub fn send_transaction(signature: Signature) -> Self { - AccountFetchEntrypoint::SendTransaction(signature).into() - } - - pub fn simulate_transaction(signature: Signature) -> Self { - AccountFetchEntrypoint::SimulateTransaction(signature).into() - } - - pub fn subscription_update(reason: AccountFetchReason) -> Self { - Self::new(AccountFetchEntrypoint::SubscriptionUpdate, reason) - } - - pub fn project_ata() -> Self { - Self::new( - AccountFetchEntrypoint::ProjectAta, - AccountFetchReason::AtaProjection, - ) - } - - pub fn internal(reason: AccountFetchReason) -> Self { - Self::new(AccountFetchEntrypoint::Internal, reason) - } - - pub fn entrypoint(&self) -> AccountFetchEntrypoint { - self.entrypoint - } - - pub fn reason(&self) -> AccountFetchReason { - self.reason - } - - pub fn with_reason(self, reason: AccountFetchReason) -> Self { - Self { reason, ..self } - } - - pub fn should_count_remote_account_claims(&self) -> bool { - // We only count fetches due to direct user requests, not due to internal - // fetches, i.e. to get a companion account - if self.reason != AccountFetchReason::RequestedAccount { - return false; - } - matches!( - self.entrypoint, - AccountFetchEntrypoint::RpcGetAccount - | AccountFetchEntrypoint::RpcGetMultipleAccounts - | AccountFetchEntrypoint::SendTransaction(_) - | AccountFetchEntrypoint::SimulateTransaction(_) - ) - } - - pub fn add_remote_account_claims(&self, count: usize) { - self.remote_account_claims.fetch_add(count as u64, Ordering::Relaxed); - } - - pub fn remote_account_claims_value(&self) -> u64 { - self.remote_account_claims.load(Ordering::Relaxed) - } - - pub fn signature(&self) -> Option<&Signature> { - self.entrypoint.signature() - } -} - -/// Starts shared requested-account fetch tracking from a copyable origin. -impl From for AccountFetchContext { - fn from(entrypoint: AccountFetchEntrypoint) -> Self { - Self::new(entrypoint, AccountFetchReason::RequestedAccount) - } -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum ChainlinkPendingFetchLayer { - RemoteAccountProvider, -} - -impl ChainlinkPendingFetchLayer { - fn as_str(&self) -> &str { - match self { - Self::RemoteAccountProvider => "remote_account_provider", - } - } -} - -impl fmt::Display for ChainlinkPendingFetchLayer { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - write!(f, "{}", self.as_str()) - } -} - -impl LabelValue for ChainlinkPendingFetchLayer { - fn value(&self) -> &str { - self.as_str() - } -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum ChainlinkPendingFetchOutcome { - Owned, - JoinedExisting, - OwnerSucceeded, - OwnerFailed, - ResolvedBySubscriptionUpdate, - RpcFetchCompletedAfterUpdate, -} - -impl ChainlinkPendingFetchOutcome { - fn as_str(&self) -> &str { - match self { - Self::Owned => "owned", - Self::JoinedExisting => "joined_existing", - Self::OwnerSucceeded => "owner_succeeded", - Self::OwnerFailed => "owner_failed", - Self::ResolvedBySubscriptionUpdate => "resolved_by_subscription_update", - Self::RpcFetchCompletedAfterUpdate => "rpc_fetch_completed_after_update", - } - } -} - -impl fmt::Display for ChainlinkPendingFetchOutcome { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - write!(f, "{}", self.as_str()) - } -} - -impl LabelValue for ChainlinkPendingFetchOutcome { - fn value(&self) -> &str { - self.as_str() - } -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum ChainlinkCompanionFetchKind { - ProgramData, - DelegationRecord, - AtaProjection, -} - -impl ChainlinkCompanionFetchKind { - fn as_str(&self) -> &str { - match self { - Self::ProgramData => "program_data", - Self::DelegationRecord => "delegation_record", - Self::AtaProjection => "ata_projection", - } - } -} - -impl fmt::Display for ChainlinkCompanionFetchKind { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - write!(f, "{}", self.as_str()) - } -} - -impl LabelValue for ChainlinkCompanionFetchKind { - fn value(&self) -> &str { - self.as_str() - } -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum ChainlinkCompanionFetchOutcome { - Succeeded, - FailedRpc, - FailedSlotMismatch, - FailedMinContextSlot, -} - -impl ChainlinkCompanionFetchOutcome { - fn as_str(&self) -> &str { - match self { - Self::Succeeded => "succeeded", - Self::FailedRpc => "failed_rpc", - Self::FailedSlotMismatch => "failed_slot_mismatch", - Self::FailedMinContextSlot => "failed_min_context_slot", - } - } -} - -impl fmt::Display for ChainlinkCompanionFetchOutcome { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - write!(f, "{}", self.as_str()) - } -} - -impl LabelValue for ChainlinkCompanionFetchOutcome { - fn value(&self) -> &str { - self.as_str() - } -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum ChainlinkCloneRemoteResult { - Found, - NotFound, - Failed, -} - -impl ChainlinkCloneRemoteResult { - fn as_str(&self) -> &str { - match self { - Self::Found => "found", - Self::NotFound => "not_found", - Self::Failed => "failed", - } - } -} - -impl fmt::Display for ChainlinkCloneRemoteResult { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - write!(f, "{}", self.as_str()) - } -} - -impl LabelValue for ChainlinkCloneRemoteResult { - fn value(&self) -> &str { - self.as_str() - } -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum ChainlinkCloneIntent { - NormalAccount, - EmptyPlaceholder, - ProgramData, - DelegationRecord, - ActionDependency, - Unknown, -} - -impl ChainlinkCloneIntent { - fn as_str(&self) -> &str { - match self { - Self::NormalAccount => "normal_account", - Self::EmptyPlaceholder => "empty_placeholder", - Self::ProgramData => "program_data", - Self::DelegationRecord => "delegation_record", - Self::ActionDependency => "action_dependency", - Self::Unknown => "unknown", - } - } -} - -impl fmt::Display for ChainlinkCloneIntent { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - write!(f, "{}", self.as_str()) - } -} - -impl LabelValue for ChainlinkCloneIntent { - fn value(&self) -> &str { - self.as_str() - } -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum ChainlinkCloneOutcome { - Submitted, - SubmitFailed, - CloneSucceeded, - CloneFailed, - Skipped, -} - -impl ChainlinkCloneOutcome { - fn as_str(&self) -> &str { - match self { - Self::Submitted => "submitted", - Self::SubmitFailed => "submit_failed", - Self::CloneSucceeded => "clone_succeeded", - Self::CloneFailed => "clone_failed", - Self::Skipped => "skipped", - } - } -} - -impl fmt::Display for ChainlinkCloneOutcome { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - write!(f, "{}", self.as_str()) - } -} - -impl LabelValue for ChainlinkCloneOutcome { - fn value(&self) -> &str { - self.as_str() - } -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum ChainlinkEmptyPlaceholderStage { - ConvertedToEmpty, - CloneSubmitted, - CloneSubmitFailed, -} - -impl ChainlinkEmptyPlaceholderStage { - fn as_str(&self) -> &str { - match self { - Self::ConvertedToEmpty => "converted_to_empty", - Self::CloneSubmitted => "clone_submitted", - Self::CloneSubmitFailed => "clone_submit_failed", - } - } -} - -impl fmt::Display for ChainlinkEmptyPlaceholderStage { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - write!(f, "{}", self.as_str()) - } -} - -impl LabelValue for ChainlinkEmptyPlaceholderStage { - fn value(&self) -> &str { - self.as_str() - } -} - -#[derive(Debug, Clone)] -pub enum SubscriptionRegistrationOrigin { - Fetch(AccountFetchContext), - Internal, -} - -impl SubscriptionRegistrationOrigin { - pub(super) fn entrypoint_str(&self) -> &str { - match self { - Self::Fetch(context) => context.entrypoint().as_str(), - Self::Internal => AccountFetchEntrypoint::Internal.as_str(), - } - } - - pub(super) fn fetch_reason_str(&self) -> &str { - match self { - Self::Fetch(context) => context.reason().as_str(), - Self::Internal => AccountFetchReason::RequestedAccount.as_str(), - } - } -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum SubscriptionReasonLabel { - DirectAccount, - DelegationRecord, - ProgramData, - UndelegationTracking, - AtaProjection, -} - -impl SubscriptionReasonLabel { - fn as_str(&self) -> &str { - match self { - Self::DirectAccount => "direct_account", - Self::DelegationRecord => "delegation_record", - Self::ProgramData => "program_data", - Self::UndelegationTracking => "undelegation_tracking", - Self::AtaProjection => "ata_projection", - } - } -} - -impl fmt::Display for SubscriptionReasonLabel { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - write!(f, "{}", self.as_str()) - } -} - -impl LabelValue for SubscriptionReasonLabel { - fn value(&self) -> &str { - self.as_str() - } -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum SubscriptionRegistrationOutcome { - AlreadyPresent, - Added, - SubscribeError, -} - -impl SubscriptionRegistrationOutcome { - fn as_str(&self) -> &str { - match self { - Self::AlreadyPresent => "already_present", - Self::Added => "added", - Self::SubscribeError => "subscribe_error", - } - } -} - -impl fmt::Display for SubscriptionRegistrationOutcome { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - write!(f, "{}", self.as_str()) - } -} - -impl LabelValue for SubscriptionRegistrationOutcome { - fn value(&self) -> &str { - self.as_str() - } -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum SubscriptionReleaseOutcome { - Unsubscribed, - AlreadyAbsent, - UnsubscribeFailed, - RetainedIntentionally, - RetainedOtherReasons, -} - -impl SubscriptionReleaseOutcome { - fn as_str(&self) -> &str { - match self { - Self::Unsubscribed => "unsubscribed", - Self::AlreadyAbsent => "already_absent", - Self::UnsubscribeFailed => "unsubscribe_failed", - Self::RetainedIntentionally => "retained_intentionally", - Self::RetainedOtherReasons => "retained_other_reasons", - } - } -} - -impl fmt::Display for SubscriptionReleaseOutcome { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - write!(f, "{}", self.as_str()) - } -} - -impl LabelValue for SubscriptionReleaseOutcome { - fn value(&self) -> &str { - self.as_str() - } -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum SubscriptionCleanupSource { - NormalRelease, - ManualUnsubscribe, - DelegatedAccountSilent, - Reconciler, -} - -impl SubscriptionCleanupSource { - fn as_str(&self) -> &str { - match self { - Self::NormalRelease => "normal_release", - Self::ManualUnsubscribe => "manual_unsubscribe", - Self::DelegatedAccountSilent => "delegated_account_silent", - Self::Reconciler => "reconciler", - } - } -} - -impl fmt::Display for SubscriptionCleanupSource { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - write!(f, "{}", self.as_str()) - } -} - -impl LabelValue for SubscriptionCleanupSource { - fn value(&self) -> &str { - self.as_str() - } -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum SubscriptionCleanupOutcome { - Unsubscribed, - AlreadyAbsent, - UnsubscribeFailed, - RemovalUpdateFailed, - RetainedIntentionally, -} - -impl SubscriptionCleanupOutcome { - fn as_str(&self) -> &str { - match self { - Self::Unsubscribed => "unsubscribed", - Self::AlreadyAbsent => "already_absent", - Self::UnsubscribeFailed => "unsubscribe_failed", - Self::RemovalUpdateFailed => "removal_update_failed", - Self::RetainedIntentionally => "retained_intentionally", - } - } -} - -impl fmt::Display for SubscriptionCleanupOutcome { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - write!(f, "{}", self.as_str()) - } -} - -impl LabelValue for SubscriptionCleanupOutcome { - fn value(&self) -> &str { - self.as_str() - } -} - -pub trait LabelValue { - fn value(&self) -> &str; -} - -impl LabelValue for &str { - fn value(&self) -> &str { - self - } -} - -impl LabelValue for String { - fn value(&self) -> &str { - self - } -} - -impl LabelValue for Result -where - T: LabelValue, - E: LabelValue, -{ - fn value(&self) -> &str { - match self { - Ok(ok) => ok.value(), - Err(err) => err.value(), - } - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn account_fetch_entrypoint_labels_are_static() { - let signature = Signature::from([1u8; 64]); - let cases = [ - (AccountFetchEntrypoint::RpcGetAccount, "rpc_get_account"), - ( - AccountFetchEntrypoint::RpcGetMultipleAccounts, - "rpc_get_multiple_accounts", - ), - ( - AccountFetchEntrypoint::SendTransaction(signature), - "send_transaction", - ), - ( - AccountFetchEntrypoint::SimulateTransaction(signature), - "simulate_transaction", - ), - ( - AccountFetchEntrypoint::SubscriptionUpdate, - "subscription_update", - ), - (AccountFetchEntrypoint::ProjectAta, "project_ata"), - (AccountFetchEntrypoint::Internal, "internal"), - ]; - - for (entrypoint, expected) in cases { - assert_eq!(entrypoint.as_str(), expected); - assert_eq!(entrypoint.to_string(), expected); - assert_eq!(entrypoint.value(), expected); - } - } - - #[test] - fn account_fetch_reason_labels_are_static() { - let cases = [ - (AccountFetchReason::RequestedAccount, "requested_account"), - (AccountFetchReason::DelegationRecord, "delegation_record"), - (AccountFetchReason::ProgramData, "program_data"), - ( - AccountFetchReason::ActionDependencyMissing, - "action_dependency_missing", - ), - ( - AccountFetchReason::ActionDependencyForcedRefresh, - "action_dependency_forced_refresh", - ), - ( - AccountFetchReason::UndelegatingRefresh, - "undelegating_refresh", - ), - ( - AccountFetchReason::SubscriptionUpdateClone, - "subscription_update_clone", - ), - ( - AccountFetchReason::SubscriptionUpdateGreedyDiscovery, - "subscription_update_greedy_discovery", - ), - (AccountFetchReason::AtaProjection, "ata_projection"), - (AccountFetchReason::ProgramLoad, "program_load"), - (AccountFetchReason::Clock, "clock"), - ]; - - for (reason, expected) in cases { - assert_eq!(reason.as_str(), expected); - assert_eq!(reason.to_string(), expected); - assert_eq!(reason.value(), expected); - } - } - - #[test] - fn account_fetch_context_signature_is_for_transaction_entrypoints() { - let signature = Signature::from([1u8; 64]); - let send_context = AccountFetchContext::send_transaction(signature); - let simulate_context = AccountFetchContext::simulate_transaction(signature); - for context in [&send_context, &simulate_context] { - assert_eq!(context.signature(), Some(&signature)); - assert_eq!(context.entrypoint().signature(), Some(&signature)); - } - - let contexts = [ - AccountFetchContext::rpc_get_account(), - AccountFetchContext::rpc_get_multiple_accounts(), - AccountFetchContext::subscription_update(AccountFetchReason::SubscriptionUpdateClone), - AccountFetchContext::project_ata(), - AccountFetchContext::internal(AccountFetchReason::Clock), - ]; - - for context in contexts { - assert_eq!(context.signature(), None); - assert_eq!(context.entrypoint().signature(), None); - } - } - - #[test] - fn simulation_requested_accounts_count_remote_account_claims() { - let context = AccountFetchContext::simulate_transaction(Signature::from([1u8; 64])); - assert!(context.should_count_remote_account_claims()); - assert!( - !context - .with_reason(AccountFetchReason::ProgramData) - .should_count_remote_account_claims() - ); - } -} diff --git a/rpc-client/Cargo.toml b/rpc-client/Cargo.toml index f470525d4..9c0656e90 100644 --- a/rpc-client/Cargo.toml +++ b/rpc-client/Cargo.toml @@ -12,7 +12,7 @@ doctest = false [dependencies] futures-util = { workspace = true } -metrics = { workspace = true } +nucleus = { workspace = true, features = ["metrics"] } serde_json = { workspace = true } solana-account = { workspace = true } solana-account-decoder-client-types = { workspace = true } diff --git a/rpc-client/src/lib.rs b/rpc-client/src/lib.rs index b03e09a32..5386e64f9 100644 --- a/rpc-client/src/lib.rs +++ b/rpc-client/src/lib.rs @@ -830,3 +830,5 @@ impl MagicblockRpcClient { &self.client } } + +mod metrics; diff --git a/rpc-client/src/metrics.rs b/rpc-client/src/metrics.rs new file mode 100644 index 000000000..98a59903e --- /dev/null +++ b/rpc-client/src/metrics.rs @@ -0,0 +1,69 @@ +//! Event-driven rpc_client metrics registered in the default Prometheus registry. + +use nucleus::metrics::{IntCounter, Opts, counter}; +use std::sync::OnceLock; + +static METRICS: OnceLock = OnceLock::new(); + +struct Metrics { + ws_subscribe: IntCounter, + ws_notification: IntCounter, + ws_fallback: IntCounter, + status_batch: IntCounter, + status_signatures: IntCounter, +} + +fn metrics() -> &'static Metrics { + METRICS.get_or_init(|| Metrics { + ws_subscribe: counter( + Opts::new( + "rpc_client_signature_ws_subscription_attempts_total", + "Signature WebSocket subscription attempts", + ), + 0, + ), + ws_notification: counter( + Opts::new( + "rpc_client_signature_ws_notifications_total", + "Signature WebSocket notifications", + ), + 0, + ), + ws_fallback: counter( + Opts::new( + "rpc_client_signature_http_fallbacks_total", + "Signature confirmation fallbacks to HTTP", + ), + 0, + ), + status_batch: counter( + Opts::new( + "rpc_client_signature_status_batches_total", + "Signature status HTTP batches", + ), + 0, + ), + status_signatures: counter( + Opts::new( + "rpc_client_signature_status_entries_total", + "Signatures queried in HTTP status batches", + ), + 0, + ), + }) +} + +pub(crate) fn subscribe() { + metrics().ws_subscribe.inc(); +} +pub(crate) fn notification() { + metrics().ws_notification.inc(); +} +pub(crate) fn fallback() { + metrics().ws_fallback.inc(); +} +pub(crate) fn status_batch(count: usize) { + let m = metrics(); + m.status_batch.inc(); + m.status_signatures.inc_by(count as u64); +} diff --git a/rpc-client/src/signature_confirmer.rs b/rpc-client/src/signature_confirmer.rs index ff5f8e809..d6a0fe53d 100644 --- a/rpc-client/src/signature_confirmer.rs +++ b/rpc-client/src/signature_confirmer.rs @@ -7,8 +7,8 @@ use std::{ time::{Duration, Instant}, }; +use crate::metrics; use futures_util::StreamExt; -use metrics::metrics; use solana_commitment_config::CommitmentConfig; use solana_pubsub_client::nonblocking::pubsub_client::PubsubClient; use solana_rpc_client::nonblocking::rpc_client::RpcClient; @@ -115,7 +115,7 @@ impl SignatureConfirmer { let pubsub_client = match timeout(fallback_delay, self.pubsub_client()).await { Ok(Some(pubsub_client)) => pubsub_client, Ok(None) => { - metrics::inc_rpc_client_signature_ws_fallback_count(); + metrics::fallback(); return self .wait_with_batched_polling( signature, @@ -130,7 +130,7 @@ impl SignatureConfirmer { signature = %signature, "Signature confirmation websocket connection timed out; falling back to batched polling" ); - metrics::inc_rpc_client_signature_ws_fallback_count(); + metrics::fallback(); return self .wait_with_batched_polling( signature, @@ -160,7 +160,7 @@ impl SignatureConfirmer { error = ?err, "Signature confirmation websocket subscription failed; falling back to batched polling" ); - metrics::inc_rpc_client_signature_ws_fallback_count(); + metrics::fallback(); self.clear_pubsub_client().await; return self .wait_with_batched_polling( @@ -176,7 +176,7 @@ impl SignatureConfirmer { signature = %signature, "Signature confirmation websocket subscription timed out; falling back to batched polling" ); - metrics::inc_rpc_client_signature_ws_fallback_count(); + metrics::fallback(); return self .wait_with_batched_polling( signature, @@ -188,12 +188,12 @@ impl SignatureConfirmer { } }; let mut unsubscribe = Some(unsubscribe); - metrics::inc_rpc_client_signature_ws_subscribe_count(); + metrics::subscribe(); let fallback_delay = fallback_delay.saturating_sub(start.elapsed()); let fallback = async { sleep(fallback_delay).await; - metrics::inc_rpc_client_signature_ws_fallback_count(); + metrics::fallback(); let remaining = timeout_duration.saturating_sub(start.elapsed()); self.wait_with_batched_polling(signature, commitment, remaining, poll_interval) .await @@ -209,7 +209,7 @@ impl SignatureConfirmer { if let Some(unsubscribe) = unsubscribe.take() { unsubscribe().await; } - metrics::inc_rpc_client_signature_ws_notification_count(); + metrics::notification(); return Some(status); } } @@ -376,8 +376,7 @@ impl SignatureConfirmer { ) -> Vec<(Signature, TransactionStatus)> { let mut fetched = Vec::new(); for chunk in signatures.chunks(self.config.batch_size) { - metrics::inc_rpc_client_signature_status_batch_count(); - metrics::inc_rpc_client_signature_status_batch_signatures_count(chunk.len() as u64); + metrics::status_batch(chunk.len()); match self.rpc_client.get_signature_statuses(chunk).await { Ok(response) => { fetched.extend(chunk.iter().copied().zip(response.value).filter_map( diff --git a/table-mania/Cargo.toml b/table-mania/Cargo.toml index 5bed43633..fe5d438b3 100644 --- a/table-mania/Cargo.toml +++ b/table-mania/Cargo.toml @@ -11,7 +11,7 @@ version.workspace = true doctest = false [dependencies] -metrics = { workspace = true } +nucleus = { workspace = true, features = ["metrics"] } rpc-client = { workspace = true } parking_lot = { workspace = true } rand = { workspace = true } diff --git a/table-mania/src/lib.rs b/table-mania/src/lib.rs index 54db169ce..dc9a14886 100644 --- a/table-mania/src/lib.rs +++ b/table-mania/src/lib.rs @@ -11,3 +11,5 @@ pub use compute_budget::*; pub use find_tables::find_open_tables; pub use lookup_table_rc::{LookupTableRc, MAX_ENTRIES_AS_PART_OF_EXTEND}; pub use manager::*; + +mod metrics; diff --git a/table-mania/src/lookup_table_rc.rs b/table-mania/src/lookup_table_rc.rs index 2661ad63c..23cf51e45 100644 --- a/table-mania/src/lookup_table_rc.rs +++ b/table-mania/src/lookup_table_rc.rs @@ -6,7 +6,7 @@ use std::{ time::Instant, }; -use metrics::metrics; +use crate::metrics; use parking_lot::{Mutex, RwLock, RwLockReadGuard, RwLockWriteGuard}; use rpc_client::{MagicBlockRpcClientError, MagicBlockSendTransactionConfig, MagicblockRpcClient}; use solana_address_lookup_table_interface::{ @@ -739,7 +739,7 @@ impl LookupTableRc { } pub async fn is_closed(&self, rpc_client: &MagicblockRpcClient) -> TableManiaResult { - metrics::inc_table_mania_close_a_count(); + metrics::closure_check(); let acc = rpc_client.get_account(self.table_address()).await?; Ok(acc.is_none()) } diff --git a/table-mania/src/manager.rs b/table-mania/src/manager.rs index 8fb30b137..5c906d693 100644 --- a/table-mania/src/manager.rs +++ b/table-mania/src/manager.rs @@ -7,7 +7,7 @@ use std::{ time::{Duration, Instant}, }; -use metrics::metrics; +use crate::metrics; use rpc_client::MagicblockRpcClient; use solana_address_lookup_table_interface::state::AddressLookupTable; use solana_commitment_config::CommitmentConfig; @@ -678,7 +678,7 @@ impl TableMania { .await; loop { - metrics::inc_table_mania_a_count(); + metrics::readiness_fetch(); // Fetch the tables from chain let remote_table_accs = self .rpc_client diff --git a/table-mania/src/metrics.rs b/table-mania/src/metrics.rs new file mode 100644 index 000000000..3bbbc2629 --- /dev/null +++ b/table-mania/src/metrics.rs @@ -0,0 +1,37 @@ +//! Event-driven table_mania metrics registered in the default Prometheus registry. + +use nucleus::metrics::{IntCounter, Opts, counter}; +use std::sync::OnceLock; + +static METRICS: OnceLock = OnceLock::new(); + +struct Metrics { + fetches: IntCounter, + closure_checks: IntCounter, +} + +fn metrics() -> &'static Metrics { + METRICS.get_or_init(|| Metrics { + fetches: counter( + Opts::new( + "lookup_table_readiness_fetch_attempts_total", + "Remote lookup-table readiness fetch attempts", + ), + 0, + ), + closure_checks: counter( + Opts::new( + "lookup_table_closure_checks_total", + "Remote lookup-table closure checks", + ), + 0, + ), + }) +} + +pub(crate) fn readiness_fetch() { + metrics().fetches.inc(); +} +pub(crate) fn closure_check() { + metrics().closure_checks.inc(); +} diff --git a/test-integration/Cargo.lock b/test-integration/Cargo.lock index 753aa57c2..1347828f9 100644 --- a/test-integration/Cargo.lock +++ b/test-integration/Cargo.lock @@ -3788,7 +3788,6 @@ dependencies = [ "magicblock-delegation-program-api 3.1.0 (git+https://github.com/magicblock-labs/delegation-program.git?rev=fb6668c7b5ffd6a89bedee1d13b16402fd0e65de)", "magicblock-engine", "magicblock-engine-nucleus", - "magicblock-metrics", "magicblock-program", "magicblock-rpc-client", "magicblock-table-mania", @@ -4088,21 +4087,6 @@ dependencies = [ "wincode", ] -[[package]] -name = "magicblock-metrics" -version = "1.0.1" -dependencies = [ - "http-body-util", - "hyper", - "hyper-util", - "lazy_static", - "magicblock-engine-nucleus", - "prometheus", - "solana-signature", - "tokio", - "tracing", -] - [[package]] name = "magicblock-processor" version = "0.15.0" @@ -4181,7 +4165,7 @@ name = "magicblock-rpc-client" version = "1.0.1" dependencies = [ "futures-util", - "magicblock-metrics", + "magicblock-engine-nucleus", "serde_json", "solana-account 4.3.1", "solana-account-decoder-client-types", @@ -4206,7 +4190,7 @@ dependencies = [ name = "magicblock-table-mania" version = "1.0.1" dependencies = [ - "magicblock-metrics", + "magicblock-engine-nucleus", "magicblock-rpc-client", "parking_lot", "rand 0.9.5", @@ -5009,7 +4993,6 @@ dependencies = [ "lazy_static", "memchr", "parking_lot", - "protobuf", "thiserror 2.0.20", ] @@ -5066,26 +5049,6 @@ dependencies = [ "prost", ] -[[package]] -name = "protobuf" -version = "3.7.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d65a1d4ddae7d8b5de68153b48f6aa3bba8cb002b243dbdbc55a5afbc98f99f4" -dependencies = [ - "once_cell", - "protobuf-support", - "thiserror 1.0.69", -] - -[[package]] -name = "protobuf-support" -version = "3.7.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3e36c2f31e0a47f9280fb347ef5e461ffcd2c52dd520d8e216b52f93b0b0d7d6" -dependencies = [ - "thiserror 1.0.69", -] - [[package]] name = "protoc-bin-vendored" version = "3.2.0"