reth_db/
metrics.rs

1use crate::Tables;
2use metrics::Histogram;
3use reth_metrics::{metrics::Counter, Metrics};
4use reth_primitives_traits::FastInstant as Instant;
5use rustc_hash::FxHashMap;
6use std::{array, sync::Arc, time::Duration};
7use strum::{EnumCount, EnumIter, IntoEnumIterator};
8
9const LARGE_VALUE_THRESHOLD_BYTES: usize = 4096;
10
11/// Caches metric handles for database environment to make sure handles are not re-created
12/// on every operation.
13///
14/// Requires a metric recorder to be registered before creating an instance of this struct.
15/// Otherwise, metric recording will no-op.
16#[derive(Debug)]
17pub(crate) struct DatabaseEnvMetrics {
18    /// Caches per-table operation metric handles for all database operation metrics, indexed by
19    /// the table's position in [`Tables::ALL`].
20    operations: [TableOperationMetrics; Tables::COUNT],
21    /// Caches `TransactionMetrics` handles for counters grouped by only transaction mode.
22    /// Updated both at tx open and close.
23    transactions: FxHashMap<TransactionMode, TransactionMetrics>,
24    /// Caches `TransactionOutcomeMetrics` handles for counters grouped by transaction mode and
25    /// outcome. Can only be updated at tx close, as outcome is only known at that point.
26    transaction_outcomes:
27        FxHashMap<(TransactionMode, TransactionOutcome), TransactionOutcomeMetrics>,
28}
29
30/// Per-table operation metric handles cached for hot cursor paths.
31pub(crate) type TableOperationMetrics = Arc<[OperationMetrics; Operation::COUNT]>;
32
33impl DatabaseEnvMetrics {
34    pub(crate) fn new() -> Self {
35        // Pre-populate metric handle maps with all possible combinations of labels
36        // to avoid runtime locks on the map when recording metrics.
37        Self {
38            operations: Self::generate_operation_handles(),
39            transactions: Self::generate_transaction_handles(),
40            transaction_outcomes: Self::generate_transaction_outcome_handles(),
41        }
42    }
43
44    /// Generate pre-bound operation handles for each table, in [`Tables::ALL`] order.
45    fn generate_operation_handles() -> [TableOperationMetrics; Tables::COUNT] {
46        array::from_fn(|table_index| {
47            let table_name = Tables::ALL[table_index].name();
48            let metrics = array::from_fn(|index| {
49                let operation = Operation::from_index(index);
50                OperationMetrics::new_with_labels(&[
51                    (Labels::Table.as_str(), table_name),
52                    (Labels::Operation.as_str(), operation.as_str()),
53                ])
54            });
55
56            Arc::new(metrics)
57        })
58    }
59
60    /// Generate a map of all possible transaction modes to metric handles.
61    /// Used for tracking a counter of open transactions.
62    fn generate_transaction_handles() -> FxHashMap<TransactionMode, TransactionMetrics> {
63        TransactionMode::iter()
64            .map(|mode| {
65                (
66                    mode,
67                    TransactionMetrics::new_with_labels(&[(
68                        Labels::TransactionMode.as_str(),
69                        mode.as_str(),
70                    )]),
71                )
72            })
73            .collect()
74    }
75
76    /// Generate a map of all possible transaction mode and outcome handles.
77    /// Used for tracking various stats for finished transactions (e.g. commit duration).
78    fn generate_transaction_outcome_handles(
79    ) -> FxHashMap<(TransactionMode, TransactionOutcome), TransactionOutcomeMetrics> {
80        let mut transaction_outcomes = FxHashMap::with_capacity_and_hasher(
81            TransactionMode::COUNT * TransactionOutcome::COUNT,
82            Default::default(),
83        );
84        for mode in TransactionMode::iter() {
85            for outcome in TransactionOutcome::iter() {
86                transaction_outcomes.insert(
87                    (mode, outcome),
88                    TransactionOutcomeMetrics::new_with_labels(&[
89                        (Labels::TransactionMode.as_str(), mode.as_str()),
90                        (Labels::TransactionOutcome.as_str(), outcome.as_str()),
91                    ]),
92                );
93            }
94        }
95        transaction_outcomes
96    }
97
98    /// Record a metric for database operation executed in `f`.
99    ///
100    /// Operations on tables defined outside of [`Tables`] (`table` is `None`) are not recorded.
101    pub(crate) fn record_operation<R>(
102        &self,
103        table: Option<Tables>,
104        operation: Operation,
105        value_size: Option<usize>,
106        f: impl FnOnce() -> R,
107    ) -> R {
108        if let Some(table) = table {
109            self.operations[table as usize][operation.index()].record(value_size, f)
110        } else {
111            f()
112        }
113    }
114
115    /// Returns pre-bound operation metric handles for a single table.
116    pub(crate) fn table_operation_metrics(&self, table: Tables) -> TableOperationMetrics {
117        self.operations[table as usize].clone()
118    }
119
120    /// Record metrics for opening a database transaction.
121    pub(crate) fn record_opened_transaction(&self, mode: TransactionMode) {
122        self.transactions
123            .get(&mode)
124            .expect("transaction mode metric handle not found")
125            .record_open();
126    }
127
128    /// Record metrics for closing a database transactions.
129    #[cfg(feature = "mdbx")]
130    pub(crate) fn record_closed_transaction(
131        &self,
132        mode: TransactionMode,
133        outcome: TransactionOutcome,
134        open_duration: Duration,
135        close_duration: Option<Duration>,
136        commit_latency: Option<reth_libmdbx::CommitLatency>,
137    ) {
138        self.transactions
139            .get(&mode)
140            .expect("transaction mode metric handle not found")
141            .record_close();
142
143        self.transaction_outcomes
144            .get(&(mode, outcome))
145            .expect("transaction outcome metric handle not found")
146            .record(open_duration, close_duration, commit_latency);
147    }
148}
149
150/// Transaction mode for the database, either read-only or read-write.
151#[derive(Debug, Clone, Copy, Eq, PartialEq, Hash, EnumCount, EnumIter)]
152pub(crate) enum TransactionMode {
153    /// Read-only transaction mode.
154    ReadOnly,
155    /// Read-write transaction mode.
156    ReadWrite,
157}
158
159impl TransactionMode {
160    /// Returns the transaction mode as a string.
161    pub(crate) const fn as_str(&self) -> &'static str {
162        match self {
163            Self::ReadOnly => "read-only",
164            Self::ReadWrite => "read-write",
165        }
166    }
167
168    /// Returns `true` if the transaction mode is read-only.
169    pub(crate) const fn is_read_only(&self) -> bool {
170        matches!(self, Self::ReadOnly)
171    }
172}
173
174/// Transaction outcome after a database operation - commit, abort, or drop.
175#[derive(Debug, Clone, Copy, Eq, PartialEq, Hash, EnumCount, EnumIter)]
176pub(crate) enum TransactionOutcome {
177    /// Successful commit of the transaction.
178    Commit,
179    /// Aborted transaction.
180    Abort,
181    /// Dropped transaction.
182    Drop,
183}
184
185impl TransactionOutcome {
186    /// Returns the transaction outcome as a string.
187    pub(crate) const fn as_str(&self) -> &'static str {
188        match self {
189            Self::Commit => "commit",
190            Self::Abort => "abort",
191            Self::Drop => "drop",
192        }
193    }
194
195    /// Returns `true` if the transaction outcome is a commit.
196    pub(crate) const fn is_commit(&self) -> bool {
197        matches!(self, Self::Commit)
198    }
199}
200
201/// Types of operations conducted on the database: get, put, delete, and various cursor operations.
202#[derive(Debug, Clone, Copy, Eq, PartialEq, Hash, EnumCount, EnumIter)]
203pub(crate) enum Operation {
204    /// Database get operation.
205    Get,
206    /// Database put upsert operation.
207    PutUpsert,
208    /// Database put append operation.
209    PutAppend,
210    /// Database delete operation.
211    Delete,
212    /// Database cursor upsert operation.
213    CursorUpsert,
214    /// Database cursor insert operation.
215    CursorInsert,
216    /// Database cursor append operation.
217    CursorAppend,
218    /// Database cursor append duplicates operation.
219    CursorAppendDup,
220    /// Database cursor delete current operation.
221    CursorDeleteCurrent,
222    /// Database cursor delete current duplicates operation.
223    CursorDeleteCurrentDuplicates,
224}
225
226impl Operation {
227    /// Returns the index of the operation in the cached per-table operation array.
228    pub(crate) const fn index(&self) -> usize {
229        match self {
230            Self::Get => 0,
231            Self::PutUpsert => 1,
232            Self::PutAppend => 2,
233            Self::Delete => 3,
234            Self::CursorUpsert => 4,
235            Self::CursorInsert => 5,
236            Self::CursorAppend => 6,
237            Self::CursorAppendDup => 7,
238            Self::CursorDeleteCurrent => 8,
239            Self::CursorDeleteCurrentDuplicates => 9,
240        }
241    }
242
243    /// Returns the operation for the given index in the cached per-table operation array.
244    const fn from_index(index: usize) -> Self {
245        match index {
246            0 => Self::Get,
247            1 => Self::PutUpsert,
248            2 => Self::PutAppend,
249            3 => Self::Delete,
250            4 => Self::CursorUpsert,
251            5 => Self::CursorInsert,
252            6 => Self::CursorAppend,
253            7 => Self::CursorAppendDup,
254            8 => Self::CursorDeleteCurrent,
255            9 => Self::CursorDeleteCurrentDuplicates,
256            _ => panic!("invalid operation index"),
257        }
258    }
259
260    /// Returns the operation as a string.
261    pub(crate) const fn as_str(&self) -> &'static str {
262        match self {
263            Self::Get => "get",
264            Self::PutUpsert => "put-upsert",
265            Self::PutAppend => "put-append",
266            Self::Delete => "delete",
267            Self::CursorUpsert => "cursor-upsert",
268            Self::CursorInsert => "cursor-insert",
269            Self::CursorAppend => "cursor-append",
270            Self::CursorAppendDup => "cursor-append-dup",
271            Self::CursorDeleteCurrent => "cursor-delete-current",
272            Self::CursorDeleteCurrentDuplicates => "cursor-delete-current-duplicates",
273        }
274    }
275}
276
277/// Enum defining labels for various aspects used in metrics.
278enum Labels {
279    /// Label representing a table.
280    Table,
281    /// Label representing a transaction mode.
282    TransactionMode,
283    /// Label representing a transaction outcome.
284    TransactionOutcome,
285    /// Label representing a database operation.
286    Operation,
287}
288
289impl Labels {
290    /// Converts each label variant into its corresponding string representation.
291    pub(crate) const fn as_str(&self) -> &'static str {
292        match self {
293            Self::Table => "table",
294            Self::TransactionMode => "mode",
295            Self::TransactionOutcome => "outcome",
296            Self::Operation => "operation",
297        }
298    }
299}
300
301#[derive(Metrics, Clone)]
302#[metrics(scope = "database.transaction")]
303pub(crate) struct TransactionMetrics {
304    /// Total number of opened database transactions (cumulative)
305    opened_total: Counter,
306    /// Total number of closed database transactions (cumulative)
307    closed_total: Counter,
308}
309
310impl TransactionMetrics {
311    pub(crate) fn record_open(&self) {
312        self.opened_total.increment(1);
313    }
314
315    pub(crate) fn record_close(&self) {
316        self.closed_total.increment(1);
317    }
318}
319
320#[derive(Metrics, Clone)]
321#[metrics(scope = "database.transaction")]
322pub(crate) struct TransactionOutcomeMetrics {
323    /// The time a database transaction has been open
324    open_duration_seconds: Histogram,
325    /// The time it took to close a database transaction
326    close_duration_seconds: Histogram,
327    /// The time it took to prepare a transaction commit
328    commit_preparation_duration_seconds: Histogram,
329    /// Duration of GC update during transaction commit by wall clock
330    commit_gc_wallclock_duration_seconds: Histogram,
331    /// The time it took to conduct audit of a transaction commit
332    commit_audit_duration_seconds: Histogram,
333    /// The time it took to write dirty/modified data pages to a filesystem during transaction
334    /// commit
335    commit_write_duration_seconds: Histogram,
336    /// The time it took to sync written data to the disk/storage during transaction commit
337    commit_sync_duration_seconds: Histogram,
338    /// The time it took to release resources during transaction commit
339    commit_ending_duration_seconds: Histogram,
340    /// The total duration of a transaction commit
341    commit_whole_duration_seconds: Histogram,
342    /// User-mode CPU time spent on GC update during transaction commit
343    commit_gc_cputime_duration_seconds: Histogram,
344}
345
346impl TransactionOutcomeMetrics {
347    /// Record transaction closing with the duration it was open and the duration it took to close
348    /// it.
349    #[cfg(feature = "mdbx")]
350    pub(crate) fn record(
351        &self,
352        open_duration: Duration,
353        close_duration: Option<Duration>,
354        commit_latency: Option<reth_libmdbx::CommitLatency>,
355    ) {
356        self.open_duration_seconds.record(open_duration);
357
358        if let Some(close_duration) = close_duration {
359            self.close_duration_seconds.record(close_duration)
360        }
361
362        if let Some(commit_latency) = commit_latency {
363            self.commit_preparation_duration_seconds.record(commit_latency.preparation());
364            self.commit_gc_wallclock_duration_seconds.record(commit_latency.gc_wallclock());
365            self.commit_audit_duration_seconds.record(commit_latency.audit());
366            self.commit_write_duration_seconds.record(commit_latency.write());
367            self.commit_sync_duration_seconds.record(commit_latency.sync());
368            self.commit_ending_duration_seconds.record(commit_latency.ending());
369            self.commit_whole_duration_seconds.record(commit_latency.whole());
370            self.commit_gc_cputime_duration_seconds.record(commit_latency.gc_cputime());
371        }
372    }
373}
374
375#[derive(Metrics, Clone)]
376#[metrics(scope = "database.operation")]
377pub(crate) struct OperationMetrics {
378    /// Total number of database operations made
379    calls_total: Counter,
380    /// The time it took to execute a database operation (`put/upsert/insert/append/append_dup`)
381    /// with value larger than [`LARGE_VALUE_THRESHOLD_BYTES`] bytes.
382    large_value_duration_seconds: Histogram,
383}
384
385impl OperationMetrics {
386    /// Record operation metric.
387    ///
388    /// The duration it took to execute the closure is recorded only if the provided `value_size` is
389    /// larger than [`LARGE_VALUE_THRESHOLD_BYTES`].
390    pub(crate) fn record<R>(&self, value_size: Option<usize>, f: impl FnOnce() -> R) -> R {
391        self.calls_total.increment(1);
392
393        // Record duration only for large values to prevent the performance hit of clock syscall
394        // on small operations
395        if value_size.is_some_and(|size| size > LARGE_VALUE_THRESHOLD_BYTES) {
396            let start = Instant::now();
397            let result = f();
398            self.large_value_duration_seconds.record(start.elapsed());
399            result
400        } else {
401            f()
402        }
403    }
404}
reth_db/metrics.rs

reth_db/
metrics.rs
