Merge remote-tracking branch 'origin/migrate-quickwit-metrics-to-metrics-rs' into migrate-quickwit-metrics-to-metrics-rs

# Conflicts:
#	quickwit/quickwit-lambda-client/src/invoker.rs

Author: shuheiktgw

quickwit/quickwit-actors/src/mailbox.rs

@@ -16,10 +16,10 @@ use std::any::Any;
 use std::convert::Infallible;
 use std::fmt;
 use std::sync::atomic::{AtomicUsize, Ordering};
-use std::sync::{Arc, LazyLock, Weak};
+use std::sync::{Arc, Weak};
 use std::time::Instant;
 
-use quickwit_metrics::{Counter, Gauge, GaugeGuard, gauge};
+use quickwit_metrics::{Counter, GaugeGuard, LazyGauge, lazy_gauge};
 use tokio::sync::oneshot;
 
 use crate::channel_with_priority::{Receiver, Sender, TrySendError};
@@ -205,7 +205,7 @@ impl<A: Actor> Mailbox<A> {
                     let now = Instant::now();
                     self.inner.tx.send_low_priority(envelope).await?;
                     let elapsed = now.elapsed();
-                    backpressure_micros_counter.increment(elapsed.as_micros() as u64);
+                    backpressure_micros_counter.inc_by(elapsed.as_micros() as u64);
                 } else {
                     self.inner.tx.send_low_priority(envelope).await?;
                 }
@@ -311,13 +311,11 @@ struct InboxInner<A: Actor> {
     _inboxes_count_gauge_guard: GaugeGuard,
 }
 
