jleony/nanoreth
1
0
Fork 0
mirror of https://github.com/hl-archive-node/nanoreth.git synced 2025-12-06 10:59:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore: add _total suffix to counter metrics (#8263)

Browse Source 
Co-authored-by: Matthias Seitz <matthias.seitz@outlook.de>
This commit is contained in:
Håvard Anda Estensen
2024-05-29 17:05:31 +02:00
committed by GitHub
parent d7e822f9d6
commit 944180348f
9 changed files with 58 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
crates/consensus/beacon/src/engine/hooks/prune.rs
View File

@ -99,7 +99,7 @@ impl<DB: Database + 'static> PruneHook<DB> {
let _ = tx.send((pruner, result));
}),
);
self.metrics.runs.increment(1);
self.metrics.runs_total.increment(1);
self.pruner_state = PrunerState::Running(rx);
Some(EngineHookEvent::Started)
@ -160,7 +160,7 @@ enum PrunerState<DB> {
#[metrics(scope = "consensus.engine.prune")]
struct Metrics {
/// The number of times the pruner was run.
runs: Counter,
runs_total: Counter,
}
impl From<PrunerError> for EngineHookError {

38
crates/metrics/src/common/mpsc.rs
View File

@ -55,11 +55,11 @@ impl<T> UnboundedMeteredSender<T> {
pub fn send(&self, message: T) -> Result<(), SendError<T>> {
match self.sender.send(message) {
Ok(()) => {
self.metrics.messages_sent.increment(1);
self.metrics.messages_sent_total.increment(1);
Ok(())
}
Err(error) => {
self.metrics.send_errors.increment(1);
self.metrics.send_errors_total.increment(1);
Err(error)
}
}
@ -94,7 +94,7 @@ impl<T> UnboundedMeteredReceiver<T> {
pub async fn recv(&mut self) -> Option<T> {
let msg = self.receiver.recv().await;
if msg.is_some() {
self.metrics.messages_received.increment(1);
self.metrics.messages_received_total.increment(1);
}
msg
}
@ -102,7 +102,7 @@ impl<T> UnboundedMeteredReceiver<T> {
/// Tries to receive the next value for this receiver.
pub fn try_recv(&mut self) -> Result<T, TryRecvError> {
let msg = self.receiver.try_recv()?;
self.metrics.messages_received.increment(1);
self.metrics.messages_received_total.increment(1);
Ok(msg)
}
@ -115,7 +115,7 @@ impl<T> UnboundedMeteredReceiver<T> {
pub fn poll_recv(&mut self, cx: &mut Context<'_>) -> Poll<Option<T>> {
let msg = ready!(self.receiver.poll_recv(cx));
if msg.is_some() {
self.metrics.messages_received.increment(1);
self.metrics.messages_received_total.increment(1);
}
Poll::Ready(msg)
}
@ -161,11 +161,11 @@ impl<T> MeteredSender<T> {
pub fn try_send(&self, message: T) -> Result<(), TrySendError<T>> {
match self.sender.try_send(message) {
Ok(()) => {
self.metrics.messages_sent.increment(1);
self.metrics.messages_sent_total.increment(1);
Ok(())
}
Err(error) => {
self.metrics.send_errors.increment(1);
self.metrics.send_errors_total.increment(1);
Err(error)
}
}
@ -176,11 +176,11 @@ impl<T> MeteredSender<T> {
pub async fn send(&self, value: T) -> Result<(), SendError<T>> {
match self.sender.send(value).await {
Ok(()) => {
self.metrics.messages_sent.increment(1);
self.metrics.messages_sent_total.increment(1);
Ok(())
}
Err(error) => {
self.metrics.send_errors.increment(1);
self.metrics.send_errors_total.increment(1);
Err(error)
}
}
@ -214,7 +214,7 @@ impl<T> MeteredReceiver<T> {
pub async fn recv(&mut self) -> Option<T> {
let msg = self.receiver.recv().await;
if msg.is_some() {
self.metrics.messages_received.increment(1);
self.metrics.messages_received_total.increment(1);
}
msg
}
@ -222,7 +222,7 @@ impl<T> MeteredReceiver<T> {
/// Tries to receive the next value for this receiver.
pub fn try_recv(&mut self) -> Result<T, TryRecvError> {
let msg = self.receiver.try_recv()?;
self.metrics.messages_received.increment(1);
self.metrics.messages_received_total.increment(1);
Ok(msg)
}
@ -235,7 +235,7 @@ impl<T> MeteredReceiver<T> {
pub fn poll_recv(&mut self, cx: &mut Context<'_>) -> Poll<Option<T>> {
let msg = ready!(self.receiver.poll_recv(cx));
if msg.is_some() {
self.metrics.messages_received.increment(1);
self.metrics.messages_received_total.increment(1);
}
Poll::Ready(msg)
}
@ -254,9 +254,9 @@ impl<T> Stream for MeteredReceiver<T> {
#[metrics(dynamic = true)]
struct MeteredSenderMetrics {
/// Number of messages sent
messages_sent: Counter,
messages_sent_total: Counter,
/// Number of failed message deliveries
send_errors: Counter,
send_errors_total: Counter,
}
/// Throughput metrics for [MeteredReceiver]
@ -264,7 +264,7 @@ struct MeteredSenderMetrics {
#[metrics(dynamic = true)]
struct MeteredReceiverMetrics {
/// Number of messages received
messages_received: Counter,
messages_received_total: Counter,
}
/// A wrapper type around [PollSender] that updates metrics on send.
@ -294,7 +294,7 @@ impl<T: Send + 'static> MeteredPollSender<T> {
Poll::Ready(Ok(permit)) => Poll::Ready(Ok(permit)),
Poll::Ready(Err(error)) => Poll::Ready(Err(error)),
Poll::Pending => {
self.metrics.back_pressure.increment(1);
self.metrics.back_pressure_total.increment(1);
Poll::Pending
}
}
@ -305,7 +305,7 @@ impl<T: Send + 'static> MeteredPollSender<T> {
pub fn send_item(&mut self, item: T) -> Result<(), PollSendError<T>> {
match self.sender.send_item(item) {
Ok(()) => {
self.metrics.messages_sent.increment(1);
self.metrics.messages_sent_total.increment(1);
Ok(())
}
Err(error) => Err(error),
@ -324,7 +324,7 @@ impl<T> Clone for MeteredPollSender<T> {
#[metrics(dynamic = true)]
struct MeteredPollSenderMetrics {
/// Number of messages sent
messages_sent: Counter,
messages_sent_total: Counter,
/// Number of delayed message deliveries caused by a full channel
back_pressure: Counter,
back_pressure_total: Counter,
}

28
crates/rpc/rpc-builder/src/metrics.rs
View File

@ -90,7 +90,7 @@ pub(crate) struct RpcRequestMetricsService<S> {
impl<S> RpcRequestMetricsService<S> {
pub(crate) fn new(service: S, metrics: RpcRequestMetrics) -> Self {
// this instance is kept alive for the duration of the connection
metrics.inner.connection_metrics.connections_opened.increment(1);
metrics.inner.connection_metrics.connections_opened_total.increment(1);
Self { inner: service, metrics }
}
}
@ -102,10 +102,10 @@ where
type Future = MeteredRequestFuture<S::Future>;
fn call(&self, req: Request<'a>) -> Self::Future {
self.metrics.inner.connection_metrics.requests_started.increment(1);
self.metrics.inner.connection_metrics.requests_started_total.increment(1);
let call_metrics = self.metrics.inner.call_metrics.get_key_value(req.method.as_ref());
if let Some((_, call_metrics)) = &call_metrics {
call_metrics.started.increment(1);
call_metrics.started_total.increment(1);
}
MeteredRequestFuture {
fut: self.inner.call(req),
@ -119,7 +119,7 @@ where
impl<S> Drop for RpcRequestMetricsService<S> {
fn drop(&mut self) {
// update connection metrics, connection closed
self.metrics.inner.connection_metrics.connections_closed.increment(1);
self.metrics.inner.connection_metrics.connections_closed_total.increment(1);
}
}
@ -153,7 +153,7 @@ impl<F: Future<Output = MethodResponse>> Future for MeteredRequestFuture<F> {
let elapsed = this.started_at.elapsed().as_secs_f64();
// update transport metrics
this.metrics.inner.connection_metrics.requests_finished.increment(1);
this.metrics.inner.connection_metrics.requests_finished_total.increment(1);
this.metrics.inner.connection_metrics.request_time_seconds.record(elapsed);
// update call metrics
@ -162,9 +162,9 @@ impl<F: Future<Output = MethodResponse>> Future for MeteredRequestFuture<F> {
{
call_metrics.time_seconds.record(elapsed);
if resp.is_success() {
call_metrics.successful.increment(1);
call_metrics.successful_total.increment(1);
} else {
call_metrics.failed.increment(1);
call_metrics.failed_total.increment(1);
}
}
}
@ -202,13 +202,13 @@ impl RpcTransport {
#[metrics(scope = "rpc_server.connections")]
struct RpcServerConnectionMetrics {
/// The number of connections opened
connections_opened: Counter,
connections_opened_total: Counter,
/// The number of connections closed
connections_closed: Counter,
connections_closed_total: Counter,
/// The number of requests started
requests_started: Counter,
requests_started_total: Counter,
/// The number of requests finished
requests_finished: Counter,
requests_finished_total: Counter,
/// Response for a single request/response pair
request_time_seconds: Histogram,
}
@ -218,11 +218,11 @@ struct RpcServerConnectionMetrics {
#[metrics(scope = "rpc_server.calls")]
struct RpcServerCallMetrics {
/// The number of calls started
started: Counter,
started_total: Counter,
/// The number of successful calls
successful: Counter,
successful_total: Counter,
/// The number of failed calls
failed: Counter,
failed_total: Counter,
/// Response for a single call
time_seconds: Histogram,
}

4
crates/rpc/rpc/src/eth/cache/metrics.rs vendored
View File

@ -9,7 +9,7 @@ pub(crate) struct CacheMetrics {
/// The number of queued consumers.
pub(crate) queued_consumers_count: Gauge,
/// The number of cache hits.
pub(crate) hits: Counter,
pub(crate) hits_total: Counter,
/// The number of cache misses.
pub(crate) misses: Counter,
pub(crate) misses_total: Counter,
}

4
crates/rpc/rpc/src/eth/cache/multi_consumer.rs vendored
View File

@ -72,9 +72,9 @@ where
pub fn get(&mut self, key: &K) -> Option<&mut V> {
let entry = self.cache.get(key);
if entry.is_some() {
self.metrics.hits.increment(1);
self.metrics.hits_total.increment(1);
} else {
self.metrics.misses.increment(1);
self.metrics.misses_total.increment(1);
}
entry
}

11
crates/tasks/src/lib.rs
View File

@ -328,12 +328,14 @@ impl TaskExecutor {
let on_shutdown = self.on_shutdown.clone();
// Clone only the specific counter that we need.
let finished_regular_tasks_metrics = self.metrics.finished_regular_tasks.clone();
let finished_regular_tasks_total_metrics =
self.metrics.finished_regular_tasks_total.clone();
// Wrap the original future to increment the finished tasks counter upon completion
let task = {
async move {
// Create an instance of IncCounterOnDrop with the counter to increment
let _inc_counter_on_drop = IncCounterOnDrop::new(finished_regular_tasks_metrics);
let _inc_counter_on_drop =
IncCounterOnDrop::new(finished_regular_tasks_total_metrics);
let fut = pin!(fut);
let _ = select(on_shutdown, fut).await;
}
@ -405,10 +407,11 @@ impl TaskExecutor {
.in_current_span();
// Clone only the specific counter that we need.
let finished_critical_tasks_metrics = self.metrics.finished_critical_tasks.clone();
let finished_critical_tasks_total_metrics =
self.metrics.finished_critical_tasks_total.clone();
let task = async move {
// Create an instance of IncCounterOnDrop with the counter to increment
let _inc_counter_on_drop = IncCounterOnDrop::new(finished_critical_tasks_metrics);
let _inc_counter_on_drop = IncCounterOnDrop::new(finished_critical_tasks_total_metrics);
let task = pin!(task);
let _ = select(on_shutdown, task).await;
};

12
crates/tasks/src/metrics.rs
View File

@ -9,24 +9,24 @@ use reth_metrics::{metrics::Counter, Metrics};
#[metrics(scope = "executor.spawn")]
pub struct TaskExecutorMetrics {
/// Number of spawned critical tasks
pub(crate) critical_tasks: Counter,
pub(crate) critical_tasks_total: Counter,
/// Number of finished spawned critical tasks
pub(crate) finished_critical_tasks: Counter,
pub(crate) finished_critical_tasks_total: Counter,
/// Number of spawned regular tasks
pub(crate) regular_tasks: Counter,
pub(crate) regular_tasks_total: Counter,
/// Number of finished spawned regular tasks
pub(crate) finished_regular_tasks: Counter,
pub(crate) finished_regular_tasks_total: Counter,
}
impl TaskExecutorMetrics {
/// Increments the counter for spawned critical tasks.
pub(crate) fn inc_critical_tasks(&self) {
self.critical_tasks.increment(1);
self.critical_tasks_total.increment(1);
}
/// Increments the counter for spawned regular tasks.
pub(crate) fn inc_regular_tasks(&self) {
self.regular_tasks.increment(1);
self.regular_tasks_total.increment(1);
}
}