411be70
use std::sync::Once;
use metrics::{counter, describe_counter};
use crate::error::QueueError;
pub(crate) const OUTCOME_SUCCESS: &str = "success";
static DESCRIBE_ONCE: Once = Once::new();
fn ensure_described() {
DESCRIBE_ONCE.call_once(describe_all);
}
fn describe_all() {
describe_counter!(
"queue_publish_total",
"Messages published to Kafka partitioned by outcome."
);
describe_counter!(
"queue_message_received_total",
"Messages received from Kafka by subscribers."
);
describe_counter!(
"queue_message_processed_total",
"Messages handled by subscriber handlers partitioned by outcome."
);
describe_counter!(
"queue_retries_total",
"Retries of failed handler executions."
);
describe_counter!(
"queue_message_dropped_total",
"Messages dropped after exhausting handler retries."
);
}
fn outcome_for<T>(result: &Result<T, QueueError>) -> &'static str {
match result {
Ok(_) => OUTCOME_SUCCESS,
Err(QueueError::Kafka(_)) => "kafka",
Err(QueueError::Config(_)) => "config",
Err(QueueError::Topic(_)) => "topic",
Err(QueueError::Serde(_)) => "serde",
Err(QueueError::Handler(_)) => "handler",
Err(QueueError::Join(_)) => "join",
Err(QueueError::Disabled) => "disabled",
Err(QueueError::Cancelled) => "cancelled",
}
}
pub(crate) fn record_publish<T>(result: &Result<T, QueueError>) {
ensure_described();
counter!("queue_publish_total", "outcome" => outcome_for(result)).increment(1);
}
pub(crate) fn record_message_received() {
ensure_described();
counter!("queue_message_received_total").increment(1);
}
pub(crate) fn record_message_processed(ok: bool) {
ensure_described();
counter!(
"queue_message_processed_total",
"outcome" => if ok { OUTCOME_SUCCESS } else { "error" }
)
.increment(1);
}
pub(crate) fn record_retry() {
ensure_described();
counter!("queue_retries_total").increment(1);
}
pub(crate) fn record_message_dropped() {
ensure_described();
counter!("queue_message_dropped_total").increment(1);
}