-static INBOX_GAUGE: LazyLock<Gauge> = LazyLock::new(|| {
-    gauge!(
+static INBOX_GAUGE: LazyGauge = lazy_gauge!(
         name: "inboxes_count",
         description: "overall count of actors",
         subsystem: "actor",
-    )
-});
+);
 
 pub struct Inbox<A: Actor> {
     inner: Arc<InboxInner<A>>,

quickwit/quickwit-cli/src/lib.rs

@@ -387,7 +387,7 @@ pub mod busy_detector {
                 .unwrap_or_default();
             let now = now.as_micros() as u64;
             let delta = now - time.load(Ordering::Relaxed);
-            THREAD_UNPARK_DURATION_MICROSECONDS.record(delta as f64);
+            THREAD_UNPARK_DURATION_MICROSECONDS.observe(delta as f64);
             if delta > ALLOWED_DELAY_MICROS {
                 emit_debug(delta, now);
             }

quickwit/quickwit-cli/src/metrics.rs

@@ -12,19 +12,15 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::sync::LazyLock;
-
 use quickwit_common::metrics::exponential_buckets;
-use quickwit_metrics::{Counter, Histogram, counter, histogram, labels};
+use quickwit_metrics::{LazyCounter, LazyHistogram, counter, labels, lazy_counter, lazy_histogram};
 use quickwit_serve::BuildInfo;
 
-static BUILD_INFO: LazyLock<Counter> = LazyLock::new(|| {
-    counter!(
+static BUILD_INFO: LazyCounter = lazy_counter!(
         name: "build_info",
         description: "Quickwit's build info",
         subsystem: "",
-    )
-});
+);
 pub fn register_build_info_metric(build_info: &BuildInfo) {
     let commit_tags = build_info.commit_tags.join(",");
     let labels = labels!(
@@ -34,14 +30,12 @@ pub fn register_build_info_metric(build_info: &BuildInfo) {
         "commit_tags" => commit_tags,
         "target" => build_info.build_target,
     );
-    counter!(parent: BUILD_INFO, labels: [labels]).increment(1);
+    counter!(parent: BUILD_INFO, labels: [labels]).inc();
 }
 
-pub(crate) static THREAD_UNPARK_DURATION_MICROSECONDS: LazyLock<Histogram> = LazyLock::new(|| {
-    histogram!(
+pub(crate) static THREAD_UNPARK_DURATION_MICROSECONDS: LazyHistogram = lazy_histogram!(
         name: "thread_unpark_duration_microseconds",
         description: "Duration for which a thread of the main tokio runtime is unparked.",
         subsystem: "cli",
         buckets: exponential_buckets(5.0, 5.0, 5).unwrap(),
-    )
-});
+);

quickwit/quickwit-cluster/src/grpc_gossip.rs

@@ -108,7 +108,7 @@ async fn perform_grpc_gossip_rounds<ClusterServiceClientFactory, Fut>(
             warn!("failed to fetch cluster state from node `{node_id}`");
             continue;
         };
-        GRPC_GOSSIP_ROUNDS_TOTAL.increment(1);
+        GRPC_GOSSIP_ROUNDS_TOTAL.inc();
 
         let mut chitchat_guard = chitchat.lock().await;
 

quickwit/quickwit-cluster/src/lib.rs

@@ -84,16 +84,16 @@ impl Socket for CountingUdpSocket {
     async fn send(&mut self, to: SocketAddr, msg: ChitchatMessage) -> anyhow::Result<()> {
         let msg_len = msg.serialized_len() as u64;
         self.socket.send(to, msg).await?;
-        GOSSIP_SENT_MESSAGES_TOTAL.increment(1);
-        GOSSIP_SENT_BYTES_TOTAL.increment(msg_len);
+        GOSSIP_SENT_MESSAGES_TOTAL.inc();
+        GOSSIP_SENT_BYTES_TOTAL.inc_by(msg_len);
         Ok(())
     }
 
     async fn recv(&mut self) -> anyhow::Result<(SocketAddr, ChitchatMessage)> {
         let (socket_addr, msg) = self.socket.recv().await?;
-        GOSSIP_RECV_MESSAGES_TOTAL.increment(1);
+        GOSSIP_RECV_MESSAGES_TOTAL.inc();
         let msg_len = msg.serialized_len() as u64;
-        GOSSIP_RECV_BYTES_TOTAL.increment(msg_len);
+        GOSSIP_RECV_BYTES_TOTAL.inc_by(msg_len);
         Ok((socket_addr, msg))
     }
 }

quickwit/quickwit-cluster/src/metrics.rs

@@ -14,110 +14,86 @@
 
 use std::collections::HashSet;
 use std::net::SocketAddr;
-use std::sync::{LazyLock, Weak};
+use std::sync::Weak;
 use std::time::Duration;
 
 use chitchat::{Chitchat, ChitchatId};
-use quickwit_metrics::{Counter, Gauge, counter, gauge};
+use quickwit_metrics::{LazyCounter, LazyGauge, lazy_counter, lazy_gauge};
 use tokio::sync::Mutex;
 
 use crate::member::NodeStateExt;
 
-pub(crate) static LIVE_NODES: LazyLock<Gauge> = LazyLock::new(|| {
-    gauge!(
+pub(crate) static LIVE_NODES: LazyGauge = lazy_gauge!(
         name: "live_nodes",
         description: "The number of live nodes observed locally.",
         subsystem: "cluster",
-    )
-});
+);
 
-pub(crate) static READY_NODES: LazyLock<Gauge> = LazyLock::new(|| {
-    gauge!(
+pub(crate) static READY_NODES: LazyGauge = lazy_gauge!(
         name: "ready_nodes",
         description: "The number of ready nodes observed locally.",
         subsystem: "cluster",
-    )
-});
+);
 
-pub(crate) static ZOMBIE_NODES: LazyLock<Gauge> = LazyLock::new(|| {
-    gauge!(
+pub(crate) static ZOMBIE_NODES: LazyGauge = lazy_gauge!(
         name: "zombie_nodes",
         description: "The number of zombie nodes observed locally.",
         subsystem: "cluster",
-    )
-});
+);
 
-pub(crate) static DEAD_NODES: LazyLock<Gauge> = LazyLock::new(|| {
-    gauge!(
+pub(crate) static DEAD_NODES: LazyGauge = lazy_gauge!(
         name: "dead_nodes",
         description: "The number of dead nodes observed locally.",
         subsystem: "cluster",
-    )
-});
+);
 
-pub(crate) static CLUSTER_STATE_SIZE_BYTES: LazyLock<Gauge> = LazyLock::new(|| {
-    gauge!(
+pub(crate) static CLUSTER_STATE_SIZE_BYTES: LazyGauge = lazy_gauge!(
         name: "cluster_state_size_bytes",
         description: "The size of the cluster state in bytes.",
         subsystem: "cluster",
-    )
-});
+);
 
-pub(crate) static NODE_STATE_KEYS: LazyLock<Gauge> = LazyLock::new(|| {
-    gauge!(
+pub(crate) static NODE_STATE_KEYS: LazyGauge = lazy_gauge!(
         name: "node_state_keys",
         description: "The number of keys in the node state.",
         subsystem: "cluster",
-    )
-});
+);
 
-pub(crate) static NODE_STATE_SIZE_BYTES: LazyLock<Gauge> = LazyLock::new(|| {
-    gauge!(
+pub(crate) static NODE_STATE_SIZE_BYTES: LazyGauge = lazy_gauge!(
         name: "node_state_size_bytes",
         description: "The size of the node state in bytes.",
         subsystem: "cluster",
-    )
-});
+);
 
-pub(crate) static GOSSIP_RECV_MESSAGES_TOTAL: LazyLock<Counter> = LazyLock::new(|| {
-    counter!(
+pub(crate) static GOSSIP_RECV_MESSAGES_TOTAL: LazyCounter = lazy_counter!(
         name: "gossip_recv_messages_total",
         description: "Total number of gossip messages received.",
         subsystem: "cluster",
-    )
-});
+);
 
-pub(crate) static GOSSIP_RECV_BYTES_TOTAL: LazyLock<Counter> = LazyLock::new(|| {
-    counter!(
+pub(crate) static GOSSIP_RECV_BYTES_TOTAL: LazyCounter = lazy_counter!(
         name: "gossip_recv_bytes_total",
         description: "Total amount of gossip data received in bytes.",
         subsystem: "cluster",
-    )
-});
+);
 
-pub(crate) static GOSSIP_SENT_MESSAGES_TOTAL: LazyLock<Counter> = LazyLock::new(|| {
-    counter!(
+pub(crate) static GOSSIP_SENT_MESSAGES_TOTAL: LazyCounter = lazy_counter!(
         name: "gossip_sent_messages_total",
         description: "Total number of gossip messages sent.",
         subsystem: "cluster",
-    )
-});
+);
 
-pub(crate) static GOSSIP_SENT_BYTES_TOTAL: LazyLock<Counter> = LazyLock::new(|| {
-    counter!(
+pub(crate) static GOSSIP_SENT_BYTES_TOTAL: LazyCounter = lazy_counter!(
         name: "gossip_sent_bytes_total",
         description: "Total amount of gossip data sent in bytes.",
         subsystem: "cluster",
-    )
-});
+);
 
-pub(crate) static GRPC_GOSSIP_ROUNDS_TOTAL: LazyLock<Counter> = LazyLock::new(|| {
-    counter!(
+pub(crate) static GRPC_GOSSIP_ROUNDS_TOTAL: LazyCounter = lazy_counter!(
         name: "grpc_gossip_rounds_total",
         description: "Total number of gRPC gossip rounds performed with peer nodes.",
         subsystem: "cluster",
-    )
-});
+);
 
 pub(crate) fn spawn_metrics_task(
     weak_chitchat: Weak<Mutex<Chitchat>>,

quickwit/quickwit-common/src/io.rs

@@ -25,7 +25,6 @@ use std::future::Future;
 use std::io;
 use std::io::IoSlice;
 use std::pin::Pin;
-use std::sync::LazyLock;
 use std::task::{Context, Poll};
 use std::time::Duration;
 
@@ -34,7 +33,7 @@ use async_speed_limit::clock::StandardClock;
 use async_speed_limit::limiter::Consume;
 use bytesize::ByteSize;
 use pin_project::pin_project;
-use quickwit_metrics::{Counter, counter};
+use quickwit_metrics::{Counter, LazyCounter, counter, lazy_counter};
 use tokio::io::AsyncWrite;
 
 use crate::{KillSwitch, Progress, ProtectedZoneGuard};
@@ -47,13 +46,11 @@ fn truncate_bytes(bytes: &[u8]) -> &[u8] {
     &bytes[..num_bytes]
 }
 
-static WRITE_BYTES: LazyLock<Counter> = LazyLock::new(|| {
-    counter!(
+static WRITE_BYTES: LazyCounter = lazy_counter!(
         name: "write_bytes",
         description: "Number of bytes written by a given component in [indexer, merger, deleter, split_downloader_{merge,delete}]",
         subsystem: "",
-    )
-});
+);
 
 /// Parameter used in `async_speed_limit`.
 ///
@@ -155,7 +152,7 @@ impl IoControls {
         if let Some(throughput_limiter) = &self.throughput_limiter_opt {
             throughput_limiter.blocking_consume(num_bytes);
         }
-        self.bytes_counter.increment(num_bytes as u64);
+        self.bytes_counter.inc_by(num_bytes as u64);
         Ok(())
     }
 }
@@ -208,7 +205,7 @@ impl<A: IoControlsAccess, W: AsyncWrite> ControlledWrite<A, W> {
             let len = *obj.as_ref().unwrap_or(&0);
             if len > 0 {
                 let waiter = this.io_controls_access.apply(|io_controls| {
-                    io_controls.bytes_counter.increment(len as u64);
+                    io_controls.bytes_counter.inc_by(len as u64);
                     io_controls
                         .throughput_limiter_opt
                         .as_ref()