Merge pull request #3869 from eval-exec/feature/drop_opentelemetry_us…

…e_tikv_prometheus

Replace opentelemetry by tikv/rust-prometheus

Cargo.toml

@@ -30,7 +30,6 @@ members = [
     "util/build-info",
     "util/logger",
     "util/logger-config",
-    "util/metrics",
     "util/metrics-config",
     "util/hash",
     "util/rational",
@@ -43,6 +42,7 @@ members = [
     "util/logger-service",
     "util/stop-handler",
     "util/runtime",
+    "util/metrics",
     "util/metrics-service",
     "util/fixed-hash",
     "util/occupied-capacity",

chain/src/chain.rs

@@ -8,7 +8,6 @@ use ckb_logger::{
     self, debug, error, info, log_enabled, log_enabled_target, trace, trace_target, warn,
 };
 use ckb_merkle_mountain_range::leaf_index_to_mmr_size;
-use ckb_metrics::metrics;
 use ckb_proposal_table::ProposalTable;
 #[cfg(debug_assertions)]
 use ckb_rust_unstable_port::IsSorted;
@@ -503,7 +502,9 @@ impl ChainService {
             if log_enabled!(ckb_logger::Level::Debug) {
                 self.print_chain(10);
             }
-            metrics!(gauge, "ckb.chain_tip", block.header().number() as i64);
+            if let Some(metrics) = ckb_metrics::handle() {
+                metrics.ckb_chain_tip.set(block.header().number() as i64);
+            }
         } else {
             self.shared.refresh_snapshot();
             info!(

freezer/src/freezer_files.rs

@@ -1,4 +1,3 @@
-use ckb_metrics::metrics;
 use fail::fail_point;
 use lru::LruCache;
 use snap::raw::{Decoder as SnappyDecoder, Encoder as SnappyEncoder};
@@ -151,12 +150,11 @@ impl FreezerFiles {
         self.write_index(self.head_id, self.head.bytes)?;
         self.number.fetch_add(1, Ordering::SeqCst);
 
-        //Gauge for tracking the size of all frozen data
-        metrics!(
-            gauge,
-            "ckb-freezer.size",
-            data_size as i64 + INDEX_ENTRY_SIZE as i64
-        );
+        if let Some(metrics) = ckb_metrics::handle() {
+            metrics
+                .ckb_freezer_size
+                .set(data_size as i64 + INDEX_ENTRY_SIZE as i64);
+        }
         Ok(())
     }
 
@@ -203,12 +201,11 @@ impl FreezerFiles {
                 })?;
             }
 
-            // Meter for measuring the effective amount of data read
-            metrics!(
-                counter,
-                "ckb-freezer.read",
-                size as u64 + 2 * INDEX_ENTRY_SIZE
-            );
+            if let Some(metrics) = ckb_metrics::handle() {
+                metrics
+                    .ckb_freezer_read
+                    .inc_by(size as u64 + 2 * INDEX_ENTRY_SIZE);
+            }
             Ok(Some(data))
         } else {
             Ok(None)

network/src/network.rs

@@ -20,7 +20,6 @@ use crate::services::{
 use crate::{Behaviour, CKBProtocol, Peer, PeerIndex, ProtocolId, ServiceControl};
 use ckb_app_config::{default_support_all_protocols, NetworkConfig, SupportProtocol};
 use ckb_logger::{debug, error, info, trace, warn};
-use ckb_metrics::metrics;
 use ckb_spawn::Spawn;
 use ckb_stop_handler::{SignalSender, StopHandler};
 use ckb_util::{Condvar, Mutex, RwLock};
@@ -192,14 +191,9 @@ impl NetworkState {
                 duration.as_secs(),
                 reason
             );
-            metrics!(
-                counter,
-                "ckb.network.ban_peer",
-                1,
-                "peer_addr" => addr.to_string(),
-                "duration" => duration.as_secs().to_string(),
-                "reason" => reason.clone(),
-            );
+            if let Some(metrics) = ckb_metrics::handle() {
+                metrics.ckb_network_ban_peer.inc();
+            }
             if let Some(peer) = self.with_peer_registry_mut(|reg| reg.remove_peer(session_id)) {
                 let message = format!("Ban for {} seconds, reason: {}", duration.as_secs(), reason);
                 self.peer_store.lock().ban_addr(

sync/src/filter/mod.rs

@@ -7,9 +7,9 @@ use get_block_filter_check_points_process::GetBlockFilterCheckPointsProcess;
 use get_block_filter_hashes_process::GetBlockFilterHashesProcess;
 use get_block_filters_process::GetBlockFiltersProcess;
 
+use crate::utils::{metric_ckb_message_bytes, MetricDirection};
 use ckb_constant::sync::BAD_MESSAGE_BAN_TIME;
 use ckb_logger::{debug_target, error_target, info_target, warn_target};
-use ckb_metrics::metrics;
 use ckb_network::{
     async_trait, bytes::Bytes, CKBProtocolContext, CKBProtocolHandler, PeerIndex, SupportProtocols,
 };
@@ -71,14 +71,12 @@ impl BlockFilter {
         let item_bytes = message.as_slice().len() as u64;
         let status = self.try_process(Arc::clone(&nc), peer, message);
 
-        metrics!(
-            counter,
-            "ckb.messages_bytes",
+        metric_ckb_message_bytes(
+            MetricDirection::In,
+            &SupportProtocols::Filter.name(),
+            message.item_name(),
+            Some(status.code()),
             item_bytes,
-            "direction" => "in",
-            "protocol_id" => SupportProtocols::Filter.protocol_id().value().to_string(),
-            "item_id" => message.item_id().to_string(),
-            "status" => (status.code() as u16).to_string(),
         );
 
         if let Some(ban_time) = status.should_ban() {

sync/src/relayer/compact_block_process.rs

@@ -7,7 +7,6 @@ use crate::SyncShared;
 use crate::{attempt, Status, StatusCode};
 use ckb_chain_spec::consensus::Consensus;
 use ckb_logger::{self, debug_target};
-use ckb_metrics::metrics;
 use ckb_network::{CKBProtocolContext, PeerIndex};
 use ckb_systemtime::unix_time_as_millis;
 use ckb_traits::HeaderProvider;
@@ -96,12 +95,12 @@ impl<'a> CompactBlockProcess<'a> {
         // into database
         match ret {
             ReconstructionResult::Block(block) => {
-                metrics!(
-                    counter,
-                    "ckb.relay.cb_transaction_count",
-                    block.transactions().len() as u64
-                );
-                metrics!(counter, "ckb.relay.cb_reconstruct_ok", 1);
+                if let Some(metrics) = ckb_metrics::handle() {
+                    metrics
+                        .ckb_relay_cb_transaction_count
+                        .inc_by(block.transactions().len() as u64);
+                    metrics.ckb_relay_cb_reconstruct_ok.inc();
+                }
 
                 pending_compact_blocks.remove(&block_hash);
                 // remove all pending request below this block epoch
@@ -124,12 +123,13 @@ impl<'a> CompactBlockProcess<'a> {
             ReconstructionResult::Missing(transactions, uncles) => {
                 let missing_transactions: Vec<u32> =
                     transactions.into_iter().map(|i| i as u32).collect();
-                metrics!(
-                    counter,
-                    "ckb.relay.cb_fresh_tx_cnt",
-                    missing_transactions.len() as u64
-                );
-                metrics!(counter, "ckb.relay.cb_reconstruct_fail", 1);
+
+                if let Some(metrics) = ckb_metrics::handle() {
+                    metrics
+                        .ckb_relay_cb_fresh_tx_cnt
+                        .inc_by(missing_transactions.len() as u64);
+                    metrics.ckb_relay_cb_reconstruct_fail.inc();
+                }
 
                 let missing_uncles: Vec<u32> = uncles.into_iter().map(|i| i as u32).collect();
                 missing_or_collided_post_process(

sync/src/relayer/mod.rs

@@ -22,12 +22,11 @@ use self::transaction_hashes_process::TransactionHashesProcess;
 use self::transactions_process::TransactionsProcess;
 use crate::block_status::BlockStatus;
 use crate::types::{ActiveChain, BlockNumberAndHash, SyncShared};
-use crate::utils::send_message_to;
+use crate::utils::{metric_ckb_message_bytes, send_message_to, MetricDirection};
 use crate::{Status, StatusCode};
 use ckb_chain::chain::ChainController;
 use ckb_constant::sync::BAD_MESSAGE_BAN_TIME;
 use ckb_logger::{debug_target, error_target, info_target, trace_target, warn_target};
-use ckb_metrics::metrics;
 use ckb_network::{
     async_trait, bytes::Bytes, tokio, CKBProtocolContext, CKBProtocolHandler, PeerIndex,
     SupportProtocols, TargetSession,
@@ -164,14 +163,12 @@ impl Relayer {
         let item_bytes = message.as_slice().len() as u64;
         let status = self.try_process(Arc::clone(&nc), peer, message);
 
-        metrics!(
-            counter,
-            "ckb.messages_bytes",
+        metric_ckb_message_bytes(
+            MetricDirection::In,
+            &SupportProtocols::RelayV2.name(),
+            message.item_name(),
+            Some(status.code()),
             item_bytes,
-            "direction" => "in",
-            "protocol_id" => SupportProtocols::RelayV2.protocol_id().value().to_string(),
-            "item_id" => message.item_id().to_string(),
-            "status" => (status.code() as u16).to_string(),
         );
 
         if let Some(ban_time) = status.should_ban() {
@@ -525,7 +522,9 @@ impl Relayer {
                             )),
                     );
                 } else {
-                    metrics!(counter, "ckb.relay.transaction_short_id_collide", 1);
+                    if let Some(metrics) = ckb_metrics::handle() {
+                        metrics.ckb_relay_transaction_short_id_collide.inc();
+                    }
                     return ReconstructionResult::Collided;
                 }
             }

sync/src/status.rs

@@ -122,6 +122,11 @@ impl StatusCode {
     pub fn with_context<S: ToString>(self, context: S) -> Status {
         Status::new(self, Some(context))
     }
+
+    /// StatusCode's name like `Ok(100)`
+    pub fn name(self) -> String {
+        format!("{:?}({})", self, self as u16)
+    }
 }
 
 /// Process message status

sync/src/synchronizer/mod.rs

@@ -14,7 +14,7 @@ pub(crate) use self::in_ibd_process::InIBDProcess;
 
 use crate::block_status::BlockStatus;
 use crate::types::{HeaderView, HeadersSyncController, IBDState, Peers, SyncShared};
-use crate::utils::send_message_to;
+use crate::utils::{metric_ckb_message_bytes, send_message_to, MetricDirection};
 use crate::{Status, StatusCode};
 
 use ckb_chain::chain::ChainController;
@@ -25,7 +25,6 @@ use ckb_constant::sync::{
 };
 use ckb_error::Error as CKBError;
 use ckb_logger::{debug, error, info, trace, warn};
-use ckb_metrics::metrics;
 use ckb_network::{
     async_trait, bytes::Bytes, tokio, CKBProtocolContext, CKBProtocolHandler, PeerIndex,
     ServiceControl, SupportProtocols,
@@ -265,14 +264,12 @@ impl Synchronizer {
         let item_bytes = message.as_slice().len() as u64;
         let status = self.try_process(nc, peer, message);
 
-        metrics!(
-            counter,
-            "ckb.messages_bytes",
+        metric_ckb_message_bytes(
+            MetricDirection::In,
+            &SupportProtocols::Sync.name(),
+            item_name,
+            Some(status.code()),
             item_bytes,
-            "direction" => "in",
-            "protocol_id" => SupportProtocols::Sync.protocol_id().value().to_string(),
-            "item_id" => message.item_id().to_string(),
-            "status" => (status.code() as u16).to_string(),
         );
 
         if let Some(ban_time) = status.should_ban() {

sync/src/types/mod.rs

@@ -15,7 +15,6 @@ use ckb_constant::sync::{
 };
 use ckb_error::Error as CKBError;
 use ckb_logger::{debug, error, trace};
-use ckb_metrics::metrics;
 use ckb_network::{CKBProtocolContext, PeerIndex, SupportProtocols};
 use ckb_shared::{shared::Shared, Snapshot};
 use ckb_store::{ChainDB, ChainStore};
@@ -1642,7 +1641,9 @@ impl SyncState {
             return;
         }
 
-        metrics!(gauge, "ckb.shared_best_number", header.number() as i64);
+        if let Some(metrics) = ckb_metrics::handle() {
+            metrics.ckb_shared_best_number.set(header.number() as i64);
+        }
         *self.shared_best_header.write() = header;
     }